HTTP API: allow connections to be listed and closed by username

deps/rabbitmq_management/priv/www/api/index.html

@@ -331,6 +331,18 @@ <h2>Reference</h2>
           DELETEing to provide a reason.
         </td>
       </tr>
+      <tr>
+        <td>X</td>
+        <td></td>
+        <td>X</td>
+        <td></td>
+        <td class="path">/api/connections/username/<i>username</i></td>
+        <td>
+          A list of all open connections for a specific username. Use pagination parameters to filter connections.
+          DELETEing a resource will close all the connections for a username. Optionally set the
+          "X-Reason" header when DELETEing to provide a reason.
+        </td>
+      </tr>
       <tr>
         <td>X</td>
         <td></td>

deps/rabbitmq_management/src/rabbit_mgmt_dispatcher.erl

@@ -114,6 +114,7 @@ dispatcher() ->
      {"/vhost-limits/:vhost",                                  rabbit_mgmt_wm_limits, []},
      {"/connections",                                          rabbit_mgmt_wm_connections, []},
      {"/connections/:connection",                              rabbit_mgmt_wm_connection, []},
+     {"/connections/username/:username",                       rabbit_mgmt_wm_connection_user_name, []},
      {"/connections/:connection/channels",                     rabbit_mgmt_wm_connection_channels, []},
      {"/channels",                                             rabbit_mgmt_wm_channels, []},
      {"/channels/:channel",                                    rabbit_mgmt_wm_channel, []},

deps/rabbitmq_management/src/rabbit_mgmt_wm_connection.erl

@@ -37,11 +37,13 @@ resource_exists(ReqData, Context) ->
 to_json(ReqData, Context) ->
     case rabbit_mgmt_util:disable_stats(ReqData) of
         false ->
-            rabbit_mgmt_util:reply(
-              maps:from_list(rabbit_mgmt_format:strip_pids(conn_stats(ReqData))), ReqData, Context);
+            ConnStats = conn_stats(ReqData),
+            ConnStatsWithoutPids = rabbit_mgmt_format:strip_pids(ConnStats),
+            ReplyData = maps:from_list(ConnStatsWithoutPids),
+            rabbit_mgmt_util:reply(ReplyData, ReqData, Context);
         true ->
-            rabbit_mgmt_util:reply([{name, rabbit_mgmt_util:id(connection, ReqData)}],
-                                   ReqData, Context)
+            ReplyData = [{name, rabbit_mgmt_util:id(connection, ReqData)}],
+            rabbit_mgmt_util:reply(ReplyData, ReqData, Context)
     end.
 
 delete_resource(ReqData, Context) ->

deps/rabbitmq_management/src/rabbit_mgmt_wm_connection_user_name.erl

@@ -0,0 +1,91 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2022 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbit_mgmt_wm_connection_user_name).
+
+-export([init/2, to_json/2, content_types_provided/2,
+         is_authorized/2, allowed_methods/2, delete_resource/2]).
+-export([variances/2]).
+
+-include_lib("rabbitmq_management_agent/include/rabbit_mgmt_records.hrl").
+-include_lib("rabbit_common/include/rabbit.hrl").
+
+%%--------------------------------------------------------------------
+
+init(Req, _State) ->
+    {cowboy_rest, rabbit_mgmt_headers:set_common_permission_headers(Req, ?MODULE), #context{}}.
+
+variances(Req, Context) ->
+    {[<<"accept-encoding">>, <<"origin">>], Req, Context}.
+
+content_types_provided(ReqData, Context) ->
+    {rabbit_mgmt_util:responder_map(to_json), ReqData, Context}.
+
+allowed_methods(ReqData, Context) ->
+    {[<<"HEAD">>, <<"GET">>, <<"DELETE">>, <<"OPTIONS">>], ReqData, Context}.
+
+to_json(ReqData, Context) ->
+    {ok, _Username, UserConns} = list_user_connections(ReqData),
+    FilteredConns = rabbit_mgmt_util:filter_tracked_conn_list(UserConns, ReqData, Context),
+    rabbit_mgmt_util:reply_list_or_paginate(FilteredConns, ReqData, Context).
+
+delete_resource(ReqData, Context) ->
+    delete_resource(list_user_connections(ReqData), ReqData, Context).
+
+delete_resource({ok, _Username, []}, ReqData, Context) ->
+    {true, ReqData, Context};
+delete_resource({ok, Username, UserConns}, ReqData, Context) ->
+    ok = close_user_connections(UserConns, Username, ReqData),
+    {true, ReqData, Context}.
+
+is_authorized(ReqData, Context) ->
+    try
+        UserConns = list_user_connections(ReqData),
+        rabbit_mgmt_util:is_authorized_user(ReqData, Context, UserConns)
+    catch
+        {error, invalid_range_parameters, Reason} ->
+            rabbit_mgmt_util:bad_request(iolist_to_binary(Reason), ReqData, Context)
+    end.
+
+%%--------------------------------------------------------------------
+
+list_user_connections(ReqData) ->
+    Username = rabbit_mgmt_util:id(username, ReqData),
+    UserConns = rabbit_connection_tracking:list_of_user(Username),
+    {ok, Username, UserConns}.
+
+close_user_connections([], _Username, _ReqData) ->
+    ok;
+close_user_connections([Conn | Rest], Username, ReqData) ->
+    ok = close_user_connection(Conn, Username, ReqData),
+    close_user_connections(Rest, Username, ReqData).
+
+close_user_connection(#tracked_connection{name = Name, pid = Pid, username = Username, type = Type}, Username, ReqData) when is_pid(Pid) ->
+    Conn = [{name, Name}, {pid, Pid}, {user, Username}, {type, Type}],
+    force_close_connection(ReqData, Conn, Pid);
+close_user_connection(#tracked_connection{pid = undefined}, _Username, _ReqData) ->
+    ok;
+close_user_connection(UnexpectedConn, Username, _ReqData) ->
+    rabbit_log:debug("~p Username: ~p", [?MODULE, Username]),
+    rabbit_log:debug("~p unexpected connection: ~p", [?MODULE, UnexpectedConn]),
+    ok.
+
+force_close_connection(ReqData, Conn, Pid) ->
+    Reason = case cowboy_req:header(<<"x-reason">>, ReqData) of
+                 undefined -> "Closed via management plugin";
+                 V -> binary_to_list(V)
+             end,
+    case proplists:get_value(type, Conn) of
+        direct ->
+            amqp_direct_connection:server_close(Pid, 320, Reason);
+        network ->
+            rabbit_networking:close_connection(Pid, Reason);
+        _ ->
+            % best effort, this will work for connections to the stream plugin
+            gen_server:cast(Pid, {shutdown, Reason})
+    end,
+    ok.

deps/rabbitmq_management/src/rabbit_mgmt_wm_connections.erl

@@ -14,7 +14,6 @@
 -import(rabbit_misc, [pget/2]).
 
 -include_lib("rabbitmq_management_agent/include/rabbit_mgmt_records.hrl").
--include_lib("rabbit_common/include/rabbit.hrl").
 
 %%--------------------------------------------------------------------