Refactor quorum queue periodic metric emission

Primarily to avoid the transient process that is spawned every
Ra "tick" to query the quorum queue process for information that could
be passed in when the process is spawned.

Refactored to make use of the overview map instead of a custom
tuple.

Also use ets:lookup_element where applicable

Author: kjnilsson

deps/rabbit/src/rabbit_fifo.erl

@@ -888,26 +888,14 @@ state_enter0(_, _, Effects) ->
     Effects.
 
 -spec tick(non_neg_integer(), state()) -> ra_machine:effects().
-tick(Ts, #?MODULE{cfg = #cfg{name = Name,
-                             resource = QName},
-                  msg_bytes_enqueue = EnqueueBytes,
-                  msg_bytes_checkout = CheckoutBytes,
-                  dlx = DlxState} = State) ->
+tick(Ts, #?MODULE{cfg = #cfg{name = _Name,
+                             resource = QName}} = State) ->
     case is_expired(Ts, State) of
         true ->
             [{mod_call, rabbit_quorum_queue, spawn_deleter, [QName]}];
         false ->
-            {_, MsgBytesDiscard} = rabbit_fifo_dlx:stat(DlxState),
-            Metrics = {Name,
-                       messages_ready(State),
-                       num_checked_out(State), % checked out
-                       messages_total(State),
-                       query_consumer_count(State), % Consumers
-                       EnqueueBytes,
-                       CheckoutBytes,
-                       MsgBytesDiscard},
             [{mod_call, rabbit_quorum_queue,
-              handle_tick, [QName, Metrics, all_nodes(State)]}]
+              handle_tick, [QName, overview(State), all_nodes(State)]}]
     end.
 
 -spec overview(state()) -> map().
@@ -918,7 +906,8 @@ overview(#?MODULE{consumers = Cons,
                   msg_bytes_enqueue = EnqueueBytes,
                   msg_bytes_checkout = CheckoutBytes,
                   cfg = Cfg,
-                  dlx = DlxState} = State) ->
+                  dlx = DlxState,
+                  waiting_consumers = WaitingConsumers} = State) ->
     Conf = #{name => Cfg#cfg.name,
              resource => Cfg#cfg.resource,
              release_cursor_interval => Cfg#cfg.release_cursor_interval,
@@ -930,9 +919,18 @@ overview(#?MODULE{consumers = Cons,
              msg_ttl => Cfg#cfg.msg_ttl,
              delivery_limit => Cfg#cfg.delivery_limit
             },
+    SacOverview = case active_consumer(Cons) of
+                      {SacConsumerId, _} ->
+                          NumWaiting = length(WaitingConsumers),
+                          #{single_active_consumer_id => SacConsumerId,
+                            single_active_num_waiting_consumers => NumWaiting};
+                      _ ->
+                          #{}
+                  end,
     Overview = #{type => ?MODULE,
                  config => Conf,
-                 num_consumers => maps:size(Cons),
+                 num_consumers => map_size(Cons),
+                 num_active_consumers => query_consumer_count(State),
                  num_checked_out => num_checked_out(State),
                  num_enqueuers => maps:size(Enqs),
                  num_ready_messages => messages_ready(State),
@@ -944,9 +942,10 @@ overview(#?MODULE{consumers = Cons,
                  enqueue_message_bytes => EnqueueBytes,
                  checkout_message_bytes => CheckoutBytes,
                  in_memory_message_bytes => 0, %% backwards compat
-                 smallest_raft_index => smallest_raft_index(State)},
+                 smallest_raft_index => smallest_raft_index(State)
+                 },
     DlxOverview = rabbit_fifo_dlx:overview(DlxState),
-    maps:merge(Overview, DlxOverview).
+    maps:merge(maps:merge(Overview, DlxOverview), SacOverview).
 
 -spec get_checked_out(consumer_id(), msg_id(), msg_id(), state()) ->
     [delivery_msg()].

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -435,30 +435,56 @@ spawn_notify_decorators(QName, Fun, Args) ->
     catch notify_decorators(QName, Fun, Args).
 
 handle_tick(QName,
-            {Name, MR, MU, M, C, MsgBytesReady, MsgBytesUnack, MsgBytesDiscard},
+            #{config := #{name := Name},
+              num_active_consumers := NumConsumers,
+              num_checked_out := NumCheckedOut,
+              num_ready_messages := NumReadyMsgs,
+              num_messages := NumMessages,
+              enqueue_message_bytes := EnqueueBytes,
+              checkout_message_bytes := CheckoutBytes,
+              num_discarded := NumDiscarded,
+              num_discard_checked_out  :=  NumDiscardedCheckedOut,
+              discard_message_bytes := DiscardBytes,
+              discard_checkout_message_bytes := DiscardCheckoutBytes,
+              smallest_raft_index := _} = Overview,
             Nodes) ->
     %% this makes calls to remote processes so cannot be run inside the
     %% ra server
     Self = self(),
     _ = spawn(
           fun() ->
                   try
-                      R = reductions(Name),
-                      rabbit_core_metrics:queue_stats(QName, MR, MU, M, R),
-                      Util = case C of
+                      Reductions = reductions(Name),
+                      rabbit_core_metrics:queue_stats(QName, NumReadyMsgs,
+                                                      NumCheckedOut, NumMessages,
+                                                      Reductions),
+                      Util = case NumConsumers of
                                  0 -> 0;
                                  _ -> rabbit_fifo:usage(Name)
                              end,
-                      Infos = [{consumers, C},
+                      Keys = ?STATISTICS_KEYS -- [consumers,
+                                                  messages_dlx,
+                                                  message_bytes_dlx,
+                                                  single_active_consumer_pid,
+                                                  single_active_consumer_ctag
+                                                 ],
+                      {SacTag, SacPid} = maps:get(single_active_consumer_id,
+                                                  Overview, {'', ''}),
+                      MsgBytesDiscarded = DiscardBytes + DiscardCheckoutBytes,
+                      MsgBytes = EnqueueBytes + CheckoutBytes + MsgBytesDiscarded,
+                      Infos = [{consumers, NumConsumers},
                                {consumer_capacity, Util},
                                {consumer_utilisation, Util},
-                               {message_bytes_ready, MsgBytesReady},
-                               {message_bytes_unacknowledged, MsgBytesUnack},
-                               {message_bytes, MsgBytesReady + MsgBytesUnack + MsgBytesDiscard},
-                               {message_bytes_persistent, MsgBytesReady + MsgBytesUnack + MsgBytesDiscard},
-                               {messages_persistent, M}
-
-                               | infos(QName, ?STATISTICS_KEYS -- [consumers])],
+                               {message_bytes_ready, EnqueueBytes},
+                               {message_bytes_unacknowledged, CheckoutBytes},
+                               {message_bytes, MsgBytes},
+                               {message_bytes_persistent, MsgBytes},
+                               {messages_persistent, NumMessages},
+                               {messages_dlx, NumDiscarded + NumDiscardedCheckedOut},
+                               {message_bytes_dlx, MsgBytesDiscarded},
+                               {single_active_consumer_ctag, SacTag},
+                               {single_active_consumer_pid, SacPid}
+                               | infos(QName, Keys)],
                       rabbit_core_metrics:queue_stats(QName, Infos),
                       ok = repair_leader_record(QName, Self),
                       ExpectedNodes = rabbit_nodes:all(),
@@ -884,8 +910,6 @@ deliver(QSs, #delivery{message = #basic_message{content = Content0} = Msg,
 state_info(S) ->
     #{pending_raft_commands => rabbit_fifo_client:pending_size(S)}.
 
-
-
 -spec infos(rabbit_types:r('queue')) -> rabbit_types:infos().
 infos(QName) ->
     infos(QName, ?STATISTICS_KEYS).
@@ -976,9 +1000,11 @@ cluster_state(Name) ->
     case whereis(Name) of
         undefined -> down;
         _ ->
-            case ets:lookup(ra_state, Name) of
-                [{_, recover}] -> recovering;
-                _ -> running
+            case ets_lookup_element(ra_state, Name, 2, undefined) of
+                recover ->
+                    recovering;
+                _ ->
+                    running
             end
     end.
 
@@ -1232,7 +1258,8 @@ queue_length(Q) ->
     Name = amqqueue:get_name(Q),
     case ets:lookup(ra_metrics, Name) of
         [] -> 0;
-        [{_, _, SnapIdx, _, _, LastIdx, _}] -> LastIdx - SnapIdx
+        [{_, _, SnapIdx, _, _, LastIdx, _}] ->
+            LastIdx - SnapIdx
     end.
 
 get_replicas(Q) ->
@@ -1374,20 +1401,10 @@ i(messages, Q) when ?is_amqqueue(Q) ->
     quorum_messages(QName);
 i(messages_ready, Q) when ?is_amqqueue(Q) ->
     QName = amqqueue:get_name(Q),
-    case ets:lookup(queue_coarse_metrics, QName) of
-        [{_, MR, _, _, _}] ->
-            MR;
-        [] ->
-            0
-    end;
+    ets_lookup_element(queue_coarse_metrics, QName, 2, 0);
 i(messages_unacknowledged, Q) when ?is_amqqueue(Q) ->
     QName = amqqueue:get_name(Q),
-    case ets:lookup(queue_coarse_metrics, QName) of
-        [{_, _, MU, _, _}] ->
-            MU;
-        [] ->
-            0
-    end;
+    ets_lookup_element(queue_coarse_metrics, QName, 3, 0);
 i(policy, Q) ->
     case rabbit_policy:name(Q) of
         none   -> '';
@@ -1405,12 +1422,8 @@ i(effective_policy_definition, Q) ->
     end;
 i(consumers, Q) when ?is_amqqueue(Q) ->
     QName = amqqueue:get_name(Q),
-    case ets:lookup(queue_metrics, QName) of
-        [{_, M, _}] ->
-            proplists:get_value(consumers, M, 0);
-        [] ->
-            0
-    end;
+    Consumers = ets_lookup_element(queue_metrics, QName, 2, []),
+    proplists:get_value(consumers, Consumers, 0);
 i(memory, Q) when ?is_amqqueue(Q) ->
     {Name, _} = amqqueue:get_pid(Q),
     try
@@ -1429,10 +1442,7 @@ i(state, Q) when ?is_amqqueue(Q) ->
     end;
 i(local_state, Q) when ?is_amqqueue(Q) ->
     {Name, _} = amqqueue:get_pid(Q),
-    case ets:lookup(ra_state, Name) of
-        [{_, State}] -> State;
-        _ -> not_member
-    end;
+    ets_lookup_element(ra_state, Name, 2, not_member);
 i(garbage_collection, Q) when ?is_amqqueue(Q) ->
     {Name, _} = amqqueue:get_pid(Q),
     try
@@ -1477,27 +1487,9 @@ i(single_active_consumer_ctag, Q) when ?is_amqqueue(Q) ->
     end;
 i(type, _) -> quorum;
 i(messages_ram, Q) when ?is_amqqueue(Q) ->
-    QPid = amqqueue:get_pid(Q),
-    case ra:local_query(QPid,
-                        fun rabbit_fifo:query_in_memory_usage/1) of
-        {ok, {_, {Length, _}}, _} ->
-            Length;
-        {error, _} ->
-            0;
-        {timeout, _} ->
-            0
-    end;
+    0;
 i(message_bytes_ram, Q) when ?is_amqqueue(Q) ->
-    QPid = amqqueue:get_pid(Q),
-    case ra:local_query(QPid,
-                        fun rabbit_fifo:query_in_memory_usage/1) of
-        {ok, {_, {_, Bytes}}, _} ->
-            Bytes;
-        {error, _} ->
-            0;
-        {timeout, _} ->
-            0
-    end;
+    0;
 i(messages_dlx, Q) when ?is_amqqueue(Q) ->
     QPid = amqqueue:get_pid(Q),
     case ra:local_query(QPid,
@@ -1524,11 +1516,10 @@ i(_K, _Q) -> ''.
 
 open_files(Name) ->
     case whereis(Name) of
-        undefined -> {node(), 0};
-        Pid -> case ets:lookup(ra_open_file_metrics, Pid) of
-                   [] -> {node(), 0};
-                   [{_, Count}] -> {node(), Count}
-               end
+        undefined ->
+            {node(), 0};
+        Pid ->
+            {node(), ets_lookup_element(ra_open_file_metrics, Pid, 2, 0)}
     end.
 
 leader(Q) when ?is_amqqueue(Q) ->
@@ -1582,12 +1573,7 @@ is_process_alive(Name, Node) ->
 -spec quorum_messages(rabbit_amqqueue:name()) -> non_neg_integer().
 
 quorum_messages(QName) ->
-    case ets:lookup(queue_coarse_metrics, QName) of
-        [{_, _, _, M, _}] ->
-            M;
-        [] ->
-            0
-    end.
+    ets_lookup_element(queue_coarse_metrics, QName, 4, 0).
 
 quorum_ctag(Int) when is_integer(Int) ->
     integer_to_binary(Int);
@@ -1712,3 +1698,11 @@ erpc_timeout(Node, _)
     infinity;
 erpc_timeout(_, Timeout) ->
     Timeout.
+
+ets_lookup_element(Tbl, Key, Pos, Default) ->
+    try ets:lookup_element(Tbl, Key, Pos) of
+        V -> V
+    catch
+        _:badarg ->
+            Default
+    end.

deps/rabbit/test/rabbit_fifo_SUITE.erl

@@ -610,7 +610,14 @@ tick_test(C) ->
 
     [{mod_call, rabbit_quorum_queue, handle_tick,
       [#resource{},
-       {?FUNCTION_NAME, 1, 1, 2, 1, 3, 3, 0},
+            #{config := #{name := ?FUNCTION_NAME},
+              num_consumers := 1,
+              num_checked_out := 1,
+              num_ready_messages := 1,
+              num_messages := 2,
+              enqueue_message_bytes := 3,
+              checkout_message_bytes := 3,
+              num_discarded := _Discards},
        [_Node]
       ]}] = rabbit_fifo:tick(1, S4),
     ok.
@@ -773,6 +780,7 @@ single_active_consumer_basic_get_test(C) ->
     {_State, {error, {unsupported, single_active_consumer}}} =
         apply(meta(C, 2), rabbit_fifo:make_checkout(Cid, {dequeue, unsettled}, #{}),
               State1),
+
     ok.
 
 single_active_consumer_revive_test(C) ->
@@ -861,6 +869,10 @@ single_active_consumer_test(C) ->
     % the first registered consumer is the active one, the others are waiting
     ?assertEqual(1, map_size(State1#rabbit_fifo.consumers)),
     ?assertMatch(#{C1 := _}, State1#rabbit_fifo.consumers),
+
+    ?assertMatch(#{single_active_consumer_id := C1,
+                   single_active_num_waiting_consumers := 3},
+                 rabbit_fifo:overview(State1)),
     ?assertEqual(3, length(rabbit_fifo:query_waiting_consumers(State1))),
     ?assertNotEqual(false, lists:keyfind(C2, 1, rabbit_fifo:query_waiting_consumers(State1))),
     ?assertNotEqual(false, lists:keyfind(C3, 1, rabbit_fifo:query_waiting_consumers(State1))),
@@ -874,6 +886,9 @@ single_active_consumer_test(C) ->
     ?assertEqual(1, map_size(State2#rabbit_fifo.consumers)),
     ?assertMatch(#{C1 := _}, State2#rabbit_fifo.consumers),
     % the cancelled consumer has been removed from waiting consumers
+    ?assertMatch(#{single_active_consumer_id := C1,
+                   single_active_num_waiting_consumers := 2},
+                 rabbit_fifo:overview(State2)),
     ?assertEqual(2, length(rabbit_fifo:query_waiting_consumers(State2))),
     ?assertNotEqual(false, lists:keyfind(C2, 1, rabbit_fifo:query_waiting_consumers(State2))),
     ?assertNotEqual(false, lists:keyfind(C4, 1, rabbit_fifo:query_waiting_consumers(State2))),