Shovel: add forwarded counter

Author: deadtrickster

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,13 +149,15 @@ handle_cast({report, Name, Type, Info, Timestamp}, State) ->
                         split_name(Name) ++ split_status(Info)),
     {noreply, State};
 
-handle_cast({report_blocked_status, Name, Status, Timestamp}, 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}])
+            true = ets:update_element(?ETS_NAME, Name, [{#entry.blocked_status, Status},
+                                                        {#entry.metrics, Metrics}])
     end,
     {noreply, State};