Publish aggregated/per-object/detailed RAFT metrics

mkuratczyk · mkuratczyk · commit 2a6fe6dc4bba · 2025-05-26T12:08:22.000+02:00
For aggregated metrics, we just pick specific metrics (currently
num_segments and commit_latency) and only publish the maximum value,
without labels (`max_` is added to the metric name). For example:

```
&gt; curl -s localhost:15692/metrics/per-object | rg -e ^rabbitmq_raft_num_segments -e ^rabbitmq_raft_commit_latency
rabbitmq_raft_commit_latency_seconds{module="rabbit_khepri",ra_system="coordination"} 0.0
rabbitmq_raft_commit_latency_seconds{queue="qq2",vhost="/"} 0.02
rabbitmq_raft_commit_latency_seconds{queue="qqq-1",vhost="/"} 0.01
rabbitmq_raft_commit_latency_seconds{queue="qqq-2",vhost="/"} 0.0
rabbitmq_raft_num_segments{module="rabbit_khepri",ra_system="coordination"} 1.0
rabbitmq_raft_num_segments{queue="qq2",vhost="/"} 132.0
rabbitmq_raft_num_segments{queue="qqq-2",vhost="/"} 245.0

&gt; curl -s localhost:15692/metrics/ | rg ^rabbitmq_raft_max
rabbitmq_raft_max_commit_latency_seconds 0.02
rabbitmq_raft_max_num_segments 245.0
```
diff --git a/deps/rabbitmq_prometheus/src/collectors/prometheus_rabbitmq_raft_metrics_collector.erl b/deps/rabbitmq_prometheus/src/collectors/prometheus_rabbitmq_raft_metrics_collector.erl
@@ -7,7 +7,6 @@
 -module(prometheus_rabbitmq_raft_metrics_collector).
 
 -behaviour(prometheus_collector).
--include_lib("prometheus/include/prometheus.hrl").
 
 -export([register/0,
          deregister_cleanup/1,
@@ -16,7 +15,8 @@
 -import(prometheus_model_helpers, [create_mf/4,
                                    counter_metric/2]).
 
--define(METRIC_NAME_PREFIX, "rabbitmq_raft_").
+-define(METRIC_NAME_PREFIX, <<"rabbitmq_raft_">>).
+-define(DETAILED_METRIC_NAME_PREFIX, <<"rabbitmq_raft_detailed_">>).
 
 %%====================================================================
 %% Collector API
@@ -28,19 +28,108 @@ register() ->
 deregister_cleanup(_) ->
     ok.
 
+collect_mf('per-object', Callback) ->
+    collect_per_object_metrics(?METRIC_NAME_PREFIX, Callback);
+collect_mf('detailed', Callback) ->
+    case get(prometheus_mf_filter) of
+        undefined ->
+            ok;
+        MFNames ->
+            case lists:member(raft_metrics, MFNames) of
+                true ->
+                    collect_detailed_metrics(?DETAILED_METRIC_NAME_PREFIX, Callback);
+                false ->
+                    ok
+            end
+    end;
 collect_mf(_Registry, Callback) ->
+    case application:get_env(rabbitmq_prometheus, return_per_object_metrics, false) of
+        false ->
+            collect_aggregate_metrics(?METRIC_NAME_PREFIX, Callback);
+        true ->
+            collect_per_object_metrics(?METRIC_NAME_PREFIX, Callback)
+    end.
+
+%% INTERNAL
+
+collect_aggregate_metrics(Prefix, Callback) ->
+    collect_max_values(Prefix, Callback),
+    collect_key_component_metrics(Prefix, Callback).
+
+collect_per_object_metrics(Prefix, Callback) ->
+    collect_key_component_metrics(Prefix, Callback),
+    collect_key_per_object_metrics(?METRIC_NAME_PREFIX, Callback).
+
+collect_detailed_metrics(Prefix, Callback) ->
+    VHostFilterFun = case get(prometheus_vhost_filter) of
+                         undefined ->
+                             fun(_) -> true end;
+                         VHosts ->
+                             fun(VHost) -> lists:member(VHost, VHosts) end
+                     end,
+    collect_all_matching_metrics(Prefix, Callback, VHostFilterFun).
+
+collect_key_per_object_metrics(Prefix, Callback) ->
     maps:foreach(
       fun(Name, #{type := Type, help := Help, values := Values}) ->
               Callback(
-                create_mf(?METRIC_NAME(Name),
+                create_mf(<<Prefix/binary, (prometheus_model_helpers:metric_name(Name))/binary>>,
                           Help,
                           Type,
                           Values))
       end,
       seshat:format(ra, [term,
-			 snapshot_index,
-			 last_applied,
-			 commit_index,
-			 last_written_index,
-			 commit_latency,
-			 num_segments])).
+                         snapshot_index,
+                         last_applied,
+                         commit_index,
+                         last_written_index,
+                         commit_latency,
+                         num_segments])).
+
+collect_all_matching_metrics(Prefix, Callback, VHostFilterFun) ->
+    maps:foreach(
+      fun(Name, #{type := Type, help := Help, values := Values0}) ->
+              Values = maps:filter(fun(#{vhost := V}, _) ->
+                                           VHostFilterFun(V);
+                                      (_, _) -> true
+                                   end, Values0),
+              Callback(
+                create_mf(<<Prefix/binary, (prometheus_model_helpers:metric_name(Name))/binary>>,
+                          Help,
+                          Type,
+                          Values))
+      end,
+      seshat:format(ra)).
+
+collect_max_values(Prefix, Callback) ->
+    %% max values for QQ metrics
+    %% eg.
+    %% rabbitmq_raft_num_segments{queue="q1",vhost="/"} 5.0
+    %% rabbitmq_raft_num_segments{queue="q2",vhost="/"} 10.0
+    %% becomes
+    %% rabbitmq_raft_max_num_segments 10.0
+    QQMetrics = [num_segments],
+    maps:foreach(
+      fun(Name, #{type := Type, help := Help, values := Values}) ->
+              Max = lists:max(maps:values(Values)),
+              Callback(
+                create_mf(<<Prefix/binary, "max_", (prometheus_model_helpers:metric_name(Name))/binary>>,
+                          Help,
+                          Type,
+                          #{#{} => Max}))
+
+      end,
+      seshat:format(ra, QQMetrics)).
+
+collect_key_component_metrics(Prefix, Callback) ->
+    WALMetrics = [wal_files, bytes_written, mem_tables],
+    SegmentWriterMetrics = [entries, segments],
+    maps:foreach(
+      fun(Name, #{type := Type, help := Help, values := Values}) ->
+              Callback(
+                create_mf(<<Prefix/binary, (prometheus_model_helpers:metric_name(Name))/binary>>,
+                          Help,
+                          Type,
+                          Values))
+      end,
+      seshat:format(ra, WALMetrics ++ SegmentWriterMetrics)).
diff --git a/deps/rabbitmq_prometheus/src/rabbit_prometheus_dispatcher.erl b/deps/rabbitmq_prometheus/src/rabbit_prometheus_dispatcher.erl
@@ -17,6 +17,7 @@ build_dispatcher() ->
         prometheus_rabbitmq_core_metrics_collector,
         prometheus_rabbitmq_global_metrics_collector,
         prometheus_rabbitmq_message_size_metrics_collector,
+        prometheus_rabbitmq_raft_metrics_collector,
         prometheus_rabbitmq_alarm_metrics_collector,
         prometheus_rabbitmq_dynamic_collector,
         prometheus_process_collector],
@@ -26,8 +27,7 @@ build_dispatcher() ->
         prometheus_vm_memory_collector,
         prometheus_mnesia_collector,
         prometheus_vm_statistics_collector,
-        prometheus_vm_msacc_collector,
-        prometheus_rabbitmq_raft_metrics_collector
+        prometheus_vm_msacc_collector
     ],
     prometheus_registry:register_collectors(
         case application:get_env(rabbitmq_prometheus, return_per_object_metrics, fasle) of
@@ -38,7 +38,8 @@ build_dispatcher() ->
     prometheus_registry:register_collectors('per-object',
         CoreCollectors ++ PerObjectCollectors),
     prometheus_registry:register_collectors('detailed', [
-        prometheus_rabbitmq_core_metrics_collector
+        prometheus_rabbitmq_core_metrics_collector,
+        prometheus_rabbitmq_raft_metrics_collector
         ]),
     prometheus_registry:register_collectors('memory-breakdown', [
         prometheus_rabbitmq_core_metrics_collector
diff --git a/deps/rabbitmq_prometheus/test/rabbit_prometheus_http_SUITE.erl b/deps/rabbitmq_prometheus/test/rabbit_prometheus_http_SUITE.erl
@@ -72,7 +72,8 @@ groups() ->
                                      vhost_status_metric,
                                      exchange_bindings_metric,
                                      exchange_names_metric,
-                                     stream_pub_sub_metrics
+                                     stream_pub_sub_metrics,
+                                     raft_metrics_per_object_test
         ]},
        {special_chars, [], [core_metrics_special_chars]},
        {authentication, [], [basic_auth]}
@@ -158,6 +159,12 @@ init_per_group(detailed_metrics, Config0) ->
         Q <- [ <<"queue-with-messages">>, <<"queue-with-consumer">> ]
     ],
 
+    amqp_channel:call(DefaultCh,
+                      #'queue.declare'{queue = <<"a_quorum_queue">>,
+                                       durable = true,
+                                       arguments = [{<<"x-queue-type">>, longstr, <<"quorum">>}]
+                                      }),
+
     DefaultConsumer = sleeping_consumer(),
     #'basic.consume_ok'{consumer_tag = DefaultCTag} =
         amqp_channel:subscribe(DefaultCh, #'basic.consume'{queue = <<"default-queue-with-consumer">>}, DefaultConsumer),
@@ -400,7 +407,14 @@ aggregated_metrics_test(Config) ->
     ?assertEqual(match, re:run(Body, "^rabbitmq_erlang_uptime_seconds ", [{capture, none}, multiline])),
     ?assertEqual(match, re:run(Body, "^rabbitmq_io_read_time_seconds_total ", [{capture, none}, multiline])),
     %% Check the first TOTALS metric value
-    ?assertEqual(match, re:run(Body, "^rabbitmq_connections ", [{capture, none}, multiline])).
+    ?assertEqual(match, re:run(Body, "^rabbitmq_connections ", [{capture, none}, multiline])),
+    ?assertEqual(nomatch, re:run(Body, "^rabbitmq_raft_commit_latency_seconds", [{capture, none}, multiline])),
+    ?assertEqual(match, re:run(Body, "^rabbitmq_raft_bytes_written.*ra_log_segment_writer", [{capture, none}, multiline])),
+    ?assertEqual(match, re:run(Body, "^rabbitmq_raft_bytes_written.*ra_log_wal", [{capture, none}, multiline])),
+    ?assertEqual(match, re:run(Body, "^rabbitmq_raft_entries{", [{capture, none}, multiline])),
+    ?assertEqual(match, re:run(Body, "^rabbitmq_raft_mem_tables{", [{capture, none}, multiline])),
+    ?assertEqual(match, re:run(Body, "^rabbitmq_raft_segments{", [{capture, none}, multiline])),
+    ?assertEqual(match, re:run(Body, "^rabbitmq_raft_wal_files{", [{capture, none}, multiline])).
 
 endpoint_per_object_metrics(Config) ->
     per_object_metrics_test(Config, "/metrics/per-object").
@@ -438,7 +452,8 @@ per_object_metrics_test(Config, Path) ->
     ?assertEqual(match, re:run(Body, "^rabbitmq_io_read_time_seconds_total ", [{capture, none}, multiline])),
     ?assertEqual(match, re:run(Body, "^rabbitmq_raft_commit_latency_seconds{", [{capture, none}, multiline])),
     %% Check the first TOTALS metric value
-    ?assertEqual(match, re:run(Body, "^rabbitmq_connections ", [{capture, none}, multiline])).
+    ?assertEqual(match, re:run(Body, "^rabbitmq_connections ", [{capture, none}, multiline])),
+    ?assertEqual(match, re:run(Body, "^rabbitmq_raft_num_segments{", [{capture, none}, multiline])).
 
 memory_breakdown_metrics_test(Config) ->
     {_Headers, Body} = http_get_with_pal(Config, "/metrics/memory-breakdown", [], 200),
@@ -552,7 +567,8 @@ queue_consumer_count_all_vhosts_per_object_test(Config) ->
                    #{queue => "vhost-2-queue-with-consumer",vhost => "vhost-2"} => [1],
                    #{queue => "vhost-2-queue-with-messages",vhost => "vhost-2"} => [0],
                    #{queue => "default-queue-with-consumer",vhost => "/"} => [1],
-                   #{queue => "default-queue-with-messages",vhost => "/"} => [0]},
+                   #{queue => "default-queue-with-messages",vhost => "/"} => [0],
+                   #{queue => "a_quorum_queue",vhost => "/"} => [0]},
 
                  rabbitmq_detailed_queue_info =>
                  #{#{queue => "default-queue-with-consumer",
@@ -578,7 +594,10 @@ queue_consumer_count_all_vhosts_per_object_test(Config) ->
                    #{queue => "vhost-2-queue-with-messages",
                      vhost => "vhost-2",
                      queue_type => "rabbit_classic_queue",
-                     membership => "leader"} => [1]}
+                     membership => "leader"} => [1],
+                   #{membership => "leader",
+                     queue => "a_quorum_queue",vhost => "/",
+                     queue_type => "rabbit_quorum_queue"} => [1]}
                 },
 
     %% No vhost given, all should be returned
@@ -596,7 +615,8 @@ queue_coarse_metrics_per_object_test(Config) ->
     Expected2 =  #{#{queue => "vhost-2-queue-with-consumer", vhost => "vhost-2"} => [11],
                    #{queue => "vhost-2-queue-with-messages", vhost => "vhost-2"} => [11]},
     ExpectedD =  #{#{queue => "default-queue-with-consumer", vhost => "/"} => [3],
-                   #{queue => "default-queue-with-messages", vhost => "/"} => [3]},
+                   #{queue => "default-queue-with-messages", vhost => "/"} => [3],
+                   #{queue => "a_quorum_queue",vhost => "/"} => [0]},
 
     {_, Body1} = http_get_with_pal(Config, "/metrics/detailed?vhost=vhost-1&family=queue_coarse_metrics", [], 200),
     ?assertEqual(Expected1,
@@ -704,7 +724,8 @@ queue_metrics_per_object_test(Config) ->
     Expected2 =  #{#{queue => "vhost-2-queue-with-consumer", vhost => "vhost-2"} => [11],
                    #{queue => "vhost-2-queue-with-messages", vhost => "vhost-2"} => [1]},
     ExpectedD =  #{#{queue => "default-queue-with-consumer", vhost => "/"} => [3],
-                   #{queue => "default-queue-with-messages", vhost => "/"} => [1]},
+                   #{queue => "default-queue-with-messages", vhost => "/"} => [1],
+                   #{queue => "a_quorum_queue",vhost => "/"} => [0]},
     {_, Body1} = http_get_with_pal(Config, "/metrics/detailed?vhost=vhost-1&family=queue_metrics", [], 200),
     ?assertEqual(Expected1,
                  map_get(rabbitmq_detailed_queue_messages_ram, parse_response(Body1))),
@@ -835,6 +856,21 @@ core_metrics_special_chars(Config) ->
                       maps:to_list(LabelValue3)),
     ok.
 
+raft_metrics_per_object_test(Config) ->
+    Expected1 =  #{#{module => "rabbit_khepri", ra_system => "coordination"} => ["0.0"],
+                   #{module => "rabbit_stream_coordinator", ra_system => "coordination"} => ["0.0"]},
+    Expected2 =  Expected1#{#{queue => "a_quorum_queue", vhost => "/"} => ["0.0"]},
+
+    {_, Body1} = http_get_with_pal(Config, "/metrics/detailed?family=raft_metrics&vhost=foo", [], 200),
+    ?assertEqual(Expected1,
+                 map_get(rabbitmq_raft_detailed_num_segments, parse_response(Body1))),
+
+    {_, Body2} = http_get_with_pal(Config, "/metrics/detailed?family=raft_metrics&vhost=/", [], 200),
+    ?assertEqual(Expected2,
+                 map_get(rabbitmq_raft_detailed_num_segments, parse_response(Body2))),
+
+    ok.
+
 basic_auth(Config) ->
     http_get(Config, [{"accept-encoding", "deflate"}], 401),
     AuthHeader = rabbit_mgmt_test_util:auth_header("guest", "guest"),
