Shovel bugfix: ensure initial metrics are reported for all protocols

AMQP1.0 doesn't emit metrics right after the initial connection,
it needs something to happen in the shovel to emit the first
blocked status report. By emitting metrics on the first running report,
all shovels will report the initial metrics as soon as they are initialised.

Author: dcorbacho

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].