RMQ-1460: Emit queue_info metric (#13583)

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.

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,62 @@ identity_info(Endpoint) ->
         }]
     }.
 
+membership(Pid, Members) when is_pid(Pid) ->
+    case node(Pid) =:= node() of
+        true ->
+            case is_process_alive(Pid) of
+                true -> leader;
+                false -> undefined
+            end;
+        false ->
+            case lists:member(node(), Members) of
+                true -> follower;
+                false -> not_a_member
+            end
+    end;
+membership({Name, Node}, Members) ->
+    case Node =:= node() of
+        true ->
+            case is_process_alive(whereis(Name)) of
+                true -> leader;
+                false -> undefined
+            end;
+        false ->
+            case lists:member(node(), Members) of
+                true -> follower;
+                false -> not_a_member
+            end
+    end;
+membership(_, _Members) ->
+    undefined.
+
+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),
+                               Members = maps:get(nodes, TypeState, []),
+                               case membership(amqqueue:get_pid(Q), Members) of
+                                   not_a_member ->
+                                       Acc;
+                                   Membership ->
+                                       QInfo = [
+                                                {vhost, VHost},
+                                                {queue, Name},
+                                                {queue_type, Type},
+                                                {membership, Membership}
+                                               ],
+                                       [{QInfo, 1}|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 +957,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

@@ -534,7 +534,17 @@ 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-messages",vhost => "vhost-1"} => [0]},
+                   rabbitmq_detailed_queue_info =>
+                   #{#{queue => "vhost-1-queue-with-consumer",
+                       vhost => "vhost-1",
+                       queue_type => "rabbit_classic_queue",
+                       membership => "leader"} => [1],
+                     #{queue => "vhost-1-queue-with-messages",
+                       vhost => "vhost-1",
+                       queue_type => "rabbit_classic_queue",
+                       membership => "leader"} => [1]}
+                  },
                  parse_response(Body)),
     ok.
 
@@ -545,7 +555,34 @@ 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]},
+
+                 rabbitmq_detailed_queue_info =>
+                 #{#{queue => "default-queue-with-consumer",
+                     vhost => "/",
+                     queue_type => "rabbit_classic_queue",
+                     membership => "leader"} => [1],
+                   #{queue => "default-queue-with-messages",
+                     vhost => "/",
+                     queue_type => "rabbit_classic_queue",
+                     membership => "leader"} => [1],
+                   #{queue => "vhost-1-queue-with-consumer",
+                     vhost => "vhost-1",
+                     queue_type => "rabbit_classic_queue",
+                     membership => "leader"} => [1],
+                   #{queue => "vhost-1-queue-with-messages",
+                     vhost => "vhost-1",
+                     queue_type => "rabbit_classic_queue",
+                     membership => "leader"} => [1],
+                   #{queue => "vhost-2-queue-with-consumer",
+                     vhost => "vhost-2",
+                     queue_type => "rabbit_classic_queue",
+                     membership => "leader"} => [1],
+                   #{queue => "vhost-2-queue-with-messages",
+                     vhost => "vhost-2",
+                     queue_type => "rabbit_classic_queue",
+                     membership => "leader"} => [1]}
+                },
 
     %% No vhost given, all should be returned
     {_, Body1} = http_get_with_pal(Config, "/metrics/detailed?family=queue_consumer_count&per-object=1", [], 200),