Merge pull request #3779 from rabbitmq/vhost-exchange-count

Add optional metrics for vhost and exchange count

3 files changed, 152 insertions, 4 deletions

diff --git a/deps/rabbitmq_prometheus/metrics-detailed.md b/deps/rabbitmq_prometheus/metrics-detailed.md
index 262fca04bb..bec732f8db 100644
--- a/deps/rabbitmq_prometheus/metrics-detailed.md
+++ b/deps/rabbitmq_prometheus/metrics-detailed.md
@@ -236,3 +236,32 @@ Group `channel_queue_exchange_metrics`:
 | Metric                                           | Description                                  |
 |--------------------------------------------------|----------------------------------------------|
 | rabbitmq_detailed_queue_messages_published_total | Total number of messages published to queues |
+
+### Virtual hosts and exchange metrics
+
+These additional metrics can be useful when virtual hosts or exchanges are
+created on a shared cluster in a self-service way. They are different
+from the rest of the metrics: they are cluster-wide and not node-local.
+These metrics **must not** be aggregated across cluster nodes.
+
+Group `vhost_status`:
+
+| Metric                        | Description                      |
+|-------------------------------|----------------------------------|
+| rabbitmq_cluster_vhost_status | Whether a given vhost is running |
+
+Group `exchange_names`:
+
+| Metric                         | Description                                                                                                                |
+|--------------------------------|----------------------------------------------------------------------------------------------------------------------------|
+| rabbitmq_cluster_exchange_name | Enumerates exchanges without any additional info. This value is cluster-wide. A cheaper alternative to `exchange_bindings` |
+
+Group `exchange_bindings`:
+
+| Metric                             | Description                                                     |
+|------------------------------------|-----------------------------------------------------------------|
+| rabbitmq_cluster_exchange_bindings | Number of bindings for an exchange. This value is cluster-wide. |
+
+
+
+
diff --git a/deps/rabbitmq_prometheus/src/collectors/prometheus_rabbitmq_core_metrics_collector.erl b/deps/rabbitmq_prometheus/src/collectors/prometheus_rabbitmq_core_metrics_collector.erl
index 328ec00276..7c31b71b92 100644
--- a/deps/rabbitmq_prometheus/src/collectors/prometheus_rabbitmq_core_metrics_collector.erl
+++ b/deps/rabbitmq_prometheus/src/collectors/prometheus_rabbitmq_core_metrics_collector.erl
@@ -29,6 +29,7 @@
 
 %% Used by `/metrics/detailed` endpoint
 -define(DETAILED_METRIC_NAME_PREFIX, <<"rabbitmq_detailed_">>).
+-define(CLUSTER_METRIC_NAME_PREFIX, <<"rabbitmq_cluster_">>).
 
 %% ==The source of these metrics can be found in the rabbit_core_metrics module==
 %% The relevant files are:
@@ -214,6 +215,19 @@
     ]}
 ]).
 
+%% Metrics that can be only requested through `/metrics/detailed`
+-define(METRICS_CLUSTER,[
+    {vhost_status, [
+        {2, undefined, vhost_status, gauge, "Whether a given vhost is running"}
+    ]},
+    {exchange_bindings, [
+        {2, undefined, exchange_bindings, gauge, "Number of bindings for an exchange. This value is cluster-wide."}
+    ]},
+    {exchange_names, [
+        {2, undefined, exchange_name, gauge, "Enumerates exchanges without any additional info. This value is cluster-wide. A cheaper alternative to `exchange_bindings`"}
+    ]}
+]).
+
 -define(TOTALS, [
     %% ordering differs from metrics above, refer to list comprehension
     {connection_created, connections, gauge, "Connections currently open"},
@@ -232,7 +246,8 @@ register() ->
 deregister_cleanup(_) -> ok.
 
 collect_mf('detailed', Callback) ->
-    collect(true, ?DETAILED_METRIC_NAME_PREFIX, vhosts_filter_from_pdict(), queues_filter_from_pdict(), enabled_mfs_from_pdict(), Callback),
+    collect(true, ?DETAILED_METRIC_NAME_PREFIX, vhosts_filter_from_pdict(), queues_filter_from_pdict(), enabled_mfs_from_pdict(?METRICS_RAW), Callback),
+    collect(true, ?CLUSTER_METRIC_NAME_PREFIX, vhosts_filter_from_pdict(), queues_filter_from_pdict(),  enabled_mfs_from_pdict(?METRICS_CLUSTER), Callback),
     %% identity is here to enable filtering on a cluster name (as already happens in existing dashboards)
     emit_identity_info(Callback),
     ok;
@@ -383,6 +398,14 @@ collect_metrics(_, {Type, Fun, Items}) ->
 labels(Item) ->
     label(element(1, Item)).
 
+label(L) when is_binary(L) ->
+    L;
+label(M) when is_map(M) ->
+    maps:fold(fun (K, V, Acc = <<>>) ->
+                      <<Acc/binary, K/binary, "=\"", V/binary, "\"">>;
+                  (K, V, Acc) ->
+                      <<Acc/binary, ",", K/binary, "=\"", V/binary, "\"">>
+              end, <<>>, M);
 label(#resource{virtual_host = VHost, kind = exchange, name = Name}) ->
     <<"vhost=\"", VHost/binary, "\",exchange=\"", Name/binary, "\"">>;
 label(#resource{virtual_host = VHost, kind = queue, name = Name}) ->
@@ -589,6 +612,46 @@ get_data(MF, true, VHostsFilter, _) when is_map(VHostsFilter), MF == queue_metri
               end, [], Table);
 get_data(queue_consumer_count, true, _, _) ->
     ets:tab2list(queue_metrics);
+get_data(vhost_status, _, _, _) ->
+    [ { #{<<"vhost">> => VHost},
+        case rabbit_vhost_sup_sup:is_vhost_alive(VHost) of
+            true -> 1;
+            false -> 0
+        end}
+      || VHost <- rabbit_vhost:list()  ];
+get_data(exchange_bindings, _, _, _) ->
+    Exchanges = lists:foldl(fun
+                                (#exchange{internal = true}, Acc) ->
+                                    Acc;
+                                (#exchange{name = #resource{name = <<>>}}, Acc) ->
+                                    Acc;
+                                (#exchange{name = EName, type = EType}, Acc) ->
+                                    maps:put(EName, #{type => atom_to_binary(EType), binding_count => 0}, Acc)
+                            end, #{}, rabbit_exchange:list()),
+    WithCount = ets:foldl(
+                  fun (#route{binding = #binding{source = EName}}, Acc) ->
+                          case maps:is_key(EName, Acc) of
+                              false -> Acc;
+                              true ->
+                                  maps:update_with(EName, fun (R = #{binding_count := Cnt}) ->
+                                                                  R#{binding_count => Cnt + 1}
+                                                          end, Acc)
+                          end
+                  end, Exchanges, rabbit_route),
+    maps:fold(fun(#resource{virtual_host = VHost, name = Name}, #{type := Type, binding_count := Bindings}, Acc) ->
+                      [{<<"vhost=\"", VHost/binary, "\",exchange=\"", Name/binary, "\",type=\"", Type/binary, "\"">>,
+                        Bindings}|Acc]
+              end, [], WithCount);
+get_data(exchange_names, _, _, _) ->
+    lists:foldl(fun
+                    (#exchange{internal = true}, Acc) ->
+                        Acc;
+                    (#exchange{name = #resource{name = <<>>}}, Acc) ->
+                        Acc;
+                    (#exchange{name = #resource{virtual_host = VHost, name = Name}, type = EType}, Acc) ->
+                        Label = <<"vhost=\"", VHost/binary, "\",exchange=\"", Name/binary, "\",type=\"", (atom_to_binary(EType))/binary, "\"">>,
+                        [{Label, 1}|Acc]
+                end, [], rabbit_exchange:list());
 get_data(Table, _, _, _) ->
     ets:tab2list(Table).
 
@@ -642,13 +705,13 @@ sum('', B) ->
 sum(A, B) ->
     A + B.
 
-enabled_mfs_from_pdict() ->
+enabled_mfs_from_pdict(AllMFs) ->
     case get(prometheus_mf_filter) of
         undefined ->
             [];
         MFNames ->
             MFNameSet = sets:from_list(MFNames),
-            [ MF || MF = {Table, _} <- ?METRICS_RAW, sets:is_element(Table, MFNameSet) ]
+            [ MF || MF = {Table, _} <- AllMFs, sets:is_element(Table, MFNameSet) ]
     end.
 
 vhosts_filter_from_pdict() ->
diff --git a/deps/rabbitmq_prometheus/test/rabbit_prometheus_http_SUITE.erl b/deps/rabbitmq_prometheus/test/rabbit_prometheus_http_SUITE.erl
index 10c573988c..c8be5fae28 100644
--- a/deps/rabbitmq_prometheus/test/rabbit_prometheus_http_SUITE.erl
+++ b/deps/rabbitmq_prometheus/test/rabbit_prometheus_http_SUITE.erl
@@ -54,7 +54,10 @@ groups() ->
                                      queue_consumer_count_all_vhosts_per_object_test,
                                      queue_coarse_metrics_per_object_test,
                                      queue_metrics_per_object_test,
-                                     queue_consumer_count_and_queue_metrics_mutually_exclusive_test
+                                     queue_consumer_count_and_queue_metrics_mutually_exclusive_test,
+                                     vhost_status_metric,
+                                     exchange_bindings_metric,
+                                     exchange_names_metric
         ]}
     ].
 
@@ -120,6 +123,14 @@ init_per_group(detailed_metrics, Config0) ->
                                      amqp_channel:cast(Ch,
                                                        #'basic.publish'{routing_key = QName},
                                                        #amqp_msg{payload = <<"msg">>})
+                             end, lists:seq(1, MsgNum) ),
+              ExDirect = <<QName/binary, "-direct-exchange">>,
+              #'exchange.declare_ok'{} = amqp_channel:call(Ch, #'exchange.declare'{exchange = ExDirect}),
+              ExTopic = <<QName/binary, "-topic-exchange">>,
+              #'exchange.declare_ok'{} = amqp_channel:call(Ch, #'exchange.declare'{exchange = ExTopic, type = <<"topic">>}),
+              #'queue.bind_ok'{} = amqp_channel:call(Ch, #'queue.bind'{queue = QName, exchange = ExDirect, routing_key = QName}),
+              lists:foreach( fun (Idx) ->
+                                     #'queue.bind_ok'{} = amqp_channel:call(Ch, #'queue.bind'{queue = QName, exchange = ExTopic, routing_key = integer_to_binary(Idx)})
                              end, lists:seq(1, MsgNum) )
       end)()
      || {VHost, Ch, MsgNum} <- [{<<"/">>, DefaultCh, 3}, {<<"vhost-1">>, VHost1Ch, 7}, {<<"vhost-2">>, VHost2Ch, 11}],
@@ -488,6 +499,51 @@ detailed_metrics_no_families_enabled_by_default(Config) ->
     ?assertEqual(#{}, parse_response(Body)),
     ok.
 
+vhost_status_metric(Config) ->
+    {_, Body1} = http_get_with_pal(Config, "/metrics/detailed?family=vhost_status", [], 200),
+    Expected = #{rabbitmq_cluster_vhost_status =>
+                     #{#{vhost => "vhost-1"} => [1],
+                       #{vhost => "vhost-2"} => [1],
+                       #{vhost => "/"} => [1]}},
+    ?assertEqual(Expected, parse_response(Body1)),
+    ok.
+
+exchange_bindings_metric(Config) ->
+    {_, Body1} = http_get_with_pal(Config, "/metrics/detailed?family=exchange_bindings", [], 200),
+
+    Bindings = map_get(rabbitmq_cluster_exchange_bindings, parse_response(Body1)),
+    ?assertEqual([11], map_get(#{vhost=>"vhost-2",exchange=>"vhost-2-queue-with-messages-topic-exchange",type=>"topic"}, Bindings)),
+    ?assertEqual([1],  map_get(#{vhost=>"vhost-2",exchange=>"vhost-2-queue-with-messages-direct-exchange",type=>"direct"}, Bindings)),
+    ok.
+
+exchange_names_metric(Config) ->
+    {_, Body1} = http_get_with_pal(Config, "/metrics/detailed?family=exchange_names", [], 200),
+
+    Names = maps:filter(
+              fun
+                  (#{exchange := [$a, $m, $q|_]}, _) ->
+                      false;
+                  (_, _) ->
+                      true
+              end,
+              map_get(rabbitmq_cluster_exchange_name, parse_response(Body1))),
+
+    ?assertEqual(#{ #{vhost=>"vhost-2",exchange=>"vhost-2-queue-with-messages-topic-exchange",type=>"topic"} => [1],
+                    #{vhost=>"vhost-2",exchange=>"vhost-2-queue-with-messages-direct-exchange",type=>"direct"} => [1],
+                    #{vhost=>"vhost-1",exchange=>"vhost-1-queue-with-messages-topic-exchange",type=>"topic"} => [1],
+                    #{vhost=>"vhost-1",exchange=>"vhost-1-queue-with-messages-direct-exchange",type=>"direct"} => [1],
+                    #{vhost=>"/",exchange=>"default-queue-with-messages-topic-exchange",type=>"topic"} => [1],
+                    #{vhost=>"/",exchange=>"default-queue-with-messages-direct-exchange",type=>"direct"} => [1],
+                    #{vhost=>"vhost-2",exchange=>"vhost-2-queue-with-consumer-topic-exchange",type=>"topic"} => [1],
+                    #{vhost=>"vhost-2",exchange=>"vhost-2-queue-with-consumer-direct-exchange",type=>"direct"} => [1],
+                    #{vhost=>"vhost-1",exchange=>"vhost-1-queue-with-consumer-topic-exchange",type=>"topic"} => [1],
+                    #{vhost=>"vhost-1",exchange=>"vhost-1-queue-with-consumer-direct-exchange",type=>"direct"} => [1],
+                    #{vhost=>"/",exchange=>"default-queue-with-consumer-topic-exchange",type=>"topic"} => [1],
+                    #{vhost=>"/",exchange=>"default-queue-with-consumer-direct-exchange",type=>"direct"} => [1]
+                  }, Names),
+    ok.
+
+
 http_get(Config, ReqHeaders, CodeExp) ->
     Path = proplists:get_value(prometheus_path, Config, "/metrics"),
     http_get(Config, Path, ReqHeaders, CodeExp).