Expose ra counters (#13895)

Switch from ra_metrics to ra_counters

* Expose many more metrics (they are also up to date)
* Bump Seshat, Ra, Osiris, Prometheus.erl
* switch from proplists to maps

Author: mkuratczyk

deps/rabbit/src/rabbit_global_counters.erl

@@ -132,46 +132,46 @@
 boot_step() ->
     [begin
          %% Protocol counters
-         Protocol = {protocol, Proto},
-         init([Protocol]),
+         Protocol = #{protocol => Proto},
+         init(Protocol),
          rabbit_msg_size_metrics:init(Proto),
 
          %% Protocol & Queue Type counters
-         init([Protocol, {queue_type, rabbit_classic_queue}]),
-         init([Protocol, {queue_type, rabbit_quorum_queue}]),
-         init([Protocol, {queue_type, rabbit_stream_queue}])
+         init(Protocol#{queue_type => rabbit_classic_queue}),
+         init(Protocol#{queue_type => rabbit_quorum_queue}),
+         init(Protocol#{queue_type => rabbit_stream_queue})
      end || Proto <- [amqp091, amqp10]],
 
     %% Dead Letter counters
     %%
     %% Streams never dead letter.
     %%
     %% Source classic queue dead letters.
-    init([{queue_type, rabbit_classic_queue}, {dead_letter_strategy, disabled}],
+    init(#{queue_type => rabbit_classic_queue, dead_letter_strategy => disabled},
          [?MESSAGES_DEAD_LETTERED_MAXLEN_COUNTER,
           ?MESSAGES_DEAD_LETTERED_EXPIRED_COUNTER,
           ?MESSAGES_DEAD_LETTERED_REJECTED_COUNTER]),
-    init([{queue_type, rabbit_classic_queue}, {dead_letter_strategy, at_most_once}],
+    init(#{queue_type => rabbit_classic_queue, dead_letter_strategy => at_most_once},
          [?MESSAGES_DEAD_LETTERED_MAXLEN_COUNTER,
           ?MESSAGES_DEAD_LETTERED_EXPIRED_COUNTER,
           ?MESSAGES_DEAD_LETTERED_REJECTED_COUNTER]),
     %%
     %% Source quorum queue dead letters.
     %% Only quorum queues can dead letter due to delivery-limit exceeded.
     %% Only quorum queues support dead letter strategy at-least-once.
-    init([{queue_type, rabbit_quorum_queue}, {dead_letter_strategy, disabled}],
+    init(#{queue_type => rabbit_quorum_queue, dead_letter_strategy => disabled},
          [?MESSAGES_DEAD_LETTERED_MAXLEN_COUNTER,
           ?MESSAGES_DEAD_LETTERED_EXPIRED_COUNTER,
           ?MESSAGES_DEAD_LETTERED_REJECTED_COUNTER,
           ?MESSAGES_DEAD_LETTERED_DELIVERY_LIMIT_COUNTER
          ]),
-    init([{queue_type, rabbit_quorum_queue}, {dead_letter_strategy, at_most_once}],
+    init(#{queue_type => rabbit_quorum_queue, dead_letter_strategy => at_most_once},
          [?MESSAGES_DEAD_LETTERED_MAXLEN_COUNTER,
           ?MESSAGES_DEAD_LETTERED_EXPIRED_COUNTER,
           ?MESSAGES_DEAD_LETTERED_REJECTED_COUNTER,
           ?MESSAGES_DEAD_LETTERED_DELIVERY_LIMIT_COUNTER
          ]),
-    init([{queue_type, rabbit_quorum_queue}, {dead_letter_strategy, at_least_once}],
+    init(#{queue_type => rabbit_quorum_queue, dead_letter_strategy => at_least_once},
          [?MESSAGES_DEAD_LETTERED_CONFIRMED_COUNTER,
           ?MESSAGES_DEAD_LETTERED_EXPIRED_COUNTER,
           ?MESSAGES_DEAD_LETTERED_REJECTED_COUNTER,
@@ -181,21 +181,21 @@ boot_step() ->
 init(Labels) ->
     init(Labels, []).
 
-init(Labels = [{protocol, Protocol}, {queue_type, QueueType}], Extra) ->
+init(Labels = #{protocol := Protocol, queue_type := QueueType}, Extra) ->
     _ = seshat:new_group(?MODULE),
-    Counters = seshat:new(?MODULE, Labels, ?PROTOCOL_QUEUE_TYPE_COUNTERS ++ Extra),
+    Counters = seshat:new(?MODULE, Labels, ?PROTOCOL_QUEUE_TYPE_COUNTERS ++ Extra, Labels),
     persistent_term:put({?MODULE, Protocol, QueueType}, Counters);
-init(Labels = [{protocol, Protocol}], Extra) ->
+init(Labels = #{protocol := Protocol}, Extra) ->
     _ = seshat:new_group(?MODULE),
-    Counters = seshat:new(?MODULE, Labels, ?PROTOCOL_COUNTERS ++ Extra),
+    Counters = seshat:new(?MODULE, Labels, ?PROTOCOL_COUNTERS ++ Extra, Labels),
     persistent_term:put({?MODULE, Protocol}, Counters);
-init(Labels = [{queue_type, QueueType}, {dead_letter_strategy, DLS}], DeadLetterCounters) ->
+init(Labels = #{queue_type := QueueType, dead_letter_strategy := DLS}, DeadLetterCounters) ->
     _ = seshat:new_group(?MODULE),
-    Counters = seshat:new(?MODULE, Labels, DeadLetterCounters),
+    Counters = seshat:new(?MODULE, Labels, DeadLetterCounters, Labels),
     persistent_term:put({?MODULE, QueueType, DLS}, Counters).
 
 overview() ->
-    seshat:overview(?MODULE).
+    seshat:counters(?MODULE).
 
 prometheus_format() ->
     seshat:format(?MODULE).

deps/rabbit/src/rabbit_khepri.erl

@@ -276,6 +276,7 @@ setup(_Context) ->
                       {default_ra_system, ?RA_SYSTEM}]}],
            [{persistent, true}]),
     RaServerConfig = #{cluster_name => ?RA_CLUSTER_NAME,
+                       metrics_labels => #{ra_system => ?RA_SYSTEM, module => ?MODULE},
                        friendly_name => ?RA_FRIENDLY_NAME},
     case khepri:start(?RA_SYSTEM, RaServerConfig) of
         {ok, ?STORE_ID} ->

deps/rabbit/src/rabbit_observer_cli_quorum_queues.erl

@@ -123,11 +123,11 @@ sheet_header() ->
 sheet_body(PrevState) ->
     {_, RaStates} = rabbit_quorum_queue:all_replica_states(),
     Body = [begin
-                #resource{name = Name, virtual_host = Vhost} = R = amqqueue:get_name(Q),
+                #resource{name = Name, virtual_host = Vhost} = amqqueue:get_name(Q),
                 case rabbit_amqqueue:pid_of(Q) of
                     none ->
                         empty_row(Name);
-                    {QName, _QNode} = _QQ ->
+                    {QName, _QNode} = ServerId ->
                         case whereis(QName) of
                             undefined ->
                                 empty_row(Name);
@@ -139,7 +139,12 @@ sheet_body(PrevState) ->
                                     _ ->
                                         QQCounters = maps:get({QName, node()}, ra_counters:overview()),
                                         {ok, InternalName} = rabbit_queue_type_util:qname_to_internal_name(#resource{virtual_host = Vhost, name= Name}),
-                                        [{_, CT, SnapIdx, LA, CI, LW, CL}] = ets:lookup(ra_metrics, R),
+                                        #{snapshot_index := SnapIdx,
+                                            last_written_index := LW,
+                                            term := CT,
+                                            commit_latency := CL,
+                                            commit_index := CI,
+                                            last_applied := LA} = ra:key_metrics(ServerId),
                                         [
                                          Pid,
                                          QName,

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -1598,12 +1598,10 @@ transfer_leadership(Q, Destination) ->
     end.
 
 queue_length(Q) ->
-    Name = amqqueue:get_name(Q),
-    case ets:lookup(ra_metrics, Name) of
-        [] -> 0;
-        [{_, _, SnapIdx, _, _, LastIdx, _}] ->
-            LastIdx - SnapIdx
-    end.
+    ServerId = amqqueue:get_pid(Q),
+    #{snapshot_index := SnapIdx,
+      last_written_index := LastIdx} = key_metrics_rpc(ServerId),
+    LastIdx - SnapIdx.
 
 get_replicas(Q) ->
     get_nodes(Q).
@@ -1985,6 +1983,7 @@ make_ra_conf(Q, ServerId, TickTimeout,
              SnapshotInterval, CheckpointInterval,
              Membership, MacVersion) ->
     QName = amqqueue:get_name(Q),
+    #resource{name = QNameBin} = QName,
     RaMachine = ra_machine(Q),
     [{ClusterName, _} | _] = Members = members(Q),
     UId = ra:new_uid(ra_lib:to_binary(ClusterName)),
@@ -2000,6 +1999,8 @@ make_ra_conf(Q, ServerId, TickTimeout,
                                   uid => UId,
                                   friendly_name => FName,
                                   metrics_key => QName,
+                                  metrics_labels => #{vhost => amqqueue:get_vhost(Q),
+                                                      queue => QNameBin},
                                   initial_members => Members,
                                   log_init_args => LogCfg,
                                   tick_timeout => TickTimeout,

deps/rabbit/src/rabbit_stream_coordinator.erl

@@ -1349,6 +1349,7 @@ make_ra_conf(Node, Nodes, MinMacVersion) ->
       uid => UId,
       friendly_name => atom_to_list(?MODULE),
       metrics_key => ?MODULE,
+      metrics_labels => #{ra_system => ?RA_SYSTEM, module => ?MODULE},
       initial_members => Members,
       log_init_args => #{uid => UId},
       tick_timeout => TickTimeout,

deps/rabbit/test/amqp_client_SUITE.erl

@@ -6967,11 +6967,11 @@ formatted_state(Pid) ->
     proplists:get_value("State", L2).
 
 get_global_counters(Config) ->
-    get_global_counters0(Config, [{protocol, amqp10}]).
+    get_global_counters0(Config, #{protocol => amqp10}).
 
 get_global_counters(Config, QType) ->
-    get_global_counters0(Config, [{protocol, amqp10},
-                                  {queue_type, QType}]).
+    get_global_counters0(Config, #{protocol => amqp10,
+                                   queue_type => QType}).
 
 get_global_counters0(Config, Key) ->
     Overview = rpc(Config, rabbit_global_counters, overview, []),

deps/rabbit/test/dead_lettering_SUITE.erl

@@ -1936,7 +1936,7 @@ counted(Metric, Config) ->
         metric(QueueType, Strategy, Metric, OldCounters).
 
 metric(QueueType, Strategy, Metric, Counters) ->
-    Metrics = maps:get([{queue_type, QueueType}, {dead_letter_strategy, Strategy}], Counters),
+    Metrics = maps:get(#{queue_type => QueueType, dead_letter_strategy => Strategy}, Counters),
     maps:get(Metric, Metrics).
 
 group_name(Config) ->

deps/rabbit/test/queue_type_SUITE.erl

@@ -162,7 +162,7 @@ smoke(Config) ->
     ok = publish_and_confirm(Ch, <<"non-existent_queue">>, <<"msg4">>),
     ConsumerTag3 = <<"ctag3">>,
     ok = subscribe(Ch, QName, ConsumerTag3),
-    ProtocolCounters = maps:get([{protocol, amqp091}], get_global_counters(Config)),
+    ProtocolCounters = maps:get(#{protocol => amqp091}, get_global_counters(Config)),
     ?assertEqual(#{
                    messages_confirmed_total => 4,
                    messages_received_confirm_total => 4,
@@ -177,7 +177,7 @@ smoke(Config) ->
                   "rabbit_" ++
                   binary_to_list(?config(queue_type, Config)) ++
                   "_queue"),
-    ProtocolQueueTypeCounters = maps:get([{protocol, amqp091}, {queue_type, QueueType}],
+    ProtocolQueueTypeCounters = maps:get(#{protocol => amqp091, queue_type => QueueType},
                                          get_global_counters(Config)),
     ?assertEqual(#{
                    messages_acknowledged_total => 3,
@@ -196,7 +196,7 @@ smoke(Config) ->
     ?assertMatch(
        #{consumers := 0,
          publishers := 0},
-       maps:get([{protocol, amqp091}], get_global_counters(Config))),
+       maps:get(#{protocol => amqp091}, get_global_counters(Config))),
 
     ok.
 

deps/rabbit/test/quorum_queue_SUITE.erl

@@ -3145,6 +3145,9 @@ reconnect_consumer_and_wait_channel_down(Config) ->
         {#'basic.deliver'{redelivered  = false}, _} ->
             wait_for_messages_ready(Servers, RaName, 0),
             wait_for_messages_pending_ack(Servers, RaName, 1)
+    after 30000 ->
+            flush(1),
+            exit(basic_deliver_timeout)
     end,
     Up = [Leader, F2],
     rabbit_ct_broker_helpers:block_traffic_between(F1, Leader),

deps/rabbit/test/rabbit_fifo_dlx_integration_SUITE.erl

@@ -991,5 +991,5 @@ counted(Metric, Config) ->
     metric(Metric, OldCounters).
 
 metric(Metric, Counters) ->
-    Metrics = maps:get([{queue_type, rabbit_quorum_queue}, {dead_letter_strategy, at_least_once}], Counters),
+    Metrics = maps:get(#{queue_type => rabbit_quorum_queue, dead_letter_strategy => at_least_once}, Counters),
     maps:get(Metric, Metrics).