Merge pull request #4242 from rabbitmq/mk-rabbitmq-server-4127-followup

michaelklishin · web-flow · commit 29b912475fa3 · 2022-03-10T04:11:48.000+06:00
Make it possible to delete or restart a dynamic Shovel on/from any cluster node (take two)
diff --git a/deps/rabbitmq_shovel/src/Elixir.RabbitMQ.CLI.Ctl.Commands.DeleteShovelCommand.erl b/deps/rabbitmq_shovel/src/Elixir.RabbitMQ.CLI.Ctl.Commands.DeleteShovelCommand.erl
@@ -2,7 +2,7 @@
 %% 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-2021 VMware, Inc. or its affiliates.  All rights reserved.
+%% Copyright (c) 2007-2022 VMware, Inc. or its affiliates.  All rights reserved.
 %%
 
 -module('Elixir.RabbitMQ.CLI.Ctl.Commands.DeleteShovelCommand').
@@ -68,15 +68,32 @@ banner([Name], #{vhost := VHost}) ->
 
 run([Name], #{node := Node, vhost := VHost}) ->
     ActingUser = 'Elixir.RabbitMQ.CLI.Core.Helpers':cli_acting_user(),
-    case rabbit_misc:rpc_call(Node, rabbit_shovel_util, delete_shovel, [VHost, Name, ActingUser]) of
+
+    case rabbit_misc:rpc_call(Node, rabbit_shovel_status, cluster_status_with_nodes, []) of
         {badrpc, _} = Error ->
             Error;
-        {error, not_found} ->
+        Xs when is_list(Xs) ->
             ErrMsg = rabbit_misc:format("Shovel with the given name was not found "
                                         "on the target node '~s' and / or virtual host '~s'",
                                         [Node, VHost]),
-            {error, rabbit_data_coercion:to_binary(ErrMsg)};
-        ok -> ok
+            case rabbit_shovel_status:find_matching_shovel(VHost, Name, Xs) of
+                undefined ->
+                    {error, rabbit_data_coercion:to_binary(ErrMsg)};
+                Match ->
+                    {{_Name, _VHost}, _Type, {_State, Opts}, _Timestamp} = Match,
+                    {_, HostingNode} = lists:keyfind(node, 1, Opts),
+                    case rabbit_misc:rpc_call(
+                        HostingNode, rabbit_shovel_util, delete_shovel, [VHost, Name, ActingUser]) of
+                        {badrpc, _} = Error ->
+                            Error;
+                        {error, not_found} ->
+                            ErrMsg = rabbit_misc:format("Shovel with the given name was not found "
+                                                        "on the target node '~s' and / or virtual host '~s'",
+                                                        [Node, VHost]),
+                            {error, rabbit_data_coercion:to_binary(ErrMsg)};
+                        ok -> ok
+                    end
+            end
     end.
 
 switches() ->
diff --git a/deps/rabbitmq_shovel/src/Elixir.RabbitMQ.CLI.Ctl.Commands.RestartShovelCommand.erl b/deps/rabbitmq_shovel/src/Elixir.RabbitMQ.CLI.Ctl.Commands.RestartShovelCommand.erl
@@ -54,15 +54,28 @@ banner([Name], #{node := Node, vhost := VHost}) ->
                              << " on node ">>, atom_to_binary(Node, utf8)]).
 
 run([Name], #{node := Node, vhost := VHost}) ->
-    case rabbit_misc:rpc_call(Node, rabbit_shovel_util, restart_shovel, [VHost, Name]) of
+    case rabbit_misc:rpc_call(Node, rabbit_shovel_status, cluster_status_with_nodes, []) of
         {badrpc, _} = Error ->
             Error;
-        {error, not_found} ->
+        Xs when is_list(Xs) ->
             ErrMsg = rabbit_misc:format("Shovel with the given name was not found "
                                         "on the target node '~s' and / or virtual host '~s'",
                                         [Node, VHost]),
-            {error, rabbit_data_coercion:to_binary(ErrMsg)};
-        ok -> ok
+            case rabbit_shovel_status:find_matching_shovel(VHost, Name, Xs) of
+                undefined ->
+                    {error, rabbit_data_coercion:to_binary(ErrMsg)};
+                Match ->
+                    {{_Name, _VHost}, _Type, {_State, Opts}, _Timestamp} = Match,
+                    {_, HostingNode} = lists:keyfind(node, 1, Opts),
+                    case rabbit_misc:rpc_call(
+                        HostingNode, rabbit_shovel_util, restart_shovel, [VHost, Name]) of
+                        {badrpc, _} = Error ->
+                            Error;
+                        {error, not_found} ->
+                            {error, rabbit_data_coercion:to_binary(ErrMsg)};
+                        ok -> ok
+                    end
+            end
     end.
 
 output(Output, _Opts) ->
diff --git a/deps/rabbitmq_shovel/src/rabbit_shovel_status.erl b/deps/rabbitmq_shovel/src/rabbit_shovel_status.erl
@@ -2,15 +2,16 @@
 %% 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-2021 VMware, Inc. or its affiliates.  All rights reserved.
+%% Copyright (c) 2007-2022 VMware, Inc. or its affiliates.  All rights reserved.
 %%
 
 -module(rabbit_shovel_status).
 -behaviour(gen_server).
 
 -export([start_link/0]).
 
--export([report/3, remove/1, status/0, lookup/1]).
+-export([report/3, remove/1, status/0, lookup/1, cluster_status/0, cluster_status_with_nodes/0]).
+-export([inject_node_info/2, find_matching_shovel/3]).
 
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
          terminate/2, code_change/3]).
@@ -34,6 +35,23 @@ remove(Name) ->
 status() ->
     gen_server:call(?SERVER, status, infinity).
 
+cluster_status() ->
+    Nodes = rabbit_nodes:all_running(),
+    lists:usort(rabbit_misc:append_rpc_all_nodes(Nodes, ?MODULE, status, [])).
+
+cluster_status_with_nodes() ->
+    Nodes = rabbit_nodes:all_running(),
+    lists:foldl(
+        fun(Node, Acc) ->
+            case rabbit_misc:rpc_call(Node, ?MODULE, status, []) of
+                {badrpc, _} ->
+                    Acc;
+                Xs0 when is_list(Xs0) ->
+                    Xs = inject_node_info(Node, Xs0),
+                    Acc ++ Xs
+            end
+        end, [], Nodes).
+
 lookup(Name) ->
     gen_server:call(?SERVER, {lookup, Name}, infinity).
 
@@ -83,6 +101,26 @@ terminate(_Reason, State) ->
 code_change(_OldVsn, State, _Extra) ->
     {ok, State}.
 
+inject_node_info(Node, Shovels) ->
+    lists:map(
+        fun({Name, Type, {State, Opts}, Timestamp}) ->
+            Opts1 = Opts ++ [{node, Node}],
+            {Name, Type, {State, Opts1}, Timestamp}
+        end, Shovels).
+
+find_matching_shovel(VHost, Name, Shovels) ->
+    case lists:filter(
+        fun ({{V, S}, _Kind, _Status, _}) ->
+            VHost =:= V andalso Name =:= S
+        end, Shovels) of
+            []  -> undefined;
+        [S | _] -> S
+    end.
+
+%%
+%% Implementation
+%%
+
 split_status({running, MoreInfo})         -> [{status, running} | MoreInfo];
 split_status({terminated, Reason})        -> [{status, terminated},
                                               {reason, Reason}];
@@ -94,4 +132,4 @@ split_name(Name) when is_atom(Name) -> [{name, Name}].
 
 ensure_timer(State0) ->
     State1 = rabbit_misc:stop_timer(State0, #state.timer),
-    rabbit_misc:ensure_timer(State1, #state.timer, ?CHECK_FREQUENCY, check).
+    rabbit_misc:ensure_timer(State1, #state.timer, ?CHECK_FREQUENCY, check).
diff --git a/deps/rabbitmq_shovel_management/BUILD.bazel b/deps/rabbitmq_shovel_management/BUILD.bazel
@@ -8,6 +8,7 @@ load(
     "assert_suites",
     "rabbitmq_app",
     "rabbitmq_integration_suite",
+    "rabbitmq_suite",
 )
 
 APP_NAME = "rabbitmq_shovel_management"
@@ -71,6 +72,16 @@ suites = [
         PACKAGE,
         name = "http_SUITE",
     ),
+    rabbitmq_suite(
+        name = "rabbit_shovel_mgmt_SUITE",
+        runtime_deps = [
+            "@meck//:erlang_app",
+        ],
+        deps = [
+            "//deps/rabbit_common:erlang_app",
+            "//deps/rabbitmq_management_agent:erlang_app",
+        ],
+    ),
 ]
 
 assert_suites(
diff --git a/deps/rabbitmq_shovel_management/priv/www/js/shovel.js b/deps/rabbitmq_shovel_management/priv/www/js/shovel.js
@@ -87,15 +87,15 @@ dispatcher_add(function(sammy) {
             if (sync_delete(this, '/shovels/vhost/:vhost/:name')) {
                 go_to('#/dynamic-shovels');
             } else {
-                show_popup('warn', 'Shovel not deleted because it is not running on this node.');
+                show_popup('warn', 'Shovel could not be deleted');
                 return false;
             }
         });
     sammy.del("#/shovel-restart-link", function(){
             if (sync_delete(this, '/shovels/vhost/:vhost/:name/restart')) {
                 update();
             } else {
-                show_popup('warn', 'Shovel not restarted because it is not running on this node.');
+                show_popup('warn', 'Shovel could not be restarted');
                 return false;
             }
         });
diff --git a/deps/rabbitmq_shovel_management/src/rabbit_shovel_mgmt.erl b/deps/rabbitmq_shovel_management/src/rabbit_shovel_mgmt.erl
@@ -11,12 +11,13 @@
 
 -export([dispatcher/0, web_ui/0]).
 -export([init/2, to_json/2, resource_exists/2, content_types_provided/2,
-         is_authorized/2, allowed_methods/2, delete_resource/2]).
+         is_authorized/2, allowed_methods/2, delete_resource/2, get_shovel_node/4]).
 
 -import(rabbit_misc, [pget/2]).
 
 -include_lib("rabbitmq_management_agent/include/rabbit_mgmt_records.hrl").
 -include_lib("amqp_client/include/amqp_client.hrl").
+-include_lib("rabbit_shovel_mgmt.hrl").
 
 dispatcher() -> [{"/shovels",        ?MODULE, []},
                  {"/shovels/:vhost", ?MODULE, []},
@@ -45,11 +46,10 @@ resource_exists(ReqData, Context) ->
                         none -> true;
                         Name ->
                             %% Deleting or restarting a shovel
-                            case rabbit_shovel_status:lookup({VHost, Name}) of
-                                not_found ->
-                                    rabbit_log:error("Shovel with the name '~s' was not found "
-                                                     "on the target node '~s' and / or virtual host '~s'",
-                                                     [Name, node(), VHost]),
+                            case get_shovel_node(VHost, Name, ReqData, Context) of
+                                undefined ->
+                                    rabbit_log:error("Shovel with the name '~s' was not found on virtual host '~s'",
+                                        [Name, VHost]),
                                     false;
                                 _ ->
                                     true
@@ -60,7 +60,7 @@ resource_exists(ReqData, Context) ->
 
 to_json(ReqData, Context) ->
     rabbit_mgmt_util:reply_list(
-      filter_vhost_req(status(ReqData, Context), ReqData), ReqData, Context).
+      filter_vhost_req(rabbit_shovel_mgmt_util:status(ReqData, Context), ReqData), ReqData, Context).
 
 is_authorized(ReqData, Context) ->
     rabbit_mgmt_util:is_authorized_monitor(ReqData, Context).
@@ -71,21 +71,28 @@ delete_resource(ReqData, #context{user = #user{username = Username}}=Context) ->
                 none ->
                     false;
                 Name ->
-                    %% We must distinguish between a delete and restart
-                    case is_restart(ReqData) of
-                        true ->
-                            case rabbit_shovel_util:restart_shovel(VHost, Name) of
-                                {error, ErrMsg} ->
-                                    rabbit_log:error("Error restarting shovel: ~s", [ErrMsg]),
-                                    false;
-                                ok -> true
-                            end;
-                        _ ->
-                            case rabbit_shovel_util:delete_shovel(VHost, Name, Username) of
-                                {error, ErrMsg} ->
-                                    rabbit_log:error("Error deleting shovel: ~s", [ErrMsg]),
-                                    false;
-                                ok -> true
+                    case get_shovel_node(VHost, Name, ReqData, Context) of
+                        undefined -> rabbit_log:error("Could not find shovel data for shovel '~s' in vhost: '~s'", [Name, VHost]),
+                            false;
+                        Node ->
+                            %% We must distinguish between a delete and restart
+                            case is_restart(ReqData) of
+                                true ->
+                                    rabbit_log:info("Asked to restart shovel '~s' in vhost '~s' on node '~s'", [Name, VHost, Node]),
+                                    case rpc:call(Node, rabbit_shovel_util, restart_shovel, [VHost, Name], ?SHOVEL_CALLS_TIMEOUT_MS) of
+                                        ok -> true;
+                                        {_, Msg} -> rabbit_log:error(Msg),
+                                            false
+                                    end;
+
+                                _ ->
+                                    rabbit_log:info("Asked to delete shovel '~s' in vhost '~s' on node '~s'", [Name, VHost, Node]),
+                                    case rpc:call(Node, rabbit_shovel_util, delete_shovel, [VHost, Name, Username], ?SHOVEL_CALLS_TIMEOUT_MS) of
+                                        ok -> true;
+                                        {_, Msg} -> rabbit_log:error(Msg),
+                                            false
+                                    end
+
                             end
                     end
             end,
@@ -107,52 +114,29 @@ filter_vhost_req(List, ReqData) ->
                            pget(vhost, I) =:= VHost]
     end.
 
-%% Allow users to see things in the vhosts they are authorised. But
-%% static shovels do not have a vhost, so only allow admins (not
-%% monitors) to see them.
-filter_vhost_user(List, _ReqData, #context{user = User = #user{tags = Tags}}) ->
-    VHosts = rabbit_mgmt_util:list_login_vhosts_names(User, undefined),
-    [I || I <- List, case pget(vhost, I) of
-                         undefined -> lists:member(administrator, Tags);
-                         VHost     -> lists:member(VHost, VHosts)
-                     end].
-
-status(ReqData, Context) ->
-    filter_vhost_user(
-      lists:append([status(Node) || Node <- [node() | nodes()]]),
-      ReqData, Context).
-
-status(Node) ->
-    case rpc:call(Node, rabbit_shovel_status, status, [], infinity) of
-        {badrpc, {'EXIT', _}} ->
-            [];
-        Status ->
-            [format(Node, I) || I <- Status]
+get_shovel_node(VHost, Name, ReqData, Context) ->
+    Shovels = rabbit_shovel_mgmt_util:status(ReqData, Context),
+    Match   = find_matching_shovel(VHost, Name, Shovels),
+    case Match of
+        undefined -> undefined;
+        Match     ->
+            {_, Node} = lists:keyfind(node, 1, Match),
+            Node
     end.
 
-format(Node, {Name, Type, Info, TS}) ->
-    [{node, Node}, {timestamp, format_ts(TS)}] ++
-        format_name(Type, Name) ++
-        format_info(Info).
-
-format_name(static,  Name)          -> [{name,  Name},
-                                        {type,  static}];
-format_name(dynamic, {VHost, Name}) -> [{name,  Name},
-                                        {vhost, VHost},
-                                        {type,  dynamic}].
-
-format_info(starting) ->
-    [{state, starting}];
-
-format_info({running, Props}) ->
-    [{state, running}] ++ Props;
-
-format_info({terminated, Reason}) ->
-    [{state,  terminated},
-     {reason, print("~p", [Reason])}].
-
-format_ts({{Y, M, D}, {H, Min, S}}) ->
-    print("~w-~2.2.0w-~2.2.0w ~w:~2.2.0w:~2.2.0w", [Y, M, D, H, Min, S]).
-
-print(Fmt, Val) ->
-    list_to_binary(io_lib:format(Fmt, Val)).
+%% This is similar to rabbit_shovel_status:find_matching_shovel/3
+%% but operates on a different input (a proplist of Shovel attributes)
+-spec find_matching_shovel(VHost :: vhost:name(),
+                           Name :: binary(),
+                           Shovels :: list(list(tuple()))) -> 'undefined' | list(tuple()).
+find_matching_shovel(VHost, Name, Shovels) ->
+    ShovelPred = fun (Attributes) ->
+                         lists:member({name, Name}, Attributes) andalso
+                             lists:member({vhost, VHost}, Attributes)
+                 end,
+    case lists:search(ShovelPred, Shovels) of
+        {value, Shovel} ->
+            Shovel;
+        _ ->
+            undefined
+    end.
diff --git a/deps/rabbitmq_shovel_management/src/rabbit_shovel_mgmt.hrl b/deps/rabbitmq_shovel_management/src/rabbit_shovel_mgmt.hrl
@@ -0,0 +1,8 @@
+%% 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-2021 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-define(SHOVEL_CALLS_TIMEOUT_MS, 25000).
diff --git a/deps/rabbitmq_shovel_management/src/rabbit_shovel_mgmt_util.erl b/deps/rabbitmq_shovel_management/src/rabbit_shovel_mgmt_util.erl
diff --git a/deps/rabbitmq_shovel_management/test/rabbit_shovel_mgmt_SUITE.erl b/deps/rabbitmq_shovel_management/test/rabbit_shovel_mgmt_SUITE.erl
