Report flow/blocked shovel status

Currently there is a known limitation that if the shovel is idle after
it was unblocked by credit flow, then the status won't get updated until
the next message.

Author: gomoripeti

deps/rabbitmq_shovel/src/rabbit_amqp091_shovel.erl

@@ -30,6 +30,7 @@
          close_dest/1,
          ack/3,
          nack/3,
+         status/1,
          forward/4
         ]).
 
@@ -155,16 +156,16 @@ dest_endpoint(#{dest := Dest}) ->
     maps:to_list(maps:filter(fun(K, _) -> proplists:is_defined(K, Keys) end, Dest)).
 
 forward(IncomingTag, Props, Payload, State) ->
-    State1 = control_throttle(State),
-    case is_blocked(State1) of
+    case is_blocked(State) of
         true ->
             %% We are blocked by client-side flow-control and/or
             %% `connection.blocked` message from the destination
             %% broker. Simply cache the forward.
             PendingEntry = {IncomingTag, Props, Payload},
-            add_pending(PendingEntry, State1);
+            add_pending(PendingEntry, State);
         false ->
-            do_forward(IncomingTag, Props, Payload, State1)
+            State1 = do_forward(IncomingTag, Props, Payload, State),
+            control_throttle(State1)
     end.
 
 do_forward(IncomingTag, Props, Payload,
@@ -368,6 +369,13 @@ is_blocked(#{dest := #{blocked_by := BlockReasons}}) when BlockReasons =/= [] ->
 is_blocked(_) ->
     false.
 
+status(#{dest := #{blocked_by := [flow]}}) ->
+    flow;
+status(#{dest := #{blocked_by := BlockReasons}}) when BlockReasons =/= [] ->
+    blocked;
+status(_) ->
+    credit_flow:state().
+
 add_pending(Elem, State = #{dest := Dest}) ->
     Pending = maps:get(pending, Dest, []),
     State#{dest => Dest#{pending => [Elem|Pending]}}.

deps/rabbitmq_shovel/src/rabbit_amqp10_shovel.erl

@@ -30,6 +30,7 @@
          close_dest/1,
          ack/3,
          nack/3,
+         status/1,
          forward/4
         ]).
 
@@ -302,6 +303,11 @@ nack(Tag, true, State = #{source := #{current := #{session := Session},
                                            Tag, true, accepted),
     State#{source => Src#{last_nacked_tag => Tag}}.
 
+status(#{dest := #{current := #{link_state := attached}}}) ->
+    flow;
+status(#{dest := #{current := #{link_state := credited}}}) ->
+    running.
+
 -spec forward(Tag :: tag(), Props :: #{atom() => any()},
               Payload :: binary(), state()) -> state().
 forward(_Tag, _Props, _Payload,

deps/rabbitmq_shovel/src/rabbit_shovel_behaviour.erl

@@ -27,6 +27,7 @@
          forward/4,
          ack/3,
          nack/3,
+         status/1,
          % common functions
          decr_remaining_unacked/1,
          decr_remaining/2
@@ -48,7 +49,9 @@
                    ack_mode => ack_mode(),
                    atom() => term()}.
 
--export_type([state/0, source_config/0, dest_config/0, uri/0, tag/0]).
+-type status() :: running | flow | blocked.
+
+-export_type([state/0, source_config/0, dest_config/0, uri/0, tag/0, status/0]).
 
 -callback parse(binary(), {source | destination, Conf :: proplists:proplist()}) ->
     source_config() | dest_config().
@@ -80,7 +83,7 @@
 -callback nack(Tag :: tag(), Multi :: boolean(), state()) -> state().
 -callback forward(Tag :: tag(), Props :: #{atom() => any()},
                   Payload :: binary(), state()) -> state().
-
+-callback status(state()) -> status().
 
 -spec parse(atom(), binary(), {source | destination, proplists:proplist()}) ->
     source_config() | dest_config().
@@ -151,6 +154,9 @@ ack(Tag, Multi, #{source := #{module := Mod}} = State) ->
 nack(Tag, Multi, #{source := #{module := Mod}} = State) ->
     Mod:nack(Tag, Multi, State).
 
+status(#{dest := #{module := Mod}} = State) ->
+    Mod:status(State).
+
 %% Common functions
 
 %% Count down until we stop publishing in on-confirm mode

deps/rabbitmq_shovel/src/rabbit_shovel_status.erl

@@ -121,15 +121,16 @@ find_matching_shovel(VHost, Name, Shovels) ->
 %% Implementation
 %%
 
-split_status({running, MoreInfo})         -> [{status, running} | MoreInfo];
+
 split_status({terminated, Reason})        -> [{status, terminated},
                                               {reason, Reason}];
-split_status(Status) when is_atom(Status) -> [{status, Status}].
+split_status(Status) when is_atom(Status) -> [{status, Status}];
+split_status({Status, [{_, _} | _] = MoreInfo}) -> [{status, Status} | MoreInfo].
 
 split_name({VHost, Name})           -> [{name,  Name},
                                         {vhost, VHost}];
 split_name(Name) when is_atom(Name) -> [{name, Name}].
 
 ensure_timer(State0) ->
     State1 = rabbit_misc:stop_timer(State0, #state.timer),
-    rabbit_misc:ensure_timer(State1, #state.timer, ?CHECK_FREQUENCY, check).
+    rabbit_misc:ensure_timer(State1, #state.timer, ?CHECK_FREQUENCY, check).

deps/rabbitmq_shovel/src/rabbit_shovel_worker.erl

@@ -20,7 +20,8 @@
 
 -record(state, {name :: binary() | {rabbit_types:vhost(), binary()},
                 type :: static | dynamic,
-                config :: rabbit_shovel_behaviour:state()}).
+                config :: rabbit_shovel_behaviour:state(),
+                last_reported_status :: rabbit_shovel_behaviour:status()}).
 
 start_link(Type, Name, Config) ->
     ShovelParameter = rabbit_shovel_util:get_shovel_parameter(Name),
@@ -92,8 +93,8 @@ handle_cast(init_shovel, State = #state{config = Config}) ->
     Config2 = rabbit_shovel_behaviour:init_source(Config1),
     rabbit_log_shovel:debug("Shovel ~ts has finished setting up its topology", [human_readable_name(maps:get(name, Config2))]),
     State1 = State#state{config = Config2},
-    ok = report_running(State1),
-    {noreply, State1}.
+    State2 = report_running(State1),
+    {noreply, State2}.
 
 
 handle_info(Msg, State = #state{config = Config, name = Name}) ->
@@ -116,7 +117,9 @@ handle_info(Msg, State = #state{config = Config, name = Name}) ->
                     rabbit_log_shovel:debug("Shovel ~ts decided to stop due a message from destination: ~tp", [human_readable_name(Name), Reason]),
                     {stop, Reason, State};
                 Config1 ->
-                    {noreply, State#state{config = Config1}}
+                    State1 = State#state{config = Config1},
+                    State2 = maybe_report_running(State1),
+                    {noreply, State2}
             end;
         {stop, {inbound_conn_died, heartbeat_timeout}} ->
             rabbit_log_shovel:error("Shovel ~ts detected missed heartbeats on source connection", [human_readable_name(Name)]),
@@ -131,7 +134,9 @@ handle_info(Msg, State = #state{config = Config, name = Name}) ->
             rabbit_log_shovel:error("Shovel ~ts decided to stop due a message from source: ~tp", [human_readable_name(Name), Reason]),
             {stop, Reason, State};
         Config1 ->
-            {noreply, State#state{config = Config1}}
+            State1 = State#state{config = Config1},
+            State2 = maybe_report_running(State1),
+            {noreply, State2}
     end.
 
 terminate({shutdown, autodelete}, State = #state{name = Name,
@@ -209,20 +214,33 @@ human_readable_name(Name) ->
     ShovelName          -> rabbit_misc:format("'~ts'", [ShovelName])
   end.
 
-report_running(#state{config = Config} = State) ->
+maybe_report_running(#state{config = Config,
+                            last_reported_status = LastStatus} = State) ->
+    case rabbit_shovel_behaviour:status(Config) of
+        LastStatus ->
+            State;
+        NewStatus ->
+            report_status(NewStatus, State)
+    end.
+
+report_running(State) ->
+    report_status(running, State).
+
+report_status(Status, #state{config = Config} = State) ->
     InUri = rabbit_shovel_behaviour:source_uri(Config),
     OutUri = rabbit_shovel_behaviour:dest_uri(Config),
     InProto = rabbit_shovel_behaviour:source_protocol(Config),
     OutProto = rabbit_shovel_behaviour:dest_protocol(Config),
     InEndpoint = rabbit_shovel_behaviour:source_endpoint(Config),
     OutEndpoint = rabbit_shovel_behaviour:dest_endpoint(Config),
     rabbit_shovel_status:report(State#state.name, State#state.type,
-                                {running, [{src_uri,  rabbit_data_coercion:to_binary(InUri)},
+                                {Status, [{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)
-                                }).
+                                }),
+    State#state{last_reported_status = Status}.
 
 props_to_binary(Props) ->
     [{K, rabbit_data_coercion:to_binary(V)} || {K, V} <- Props].

deps/rabbitmq_shovel/test/dynamic_SUITE.erl

@@ -459,6 +459,7 @@ credit_flow(Config) ->
                                  {<<"ack-mode">>,     <<"on-publish">>},
                                  {<<"src-delete-after">>, <<"never">>}]),
                   shovel_test_utils:await_shovel(Config, <<"test">>),
+                  running = shovel_test_utils:get_shovel_status(Config, <<"test">>),
 
                   ShovelPid = find_shovel_pid(Config),
                   #{dest :=
@@ -482,9 +483,9 @@ credit_flow(Config) ->
                   %% Wait until the shovel is blocked
                   shovel_test_utils:await(
                     fun() ->
-                            case get_shovel_state(ShovelPid) of
-                                #{dest := #{blocked_by := [flow]}} -> true;
-                                Conf -> Conf
+                            case shovel_test_utils:get_shovel_status(Config, <<"test">>) of
+                                flow -> true;
+                                Status -> Status
                             end
                     end,
                     5000),
@@ -522,8 +523,22 @@ credit_flow(Config) ->
                   #{messages := 1000} = message_count(Config, <<"dest">>),
                   [{_, 0, _}] =
                       rabbit_ct_broker_helpers:rpc(
-                        Config, 0, recon, proc_count, [message_queue_len, 1])
+                        Config, 0, recon, proc_count, [message_queue_len, 1]),
+
+                  %% To avoid frequent flipping of state, there is a 1
+                  %% second "state change interval" in credit flow
+                  %% before an unblocked process changes from flow to
+                  %% running status.
+                  timer:sleep(1000),
 
+                  %% Need to publish a message to trigger status
+                  %% reporting
+                  %% FIXME maybe add a timer in the shovel process to
+                  %% update status automatically
+                  publish(Ch, <<>>, <<"src">>, <<"hello">>),
+                  true = amqp_channel:wait_for_confirms(Ch),
+
+                  running = shovel_test_utils:get_shovel_status(Config, <<"test">>)
               after
                   resume_process(Config),
                   set_default_credit(Config, OrigCredit)
@@ -567,9 +582,7 @@ dest_resource_alarm(AckMode, Config) ->
                                  {<<"src-delete-after">>, <<"never">>}]),
 
                   %% The shovel is blocked
-                  ShovelPid = find_shovel_pid(Config),
-                  Conf = get_shovel_state(ShovelPid),
-                  #{dest := #{blocked_by := [connection_blocked]}} = Conf,
+                  blocked = shovel_test_utils:get_shovel_status(Config, <<"test">>),
 
                   %% The shoveled message triggered a
                   %% connection.blocked notification, but hasn't
@@ -646,7 +659,8 @@ dest_resource_alarm(AckMode, Config) ->
                             Cnt =:= 1001
                     end,
                     5000),
-                  #{messages := 0} = message_count(Config, <<"src">>)
+                  #{messages := 0} = message_count(Config, <<"src">>),
+                  running = shovel_test_utils:get_shovel_status(Config, <<"test">>)
               after
                   set_vm_memory_high_watermark(Config, OrigLimit)
               end

deps/rabbitmq_shovel/test/shovel_test_utils.erl

@@ -9,7 +9,8 @@
 
 -include_lib("common_test/include/ct.hrl").
 -export([set_param/3, set_param_nowait/3, await_shovel/2, await_shovel1/2,
-         shovels_from_status/0, await/1, await/2, clear_param/2]).
+         shovels_from_status/0, get_shovel_status/2,
+         await/1, await/2, clear_param/2]).
 
 make_uri(Config) ->
     Hostname = ?config(rmq_hostname, Config),
@@ -37,7 +38,21 @@ await_shovel1(_Config, Name) ->
 
 shovels_from_status() ->
     S = rabbit_shovel_status:status(),
-    [N || {{<<"/">>, N}, dynamic, {running, _}, _} <- S].
+    [N || {{<<"/">>, N}, dynamic, {Status, _}, _} <- S,
+          Status == running orelse
+              Status == flow orelse
+              Status == blocked].
+
+get_shovel_status(Config, Name) ->
+    S = rabbit_ct_broker_helpers:rpc(
+          Config, 0, rabbit_shovel_status, lookup, [{<<"/">>, Name}]),
+    case S of
+        not_found ->
+            not_found;
+        _ ->
+            {Status, _} = proplists:get_value(info, S),
+            Status
+    end.
 
 await(Pred) ->
     case Pred() of

deps/rabbitmq_shovel_management/priv/www/js/tmpl/shovels.ejs

@@ -46,7 +46,7 @@
     </td>
     <td><%= shovel.timestamp %></td>
 <% } else { %>
-    <td><%= fmt_state('green', shovel.state) %></td>
+    <td><%= fmt_object_state(shovel) %></td>
       <td><%= fmt_string(shovel.src_protocol) %></td>
       <td><%= shovel.src_uri == undefined ? fmt_string(shovel.src_uri) : fmt_string(fmt_uri_with_credentials(shovel.src_uri)) %></td>
       <td><%= fmt_shovel_endpoint('src_', shovel) %></td>

deps/rabbitmq_shovel_management/src/rabbit_shovel_mgmt_util.erl

@@ -52,8 +52,8 @@ format_name(dynamic, {VHost, Name}) -> [{name,  Name},
 format_info(starting) ->
     [{state, starting}];
 
-format_info({running, Props}) ->
-    [{state, running}] ++ Props;
+format_info({Status, [{_, _} | _] = Props}) ->
+    [{state, Status}] ++ Props;
 
 format_info({terminated, Reason}) ->
     [{state,  terminated},
@@ -63,4 +63,4 @@ format_ts({{Y, M, D}, {H, Min, S}}) ->
     print("~w-~2.2.0w-~2.2.0w ~w:~2.2.0w:~2.2.0w", [Y, M, D, H, Min, S]).
 
 print(Fmt, Val) ->
-    list_to_binary(io_lib:format(Fmt, Val)).
+    list_to_binary(io_lib:format(Fmt, Val)).