RMQ-1263: Shovel: add forwarded counter

Delayed queuese can automatically create associated Shovels to transfer Ready messages
to the desired destination. This adds forwarded messages counter which will be used
in Management UI for better Shovel internals visibility.

(cherry picked from commit a8800b6cd75d8dc42a91f88655058f2ffa3b6ea6)

deps/rabbitmq_shovel/src/Elixir.RabbitMQ.CLI.Ctl.Commands.DeleteShovelCommand.erl

@@ -77,7 +77,7 @@ run([Name], #{node := Node, vhost := VHost}) ->
                     try_force_removing(Node, VHost, Name, ActingUser),
                     {error, rabbit_data_coercion:to_binary(ErrMsg)};
                 Match ->
-                    {{_Name, _VHost}, _Type, {_State, Opts}, _Timestamp} = Match,
+                    {{_Name, _VHost}, _Type, {_State, Opts}, _Metrics, _Timestamp} = Match,
                     {_, HostingNode} = lists:keyfind(node, 1, Opts),
                     case rabbit_misc:rpc_call(
                         HostingNode, rabbit_shovel_util, delete_shovel, [VHost, Name, ActingUser]) of

deps/rabbitmq_shovel/src/rabbit_amqp091_shovel.erl

@@ -365,15 +365,17 @@ publish(IncomingTag, Method, Msg,
             ok = amqp_channel:call(OutboundChan, Method, Msg)
     end,
 
+    #{dest := Dst1} = State1 = rabbit_shovel_behaviour:incr_forwarded(State),
+
     rabbit_shovel_behaviour:decr_remaining_unacked(
       case AckMode of
           no_ack ->
-              rabbit_shovel_behaviour:decr_remaining(1, State);
+              rabbit_shovel_behaviour:decr_remaining(1, State1);
           on_confirm ->
-              State#{dest => Dst#{unacked => Unacked#{Seq => IncomingTag}}};
+              State1#{dest => Dst1#{unacked => Unacked#{Seq => IncomingTag}}};
           on_publish ->
-              State1 = rabbit_shovel_behaviour:ack(IncomingTag, false, State),
-              rabbit_shovel_behaviour:decr_remaining(1, State1)
+              State2 = rabbit_shovel_behaviour:ack(IncomingTag, false, State1),
+              rabbit_shovel_behaviour:decr_remaining(1, State2)
       end).
 
 control_throttle(State) ->

deps/rabbitmq_shovel/src/rabbit_shovel_behaviour.erl

@@ -30,7 +30,8 @@
          status/1,
          % common functions
          decr_remaining_unacked/1,
-         decr_remaining/2
+         decr_remaining/2,
+         incr_forwarded/1
         ]).
 
 -type tag() :: non_neg_integer().
@@ -155,7 +156,18 @@ nack(Tag, Multi, #{source := #{module := Mod}} = State) ->
     Mod:nack(Tag, Multi, State).
 
 status(#{dest := #{module := Mod}} = State) ->
-    Mod:status(State).
+    {Mod:status(State), metrics(State)}.
+
+incr_forwarded(State = #{dest := Dest}) ->
+    State#{dest => maps:put(forwarded, maps:get(forwarded, Dest, 0) + 1, Dest)}.
+
+metrics(_State = #{source := Source,
+                   dest := Dest}) ->
+    #{remaining => maps:get(remaining, Source, unlimited),
+     remaining_unacked => maps:get(remaining_unacked, Source, 0),
+      pending => maps:get(pending, Dest, 0),
+      forwarded => maps:get(forwarded, Dest, 0)}.
+
 
 %% Common functions
 

deps/rabbitmq_shovel/src/rabbit_shovel_status.erl

@@ -49,6 +49,12 @@
                 info :: info(),
                 blocked_status = running :: blocked_status(),
                 blocked_at :: integer() | undefined,
+                metrics :: #{remaining := rabbit_types:option(non_neg_integer()) | unlimited,
+                             ramaining_unacked := rabbit_types:option(non_neg_integer()),
+                             pending := rabbit_types:option(non_neg_integer()),
+                             forwarded := rabbit_types:option(non_neg_integer())
+                             },
+
                 timestamp :: calendar:datetime()}).
 
 start_link() ->
@@ -112,6 +118,7 @@ handle_call(status, _From, State) ->
     {reply, [{Entry#entry.name,
               Entry#entry.type,
               blocked_status_to_info(Entry),
+              Entry#entry.metrics,
               Entry#entry.timestamp}
              || Entry <- Entries], State};
 
@@ -120,6 +127,7 @@ handle_call({lookup, Name}, _From, State) ->
                [Entry] -> [{name, Name},
                            {type, Entry#entry.type},
                            {info, blocked_status_to_info(Entry)},
+                           {metrics, Entry#entry.metrics},
                            {timestamp, Entry#entry.timestamp}];
                [] -> not_found
            end,
@@ -141,6 +149,18 @@ handle_cast({report, Name, Type, Info, Timestamp}, State) ->
                         split_name(Name) ++ split_status(Info)),
     {noreply, State};
 
+handle_cast({report_blocked_status, Name, {Status, Metrics}, Timestamp}, State) ->
+    case Status of
+        flow ->
+            true = ets:update_element(?ETS_NAME, Name, [{#entry.blocked_status, flow},
+                                                        {#entry.metrics, Metrics},
+                                                        {#entry.blocked_at, Timestamp}]);
+        _ ->
+            true = ets:update_element(?ETS_NAME, Name, [{#entry.blocked_status, Status},
+                                                        {#entry.metrics, Metrics}])
+    end,
+    {noreply, State};
+%% used in tests
 handle_cast({report_blocked_status, Name, Status, Timestamp}, State) ->
     case Status of
         flow ->
@@ -178,22 +198,22 @@ code_change(_OldVsn, State, _Extra) ->
 inject_node_info(Node, Shovels) ->
     lists:map(
         %% starting
-        fun({Name, Type, State, Timestamp}) when is_atom(State) ->
+        fun({Name, Type, State, Metrics, Timestamp}) when is_atom(State) ->
              Opts = [{node, Node}],
-             {Name, Type, {State, Opts}, Timestamp};
+             {Name, Type, {State, Opts}, Metrics, Timestamp};
            %% terminated
-           ({Name, Type, {terminated, Reason}, Timestamp}) ->
-             {Name, Type, {terminated, Reason}, Timestamp};
+           ({Name, Type, {terminated, Reason}, Metrics, Timestamp}) ->
+             {Name, Type, {terminated, Reason}, Metrics, Timestamp};
             %% running
-           ({Name, Type, {State, Opts}, Timestamp}) ->
+           ({Name, Type, {State, Opts}, Metrics, Timestamp}) ->
              Opts1 = Opts ++ [{node, Node}],
-             {Name, Type, {State, Opts1}, Timestamp}
+             {Name, Type, {State, Opts1}, Metrics, Timestamp}
         end, Shovels).
 
 -spec find_matching_shovel(rabbit_types:vhost(), binary(), [status_tuple()]) -> status_tuple() | undefined.
 find_matching_shovel(VHost, Name, Shovels) ->
     case lists:filter(
-        fun ({{V, S}, _Kind, _Status, _}) ->
+        fun ({{V, S}, _Kind, _Status, _Metrics, _}) ->
             VHost =:= V andalso Name =:= S
         end, Shovels) of
             []  -> undefined;

deps/rabbitmq_shovel/test/amqp10_SUITE.erl

@@ -139,7 +139,7 @@ amqp10_destination(Config, AckMode) ->
               throw(timeout_waiting_for_deliver1)
     end,
 
-    [{test_shovel, static, {running, _Info}, _Time}] =
+    [{test_shovel, static, {running, _Info}, _Metrics, _Time}] =
         rabbit_ct_broker_helpers:rpc(Config, 0,
           rabbit_shovel_status, status, []),
     amqp10_client:detach_link(Receiver),
@@ -183,7 +183,7 @@ amqp10_source(Config, AckMode) ->
     after ?TIMEOUT -> throw(timeout_waiting_for_deliver1)
     end,
 
-    [{test_shovel, static, {running, _Info}, _Time}] =
+    [{test_shovel, static, {running, _Info}, _Metrics, _Time}] =
         rabbit_ct_broker_helpers:rpc(Config, 0,
           rabbit_shovel_status, status, []),
     rabbit_ct_client_helpers:close_channel(Chan).
@@ -267,7 +267,7 @@ setup_shovel(ShovelConfig) ->
     await_running_shovel(test_shovel).
 
 await_running_shovel(Name) ->
-    case [N || {N, _, {running, _}, _}
+    case [N || {N, _, {running, _}, _, _}
                       <- rabbit_shovel_status:status(),
                          N =:= Name] of
         [_] -> ok;

deps/rabbitmq_shovel/test/configuration_SUITE.erl

@@ -277,7 +277,7 @@ run_valid_test(Config) ->
     after ?TIMEOUT -> throw(timeout_waiting_for_deliver1)
     end,
 
-    [{test_shovel, static, {running, _Info}, _Time}] =
+    [{test_shovel, static, {running, _Info}, _Metrics, _Time}] =
         rabbit_ct_broker_helpers:rpc(Config, 0,
           rabbit_shovel_status, status, []),
 
@@ -407,7 +407,7 @@ setup_shovels2(Config) ->
     ok = application:start(rabbitmq_shovel).
 
 await_running_shovel(Name) ->
-    case [N || {N, _, {running, _}, _}
+    case [N || {N, _, {running, _}, _Metrics,  _}
                       <- rabbit_shovel_status:status(),
                          N =:= Name] of
         [_] -> ok;
@@ -415,7 +415,7 @@ await_running_shovel(Name) ->
                await_running_shovel(Name)
     end.
 await_terminated_shovel(Name) ->
-    case [N || {N, _, {terminated, _}, _}
+    case [N || {N, _, {terminated, _}, _Metrics, _}
                       <- rabbit_shovel_status:status(),
                          N =:= Name] of
         [_] -> ok;

deps/rabbitmq_shovel/test/dynamic_SUITE.erl

@@ -118,13 +118,17 @@ end_per_testcase(Testcase, Config) ->
 %% -------------------------------------------------------------------
 
 simple(Config) ->
+    Name = <<"test">>,
     with_ch(Config,
       fun (Ch) ->
               shovel_test_utils:set_param(
                 Config,
-                <<"test">>, [{<<"src-queue">>,  <<"src">>},
+                Name, [{<<"src-queue">>,  <<"src">>},
                              {<<"dest-queue">>, <<"dest">>}]),
-              publish_expect(Ch, <<>>, <<"src">>, <<"dest">>, <<"hello">>)
+              publish_expect(Ch, <<>>, <<"src">>, <<"dest">>, <<"hello">>),
+              Status = rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_shovel_status, lookup, [{<<"/">>, Name}]),
+              ?assertMatch([_|_], Status),
+              ?assertMatch(#{metrics := #{forwarded := 1}}, maps:from_list(Status))
       end).
 
 quorum_queues(Config) ->

deps/rabbitmq_shovel/test/shovel_status_command_SUITE.erl

@@ -82,11 +82,11 @@ run_starting(Config) ->
     [A] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
     Opts = #{node => A},
     case ?CMD:run([], Opts) of
-        {stream, [{{<<"/">>, <<"test">>}, dynamic, starting, _}]} ->
+        {stream, [{{<<"/">>, <<"test">>}, dynamic, starting, _, _}]} ->
             ok;
         {stream, []} ->
             throw(shovel_not_found);
-        {stream, [{{<<"/">>, <<"test">>}, dynamic, {running, _}, _}]} ->
+        {stream, [{{<<"/">>, <<"test">>}, dynamic, {running, _}, _, _}]} ->
             ct:pal("Shovel is already running, starting could not be tested!")
     end,
     shovel_test_utils:clear_param(Config, <<"test">>).
@@ -107,7 +107,7 @@ run_running(Config) ->
                    {<<"dest-queue">>, <<"dest">>}]),
     [A] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
     Opts = #{node => A},
-    {stream, [{{<<"/">>, <<"test">>}, dynamic, {running, _}, _}]}
+    {stream, [{{<<"/">>, <<"test">>}, dynamic, {running, _}, _,  _}]}
         = ?CMD:run([], Opts),
     shovel_test_utils:clear_param(Config, <<"test">>).
 

deps/rabbitmq_shovel/test/shovel_test_utils.erl

@@ -65,7 +65,7 @@ shovels_from_status() ->
 
 shovels_from_status(ExpectedState) ->
     S = rabbit_shovel_status:status(),
-    [N || {{<<"/">>, N}, dynamic, {State, _}, _} <- S, State == ExpectedState].
+    [N || {{<<"/">>, N}, dynamic, {State, _}, _, _} <- S, State == ExpectedState].
 
 get_shovel_status(Config, Name) ->
     get_shovel_status(Config, 0, Name).
@@ -111,4 +111,4 @@ restart_shovel(Config, Name) ->
 
 restart_shovel(Config, Node, Name) ->
     rabbit_ct_broker_helpers:rpc(Config,
-                        Node, rabbit_shovel_util, restart_shovel, [<<"/">>, Name]).
+                        Node, rabbit_shovel_util, restart_shovel, [<<"/">>, Name]).

deps/rabbitmq_shovel_management/src/rabbit_shovel_mgmt_util.erl

@@ -42,7 +42,7 @@ status(Node) ->
             [format(Node, I) || I <- Status]
     end.
 
-format(Node, {Name, Type, Info, TS}) ->
+format(Node, {Name, Type, Info, Metrics, TS}) ->
     [{node, Node}, {timestamp, format_ts(TS)}] ++
         format_name(Type, Name) ++
         format_info(Info).