Emit queue_info metric

To allow filtering on queue type or membership status,
we need an info metric for queues; see
https://grafana.com/blog/2021/08/04/how-to-use-promql-joins-for-more-effective-queries-of-prometheus-metrics-at-scale/#info-metrics

With this change, per-object metrics and the detailed metrics
(if queue-related families are requested) will contain
rabbitmq_queue_info / rabbitmq_detailed_queue_info with a value of 1
and labels including the queue name, vhost, queue type and membership
status.

Author: mkuratczyk

deps/rabbitmq_prometheus/src/collectors/prometheus_rabbitmq_core_metrics_collector.erl

@@ -301,14 +301,25 @@ register() ->
 deregister_cleanup(_) -> ok.
 
 collect_mf('detailed', Callback) ->
-    collect(true, ?DETAILED_METRIC_NAME_PREFIX, vhosts_filter_from_pdict(), enabled_mfs_from_pdict(?METRICS_RAW), Callback),
+    IncludedMFs = enabled_mfs_from_pdict(?METRICS_RAW),
+    collect(true, ?DETAILED_METRIC_NAME_PREFIX, vhosts_filter_from_pdict(), IncludedMFs, Callback),
     collect(true, ?CLUSTER_METRIC_NAME_PREFIX, vhosts_filter_from_pdict(), enabled_mfs_from_pdict(?METRICS_CLUSTER), Callback),
+    %% the detailed endpoint should emit queue_info only if queue metrics were requested
+    MFs = proplists:get_keys(IncludedMFs),
+    case lists:member(queue_coarse_metrics, MFs) orelse
+         lists:member(queue_consumer_count, MFs) orelse
+         lists:member(queue_metrics, MFs) of
+        true ->
+            emit_queue_info(?DETAILED_METRIC_NAME_PREFIX, vhosts_filter_from_pdict(), Callback);
+        false -> ok
+    end,
     %% identity is here to enable filtering on a cluster name (as already happens in existing dashboards)
     emit_identity_info(<<"detailed">>, Callback),
     ok;
 collect_mf('per-object', Callback) ->
     collect(true, ?METRIC_NAME_PREFIX, false, ?METRICS_RAW, Callback),
     totals(Callback),
+    emit_queue_info(?METRIC_NAME_PREFIX, false, Callback),
     emit_identity_info(<<"per-object">>, Callback),
     ok;
 collect_mf('memory-breakdown', Callback) ->
@@ -406,6 +417,47 @@ identity_info(Endpoint) ->
         }]
     }.
 
+has_leader_running_locally(Q) when is_pid(Q) ->
+    node(Q) =:= node() andalso is_process_alive(Q);
+has_leader_running_locally({Name, Node}) when Node =:= node() ->
+    is_process_alive(whereis(Name));
+has_leader_running_locally(_) ->
+    false.
+
+emit_queue_info(Prefix, VHostsFilter, Callback) ->
+    Help = <<"A metric with a constant '1' value and labels that provide some queue details">>,
+    QInfos = lists:foldl(
+               fun(Q, Acc) ->
+                       #resource{virtual_host = VHost, name = Name} = amqqueue:get_name(Q),
+                       case is_map(VHostsFilter) andalso maps:get(VHost, VHostsFilter) == false of
+                           true -> Acc;
+                           false ->
+                               Type = amqqueue:get_type(Q),
+                               TypeState = amqqueue:get_type_state(Q),
+                               QInfo0 = [{vhost, VHost}, {queue, Name}, {queue_type, Type}],
+                               Members = maps:get(nodes, TypeState, []),
+                               case {rabbit_amqqueue:is_replicated(Q),
+                                     has_leader_running_locally(amqqueue:get_pid(Q)),
+                                     lists:member(node(), Members)} of
+                                   {false, true, _} ->
+                                       %% non-replicated queue running on this node
+                                       [{QInfo0, 1}|Acc];
+                                   {true, true, _} ->
+                                       %% replicated queue with a leader on this node
+                                       QInfo = [{membership, leader} | QInfo0],
+                                       [{QInfo, 1}|Acc];
+                                   {true, false, true} ->
+                                       %% replicated queue with a non-leader member on this node
+                                       QInfo = [{membership, follower} | QInfo0],
+                                       [{QInfo, 1}|Acc];
+                                   _ ->
+                                       %% ignore queues with no local members
+                                       Acc
+                               end
+                       end
+               end, [], rabbit_amqqueue:list()),
+    Callback(prometheus_model_helpers:create_mf(<<Prefix/binary, "queue_info">>, Help, gauge, QInfos)).
+
 add_metric_family({Name, Type, Help, Metrics}, Callback) ->
     MN = <<?METRIC_NAME_PREFIX/binary, (prometheus_model_helpers:metric_name(Name))/binary>>,
     Callback(create_mf(MN, Help, Type, Metrics)).
@@ -890,4 +942,3 @@ vhosts_filter_from_pdict() ->
             Enabled = maps:from_list([ {VHost, true} || VHost <- L ]),
             maps:merge(All, Enabled)
     end.
-

deps/rabbitmq_prometheus/test/rabbit_prometheus_http_SUITE.erl

@@ -533,19 +533,32 @@ queue_consumer_count_single_vhost_per_object_test(Config) ->
 
     %% There should be exactly 2 metrics returned (2 queues in that vhost, `queue_consumer_count` has only single metric)
     ?assertEqual(#{rabbitmq_detailed_queue_consumers =>
-                       #{#{queue => "vhost-1-queue-with-consumer",vhost => "vhost-1"} => [1],
-                         #{queue => "vhost-1-queue-with-messages",vhost => "vhost-1"} => [0]}},
+                   #{#{queue => "vhost-1-queue-with-consumer",vhost => "vhost-1"} => [1],
+                     #{queue => "vhost-1-queue-with-messages",vhost => "vhost-1"} => [0]},
+                   rabbitmq_detailed_queue_info =>
+                   #{#{queue => "vhost-1-queue-with-consumer", vhost => "vhost-1", queue_type => "rabbit_classic_queue"} => [1],
+                     #{queue => "vhost-1-queue-with-messages", vhost => "vhost-1", queue_type => "rabbit_classic_queue"} => [1]}
+                  },
                  parse_response(Body)),
     ok.
 
 queue_consumer_count_all_vhosts_per_object_test(Config) ->
     Expected = #{rabbitmq_detailed_queue_consumers =>
-                     #{#{queue => "vhost-1-queue-with-consumer",vhost => "vhost-1"} => [1],
-                       #{queue => "vhost-1-queue-with-messages",vhost => "vhost-1"} => [0],
-                       #{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 => "vhost-1-queue-with-consumer",vhost => "vhost-1"} => [1],
+                   #{queue => "vhost-1-queue-with-messages",vhost => "vhost-1"} => [0],
+                   #{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]},
+
+                 rabbitmq_detailed_queue_info =>
+                 #{#{queue => "default-queue-with-consumer", vhost => "/", queue_type => "rabbit_classic_queue"} => [1],
+                   #{queue => "default-queue-with-messages", vhost => "/", queue_type => "rabbit_classic_queue"} => [1],
+                   #{queue => "vhost-1-queue-with-consumer", vhost => "vhost-1", queue_type => "rabbit_classic_queue"} => [1],
+                   #{queue => "vhost-1-queue-with-messages", vhost => "vhost-1", queue_type => "rabbit_classic_queue"} => [1],
+                   #{queue => "vhost-2-queue-with-consumer", vhost => "vhost-2", queue_type => "rabbit_classic_queue"} => [1],
+                   #{queue => "vhost-2-queue-with-messages", vhost => "vhost-2", queue_type => "rabbit_classic_queue"} => [1]}
+                },
 
     %% No vhost given, all should be returned
     {_, Body1} = http_get_with_pal(Config, "/metrics/detailed?family=queue_consumer_count&per-object=1", [], 200),