Use number of publishing channels as global publishers in amqp091

Author: dcorbacho

deps/rabbit/src/rabbit_channel.erl

@@ -165,7 +165,8 @@
              delivery_flow,
              interceptor_state,
              queue_states,
-             tick_timer
+             tick_timer,
+             publishing_mode = false :: boolean()
             }).
 
 -define(QUEUE, lqueue).
@@ -878,6 +879,7 @@ terminate(_Reason,
     rabbit_event:if_enabled(State, #ch.stats_timer,
                             fun() -> emit_stats(State) end),
     [delete_stats(Tag) || {Tag, _} <- get()],
+    maybe_decrease_global_publishers(State),
     rabbit_core_metrics:channel_closed(self()),
     rabbit_event:notify(channel_closed, [{pid, self()},
                                          {user_who_performed_action, Username}]).
@@ -1285,6 +1287,7 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
                                    confirm_enabled  = ConfirmEnabled,
                                    delivery_flow    = Flow
                                    }) ->
+    State0 = maybe_increase_global_publishers(State),
     rabbit_global_counters:messages_received(amqp091, 1),
     check_msg_size(Content, MaxMessageSize, GCThreshold),
     ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
@@ -1302,10 +1305,10 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
     DoConfirm = Tx =/= none orelse ConfirmEnabled,
     {MsgSeqNo, State1} =
         case DoConfirm orelse Mandatory of
-            false -> {undefined, State};
+            false -> {undefined, State0};
             true  -> rabbit_global_counters:messages_received_confirm(amqp091, 1),
-                     SeqNo = State#ch.publish_seqno,
-                     {SeqNo, State#ch{publish_seqno = SeqNo + 1}}
+                     SeqNo = State0#ch.publish_seqno,
+                     {SeqNo, State0#ch{publish_seqno = SeqNo + 1}}
         end,
     case rabbit_basic:message(ExchangeName, RoutingKey, DecodedContent) of
         {ok, Message} ->
@@ -2917,3 +2920,14 @@ find_queue_name_from_quorum_name(Name, QStates) ->
                   end
           end,
     rabbit_queue_type:fold_state(Fun, undefined, QStates).
+
+maybe_increase_global_publishers(#ch{publishing_mode = true} = State0) ->
+    State0;
+maybe_increase_global_publishers(State0) ->
+    rabbit_global_counters:publisher_created(amqp091),
+    State0#ch{publishing_mode = true}.
+
+maybe_decrease_global_publishers(#ch{publishing_mode = true}) ->
+    ok;
+maybe_decrease_global_publishers(#ch{publishing_mode = false}) ->
+    rabbit_global_counters:publisher_deleted(amqp091).

deps/rabbit/test/queue_type_SUITE.erl

@@ -164,14 +164,18 @@ smoke(Config) ->
     basic_ack(Ch, basic_get(Ch, QName)),
     %% global counters
     publish_and_confirm(Ch, <<"inexistent_queue">>, <<"msg4">>),
+    ConsumerTag3 = <<"ctag3">>,
+    ok = subscribe(Ch, QName, ConsumerTag3),
     ProtocolCounters = maps:get([{protocol, amqp091}], get_global_counters(Config)),
     ?assertEqual(#{
                    messages_confirmed_total => 4,
                    messages_received_confirm_total => 4,
                    messages_received_total => 4,
                    messages_routed_total => 3,
                    messages_unroutable_dropped_total => 1,
-                   messages_unroutable_returned_total => 0
+                   messages_unroutable_returned_total => 0,
+                   consumers => 1,
+                   publishers => 1
                   }, ProtocolCounters),
     QueueType = list_to_atom(
                   "rabbit_" ++

deps/rabbitmq_prometheus/metrics.md

@@ -64,6 +64,8 @@ To generate these:
 | rabbitmq_global_messages_routed_total                       | Total number of messages routed to queues or streams                                                         |
 | rabbitmq_global_messages_unroutable_dropped_total           | Total number of messages published as non-mandatory into an exchange and dropped as unroutable               |
 | rabbitmq_global_messages_unroutable_returned_total          | Total number of messages published as mandatory into an exchange and returned to the publisher as unroutable |
+| rabbitmq_global_publishers                                  | Publishers currently connected |
+| rabbitmq_global_consumers                                   | Consumers currently connected |
 
 ### Generic
 

deps/rabbitmq_prometheus/test/rabbit_prometheus_http_SUITE.erl

@@ -306,7 +306,9 @@ global_metrics_present_test(Config) ->
     ?assertEqual(match, re:run(Body, "^rabbitmq_global_messages_delivered_get_auto_ack_total{", [{capture, none}, multiline])),
     ?assertEqual(match, re:run(Body, "^rabbitmq_global_messages_get_empty_total{", [{capture, none}, multiline])),
     ?assertEqual(match, re:run(Body, "^rabbitmq_global_messages_redelivered_total{", [{capture, none}, multiline])),
-    ?assertEqual(match, re:run(Body, "^rabbitmq_global_messages_acknowledged_total{", [{capture, none}, multiline])).
+    ?assertEqual(match, re:run(Body, "^rabbitmq_global_messages_acknowledged_total{", [{capture, none}, multiline])),
+    ?assertEqual(match, re:run(Body, "^rabbitmq_global_publishers{", [{capture, none}, multiline])),
+    ?assertEqual(match, re:run(Body, "^rabbitmq_global_consumers{", [{capture, none}, multiline])).
 
 global_metrics_single_metric_family_test(Config) ->
     {_Headers, Body} = http_get_with_pal(Config, [], 200),