Merge pull request #14637 from rabbitmq/issue-14623

michaelklishin · web-flow · commit 270c43fcacdb · 2025-09-30T12:25:32.000-04:00
Shovels: fix shovel status and deletion of failed shovels
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
@@ -68,7 +68,6 @@ run([Name], #{node := Node, vhost := VHost, force := Force}) ->
                      true -> ?INTERNAL_USER;
                      false -> 'Elixir.RabbitMQ.CLI.Core.Helpers':cli_acting_user()
                  end,
-
     case rabbit_misc:rpc_call(Node, rabbit_shovel_status, cluster_status_with_nodes, []) of
         {badrpc, _} = Error ->
             Error;
@@ -82,14 +81,15 @@ run([Name], #{node := Node, vhost := VHost, force := Force}) ->
                     try_force_removing(Node, VHost, Name, ActingUser),
                     {error, rabbit_data_coercion:to_binary(ErrMsg)};
                 {{_Name, _VHost}, _Type, {_State, Opts}, _Metrics, _Timestamp} ->
-                    delete_shovel(ErrMsg, VHost, Name, ActingUser, Opts, Node);
+                    HostingNode = proplists:get_value(node, Opts, Node),
+                    delete_shovel(ErrMsg, VHost, Name, ActingUser, HostingNode, Node);
                 {{_Name, _VHost}, _Type, {_State, Opts}, _Timestamp} ->
-                    delete_shovel(ErrMsg, VHost, Name, ActingUser, Opts, Node)
+                    HostingNode = proplists:get_value(node, Opts, Node),
+                    delete_shovel(ErrMsg, VHost, Name, ActingUser, HostingNode, Node)
             end
     end.
 
-delete_shovel(ErrMsg, VHost, Name, ActingUser, Opts, Node) ->
-    {_, HostingNode} = lists:keyfind(node, 1, Opts),
+delete_shovel(ErrMsg, VHost, Name, ActingUser, HostingNode, Node) ->
     case rabbit_misc:rpc_call(
         HostingNode, rabbit_shovel_util, delete_shovel, [VHost, Name, ActingUser]) of
         {badrpc, _} = Error ->
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
@@ -63,14 +63,15 @@ run([Name], #{node := Node, vhost := VHost}) ->
                 undefined ->
                     {error, rabbit_data_coercion:to_binary(ErrMsg)};
                 {{_Name, _VHost}, _Type, {_State, Opts}, _Metrics, _Timestamp} ->
-                    restart_shovel(ErrMsg, Name, VHost, Opts);
+                    HostingNode = proplists:get_value(node, Opts, Node),
+                    restart_shovel(ErrMsg, Name, VHost, HostingNode);
                 {{_Name, _VHost}, _Type, {_State, Opts}, _Timestamp} ->
-                    restart_shovel(ErrMsg, Name, VHost, Opts)
+                    HostingNode = proplists:get_value(node, Opts, Node),
+                    restart_shovel(ErrMsg, Name, VHost, HostingNode)
             end
     end.
 
-restart_shovel(ErrMsg, Name, VHost, Opts) ->
-    {_, HostingNode} = lists:keyfind(node, 1, Opts),
+restart_shovel(ErrMsg, Name, VHost, HostingNode) ->
     case rabbit_misc:rpc_call(
         HostingNode, rabbit_shovel_util, restart_shovel, [VHost, Name]) of
         {badrpc, _} = Error ->
diff --git a/deps/rabbitmq_shovel/src/rabbit_shovel_parameters.erl b/deps/rabbitmq_shovel/src/rabbit_shovel_parameters.erl
@@ -89,7 +89,10 @@ notify_clear(VHost, <<"shovel">>, Name, _Username) ->
     OpMode = rabbit_shovel_operating_mode:operating_mode(),
     case OpMode of
         standard ->
-            rabbit_shovel_dyn_worker_sup_sup:stop_child({VHost, Name});
+            rabbit_shovel_dyn_worker_sup_sup:stop_child({VHost, Name}),
+            %% Only necessary for shovels stuck in a restart loop, as no
+            %% process is running the terminate won't be called
+            rabbit_shovel_status:remove({VHost, Name});
         _Other ->
             ?LOG_DEBUG("Shovel: ignoring a cleared runtime parameter, operating mode: ~ts", [OpMode])
     end.
diff --git a/deps/rabbitmq_shovel/src/rabbit_shovel_status.erl b/deps/rabbitmq_shovel/src/rabbit_shovel_status.erl
@@ -42,6 +42,9 @@
 -type info() :: starting
               | {running, proplists:proplist()}
               | {terminated, term()}.
+-type info_with_node() :: starting
+                        | {running, proplists:proplist()}
+                        | {terminated, proplists:proplist()}.
 -type blocked_status() :: running | flow | blocked.
 -type shovel_status() :: blocked_status() | ignore.
 
@@ -52,7 +55,7 @@
                      pending := rabbit_types:option(non_neg_integer()),
                      forwarded := rabbit_types:option(non_neg_integer())
                     } | #{}.
--type status_tuple_41x() :: {name(), type(), info(), metrics(), calendar:datetime()}.
+-type status_tuple_41x() :: {name(), type(), info() | info_with_node(), metrics(), calendar:datetime()}.
 -type status_tuple_40x_and_older() :: {name(), type(), info(), calendar:datetime()}.
 -type status_tuple() :: status_tuple_41x() | status_tuple_40x_and_older().
 
@@ -216,7 +219,8 @@ inject_node_info(Node, Shovels) ->
              {Name, Type, {State, Opts}, Metrics, Timestamp};
            %% terminated
            ({Name, Type, {terminated, Reason}, Metrics, Timestamp}) ->
-             {Name, Type, {terminated, Reason}, Metrics, Timestamp};
+             {Name, Type, {terminated, [{node, Node},
+                                        {reason, Reason}]}, Metrics, Timestamp};
             %% running
            ({Name, Type, {State, Opts}, Metrics, Timestamp}) ->
              Opts1 = Opts ++ [{node, Node}],
diff --git a/deps/rabbitmq_shovel/src/rabbit_shovel_worker.erl b/deps/rabbitmq_shovel/src/rabbit_shovel_worker.erl
@@ -82,7 +82,7 @@ handle_cast(init, State = #state{config = Config0}) ->
     catch E:R ->
       ?LOG_ERROR("Shovel ~ts could not connect to source: ~p ~p",
                  [human_readable_name(maps:get(name, Config0)), E, R]),
-      {stop, shutdown, State}
+      {stop, {shutdown, inbound_conn_failed}, State}
     end;
 handle_cast(connect_dest, State = #state{config = Config0}) ->
     try rabbit_shovel_behaviour:connect_dest(Config0) of
@@ -93,7 +93,7 @@ handle_cast(connect_dest, State = #state{config = Config0}) ->
     catch E:R ->
       ?LOG_ERROR("Shovel ~ts could not connect to destination: ~p ~p",
                  [human_readable_name(maps:get(name, Config0)), E, R]),
-      {stop, shutdown, State}
+      {stop, {shutdown, outbond_conn_failed}, State}
     end;
 handle_cast(init_shovel, State = #state{config = Config}) ->
     %% Don't trap exits until we have established connections so that
@@ -227,6 +227,20 @@ terminate({{shutdown, {server_initiated_close, Code, Reason}}, _}, State = #stat
                                 {terminated, "needed a restart"}),
     close_connections(State),
     ok;
+terminate({shutdown, outbond_conn_failed}, State = #state{name = Name}) ->
+    ?LOG_ERROR("Shovel ~ts is stopping because if failed to connect to destination",
+               [human_readable_name(Name)]),
+    rabbit_shovel_status:report(State#state.name, State#state.type,
+                                {terminated, "failed to connect to destination"}),
+    close_connections(State),
+    ok;
+terminate({shutdown, inbound_conn_failed}, State = #state{name = Name}) ->
+    ?LOG_ERROR("Shovel ~ts is stopping because it failed to connect to source",
+               [human_readable_name(Name)]),
+    rabbit_shovel_status:report(State#state.name, State#state.type,
+                                {terminated, "failed to connect to source"}),
+    close_connections(State),
+    ok;
 terminate(Reason, State = #state{name = Name}) ->
     ?LOG_ERROR("Shovel ~ts is stopping, reason: ~tp", [human_readable_name(Name), Reason]),
     rabbit_shovel_status:report(State#state.name, State#state.type,
diff --git a/deps/rabbitmq_shovel/test/delete_shovel_command_SUITE.erl b/deps/rabbitmq_shovel/test/delete_shovel_command_SUITE.erl
@@ -9,6 +9,7 @@
 
 -include_lib("amqp_client/include/amqp_client.hrl").
 -include_lib("stdlib/include/assert.hrl").
+-include_lib("rabbitmq_ct_helpers/include/rabbit_assert.hrl").
 
 -compile(export_all).
 
@@ -26,11 +27,14 @@ groups() ->
                                delete_not_found,
                                delete,
                                delete_internal,
-                               delete_internal_owner
+                               delete_internal_owner,
+                               delete_invalid_uri,
+                               delete_non_existent_dest_address
                               ]},
      {cluster_size_2, [], [
-                               clear_param_on_different_node
-                              ]}
+                           clear_param_on_different_node,
+                           delete_invalid_uri_another_node
+                          ]}
     ].
 
 %% -------------------------------------------------------------------
@@ -64,6 +68,15 @@ end_per_group(_Group, Config) ->
       rabbit_ct_client_helpers:teardown_steps() ++
       rabbit_ct_broker_helpers:teardown_steps()).
 
+init_per_testcase(Testcase, Config) when Testcase == delete_invalid_uri_another_node ->
+    case rabbit_ct_helpers:is_mixed_versions(Config) of
+        true ->
+            %% The code changes to delete shovel are compatible with older versions, however
+            %% older versions fail to delete invalid shovels
+            {skip, "not mixed versions compatible"};
+        false ->
+            rabbit_ct_helpers:testcase_started(Config, Testcase)
+    end;
 init_per_testcase(Testcase, Config) ->
     rabbit_ct_helpers:testcase_started(Config, Testcase).
 
@@ -148,3 +161,67 @@ clear_param_on_different_node(Config) ->
                                                   status, []), "Deleted shovel still reported on node A"),
     ?assertEqual([], rabbit_ct_broker_helpers:rpc(Config, B, rabbit_shovel_status,
                                                   status, []), "Deleted shovel still reported on node B").
+
+delete_invalid_uri(Config) ->
+    ok = rabbit_ct_broker_helpers:rpc(
+           Config, 0, rabbit_runtime_parameters, set,
+           [<<"/">>, <<"shovel">>, <<"myshovel">>,
+            [{<<"src-protocol">>, <<"amqp091">>},
+             {<<"src-uri">>, <<"amqp://foo">>},
+             {<<"src-queue">>,  <<"src">>},
+             {<<"dest-protocol">>, <<"amqp091">>},
+             {<<"dest-uri">>, shovel_test_utils:make_uri(Config, 0)},
+             {<<"dest-queue">>, <<"dest">>}],
+            none]),
+    ?awaitMatch([{{<<"/">>, <<"myshovel">>}, dynamic, {terminated, _}, _, _}],
+                rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_shovel_status,
+                                             status, []),
+                45_000),
+    [A] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    Opts = #{node => A, vhost => <<"/">>, force => false},
+    ok = ?CMD:run([<<"myshovel">>], Opts),
+    [] = rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_shovel_status,
+                                      status, []).
+
+delete_non_existent_dest_address(Config) ->
+    Uri = shovel_test_utils:make_uri(Config, 0),
+    ok = rabbit_ct_broker_helpers:rpc(
+           Config, 0, rabbit_runtime_parameters, set,
+           [<<"/">>, <<"shovel">>, <<"myshovel">>,
+            [{<<"src-protocol">>, <<"amqp091">>},
+             {<<"src-uri">>, Uri},
+             {<<"src-queue">>,  <<"src">>},
+             {<<"dest-protocol">>, <<"amqp10">>},
+             {<<"dest-uri">>, Uri},
+             {<<"dest-address">>, <<"/queues/q2">>}],
+            none]),
+    ?awaitMatch([{{<<"/">>, <<"myshovel">>}, dynamic, {terminated, _}, _, _}],
+                rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_shovel_status,
+                                             status, []),
+                45_000),
+    [A] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    Opts = #{node => A, vhost => <<"/">>, force => false},
+    ok = ?CMD:run([<<"myshovel">>], Opts),
+    [] = rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_shovel_status,
+                                      status, []).
+
+delete_invalid_uri_another_node(Config) ->
+    ok = rabbit_ct_broker_helpers:rpc(
+           Config, 1, rabbit_runtime_parameters, set,
+           [<<"/">>, <<"shovel">>, <<"myshovel">>,
+            [{<<"src-protocol">>, <<"amqp091">>},
+             {<<"src-uri">>, <<"amqp://foo">>},
+             {<<"src-queue">>,  <<"src">>},
+             {<<"dest-protocol">>, <<"amqp091">>},
+             {<<"dest-uri">>, shovel_test_utils:make_uri(Config, 0)},
+             {<<"dest-queue">>, <<"dest">>}],
+            none]),
+    ?awaitMatch([{{<<"/">>, <<"myshovel">>}, dynamic, {terminated, _}, _, _}],
+                rabbit_ct_broker_helpers:rpc(Config, 1, rabbit_shovel_status,
+                                             status, []),
+                45_000),
+    [A, _B] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    Opts = #{node => A, vhost => <<"/">>, force => false},
+    ok = ?CMD:run([<<"myshovel">>], Opts),
+    [] = rabbit_ct_broker_helpers:rpc(Config, 1, rabbit_shovel_status,
+                                      status, []).
