Merge pull request #14872 from rabbitmq/mergify/bp/v4.2.x/pr-14794

Shovel tests and bugfixes (backport #14794)

Author: michaelklishin

deps/rabbitmq_shovel/src/rabbit_amqp091_shovel.erl

@@ -229,6 +229,10 @@ handle_source({#'basic.deliver'{delivery_tag = Tag,
     % forward to destination
     rabbit_shovel_behaviour:forward(Tag, Msg, State);
 
+handle_source(#'basic.cancel'{}, #{name := Name}) ->
+    ?LOG_WARNING("Shovel ~tp received a 'basic.cancel' from the server", [Name]),
+    {stop, {shutdown, restart}};
+
 handle_source({'EXIT', Conn, Reason},
               #{source := #{current := {Conn, _, _}}}) ->
     {stop, {inbound_conn_died, Reason}};
@@ -251,10 +255,6 @@ handle_dest(#'basic.nack'{delivery_tag = Seq, multiple = Multiple},
                                rabbit_shovel_behaviour:nack(Tag, Multi, StateX)
                        end, Seq, Multiple, State);
 
-handle_dest(#'basic.cancel'{}, #{name := Name}) ->
-    ?LOG_WARNING("Shovel ~tp received a 'basic.cancel' from the server", [Name]),
-    {stop, {shutdown, restart}};
-
 handle_dest({'EXIT', Conn, Reason}, #{dest := #{current := {Conn, _, _}}}) ->
     {stop, {outbound_conn_died, Reason}};
 

deps/rabbitmq_shovel/src/rabbit_local_shovel.erl

@@ -256,12 +256,10 @@ source_protocol(_State) ->
 dest_protocol(_State) ->
     local.
 
-source_endpoint(#{source := #{queue := Queue,
-                              exchange := SrcX,
+source_endpoint(#{source := #{exchange := SrcX,
                               routing_key := SrcXKey}}) ->
     [{src_exchange, SrcX},
-     {src_exchange_key, SrcXKey},
-     {src_queue, Queue}];
+     {src_exchange_key, SrcXKey}];
 source_endpoint(#{source := #{queue := Queue}}) ->
     [{src_queue, Queue}];
 source_endpoint(_Config) ->

deps/rabbitmq_shovel/src/rabbit_shovel_status.erl

@@ -14,7 +14,7 @@
 
 -export([start_link/0]).
 
--export([report/3,
+-export([report/3, report/4,
          report_blocked_status/2,
          remove/1,
          status/0,
@@ -78,6 +78,12 @@ start_link() ->
 report(Name, Type, Info) ->
     gen_server:cast(?SERVER, {report, Name, Type, Info, calendar:local_time()}).
 
+-spec report(name(), type(), info(), metrics()) -> ok.
+report(Name, Type, Info, Metrics) ->
+    %% Initialise metrics for protocols that don't immediately generate a
+    %% blocked status report. This happens with AMQP 1.0
+    gen_server:cast(?SERVER, {report, Name, Type, Info, Metrics, calendar:local_time()}).
+
 -spec report_blocked_status(name(), {blocked_status(), metrics()} | blocked_status()) -> ok.
 report_blocked_status(Name, Status) ->
     gen_server:cast(?SERVER, {report_blocked_status, Name, Status, erlang:monotonic_time()}).
@@ -164,6 +170,19 @@ handle_cast({report, Name, Type, Info, Timestamp}, State) ->
                         split_name(Name) ++ split_status(Info)),
     {noreply, State};
 
+handle_cast({report, Name, Type, Info, Metrics, Timestamp}, State) ->
+    Entry = #entry{
+        name = Name,
+        type = Type,
+        info = Info,
+        metrics = Metrics,
+        timestamp = Timestamp
+    },
+    true = ets:insert(?ETS_NAME, Entry),
+    rabbit_event:notify(shovel_worker_status,
+                        split_name(Name) ++ split_status(Info)),
+    {noreply, State};
+
 handle_cast({report_blocked_status, Name, {Status, Metrics}, Timestamp}, State) ->
     case Status of
         flow ->

deps/rabbitmq_shovel/src/rabbit_shovel_worker.erl

@@ -279,13 +279,15 @@ report_running(#state{config = Config} = State) ->
     OutProto = rabbit_shovel_behaviour:dest_protocol(Config),
     InEndpoint = rabbit_shovel_behaviour:source_endpoint(Config),
     OutEndpoint = rabbit_shovel_behaviour:dest_endpoint(Config),
+    {_, Metrics} = rabbit_shovel_behaviour:status(Config),
     rabbit_shovel_status:report(State#state.name, State#state.type,
                                 {running, [{src_uri,  rabbit_data_coercion:to_binary(InUri)},
                                            {src_protocol, rabbit_data_coercion:to_binary(InProto)},
                                            {dest_protocol, rabbit_data_coercion:to_binary(OutProto)},
                                            {dest_uri, rabbit_data_coercion:to_binary(OutUri)}]
                                  ++ props_to_binary(InEndpoint) ++ props_to_binary(OutEndpoint)
-                                }).
+                                },
+                               Metrics).
 
 props_to_binary(Props) ->
     [{K, rabbit_data_coercion:to_binary(V)} || {K, V} <- Props].