Merge pull request #4727 from rabbitmq/rabbitmq-server-4622-stream-queue-consumer-count

Add consumer count to stream queue metrics

Author: michaelklishin

deps/rabbit/src/rabbit_osiris_metrics.erl

@@ -26,7 +26,8 @@
          online,
          members,
          memory,
-         readers
+         readers,
+         consumers
         ]).
 
 -record(state, {timeout :: non_neg_integer()}).
@@ -79,7 +80,6 @@ handle_info(tick, #state{timeout = Timeout} = State) ->
                               []
                       end,
 
-
               rabbit_core_metrics:queue_stats(QName, Infos),
               rabbit_event:notify(queue_stats, Infos ++ [{name, QName},
                                                          {messages, COffs},

deps/rabbit/src/rabbit_stream_queue.erl

@@ -49,7 +49,8 @@
 
 -define(INFO_KEYS, [name, durable, auto_delete, arguments, leader, members, online, state,
                     messages, messages_ready, messages_unacknowledged, committed_offset,
-                    policy, operator_policy, effective_policy_definition, type, memory]).
+                    policy, operator_policy, effective_policy_definition, type, memory,
+                    consumers]).
 
 -type appender_seq() :: non_neg_integer().
 
@@ -486,6 +487,22 @@ i(leader, Q) when ?is_amqqueue(Q) ->
 i(members, Q) when ?is_amqqueue(Q) ->
     #{nodes := Nodes} = amqqueue:get_type_state(Q),
     Nodes;
+i(consumers, Q) when ?is_amqqueue(Q) ->
+    QName = amqqueue:get_name(Q),
+    #{nodes := Nodes} = amqqueue:get_type_state(Q),
+    Spec = [{{{'$1', '_', '_'}, '_', '_', '_', '_', '_', '_'}, [{'==', {QName}, '$1'}], [true]}],
+    lists:foldl(fun(N, Acc) ->
+                        case rabbit_misc:rpc_call(N,
+                                                  ets,
+                                                  select_count,
+                                                  [consumer_created, Spec],
+                                                  10000) of
+                            Count when is_integer(Count) ->
+                                Acc + Count;
+                            _ ->
+                                Acc
+                        end
+                end, 0, Nodes);
 i(memory, Q) when ?is_amqqueue(Q) ->
     %% Return writer memory. It's not the full memory usage (we also have replica readers on
     %% the writer node), but might be good enough

deps/rabbitmq_management/test/rabbit_mgmt_http_SUITE.erl

@@ -87,6 +87,7 @@ all_tests() -> [
     exchanges_test,
     queues_test,
     quorum_queues_test,
+    stream_queues_have_consumers_field,
     queues_well_formed_json_test,
     bindings_test,
     bindings_post_test,
@@ -239,7 +240,13 @@ init_per_testcase(Testcase = permissions_vhost_test, Config) ->
     rabbit_ct_broker_helpers:delete_vhost(Config, <<"myvhost1">>),
     rabbit_ct_broker_helpers:delete_vhost(Config, <<"myvhost2">>),
     rabbit_ct_helpers:testcase_started(Config, Testcase);
-
+init_per_testcase(Testcase = stream_queues_have_consumers_field, Config) ->
+    case rabbit_ct_helpers:is_mixed_versions() of
+        true ->
+            {skip, "mixed version clusters are not supported"};
+        _ ->
+            rabbit_ct_helpers:testcase_started(Config, Testcase)
+    end;
 init_per_testcase(Testcase, Config) ->
     rabbit_ct_broker_helpers:close_all_connections(Config, 0, <<"rabbit_mgmt_SUITE:init_per_testcase">>),
     rabbit_ct_helpers:testcase_started(Config, Testcase).
@@ -1099,6 +1106,26 @@ quorum_queues_test_loop(Config, N) ->
     close_connection(Conn),
     quorum_queues_test_loop(Config, N-1).
 
+stream_queues_have_consumers_field(Config) ->
+    Good = [{durable, true}, {arguments, [{'x-queue-type', 'stream'}]}],
+    http_get(Config, "/queues/%2f/sq", ?NOT_FOUND),
+    http_put(Config, "/queues/%2f/sq", Good, {group, '2xx'}),
+
+    wait_until(fun() ->
+                       Qs = http_get(Config, "/queues/%2F"),
+                       length(Qs) == 1 andalso maps:is_key(consumers, lists:nth(1, Qs))
+               end, 50),
+
+    Queues = http_get(Config, "/queues/%2F"),
+    assert_list([#{name        => <<"sq">>,
+                   arguments => #{'x-queue-type' => <<"stream">>},
+                   consumers   => 0}],
+                Queues),
+
+
+    http_delete(Config, "/queues/%2f/sq", {group, '2xx'}),
+    ok.
+
 queues_well_formed_json_test(Config) ->
     %% TODO This test should be extended to the whole API
     Good = [{durable, true}],