WIP global counters

TODOs:
* rabbitmq_global_publishers
* rabbitmq_global_consumers
* tests

Author: ansd

deps/rabbit/src/rabbit_global_counters.erl

@@ -131,13 +131,15 @@
             ]).
 
 boot_step() ->
-    %% Protocol counters
-    init([{protocol, amqp091}]),
-
-    %% Protocol & Queue Type counters
-    init([{protocol, amqp091}, {queue_type, rabbit_classic_queue}]),
-    init([{protocol, amqp091}, {queue_type, rabbit_quorum_queue}]),
-    init([{protocol, amqp091}, {queue_type, rabbit_stream_queue}]),
+    [begin
+         %% Protocol counters
+         init([{protocol, Proto}]),
+
+         %% Protocol & Queue Type counters
+         init([{protocol, Proto}, {queue_type, rabbit_classic_queue}]),
+         init([{protocol, Proto}, {queue_type, rabbit_quorum_queue}]),
+         init([{protocol, Proto}, {queue_type, rabbit_stream_queue}])
+     end || Proto <- [amqp091, amqp10]],
 
     %% Dead Letter counters
     %%

deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_session.erl

@@ -13,6 +13,7 @@
 -include_lib("amqp10_common/include/amqp10_types.hrl").
 -include("rabbit_amqp1_0.hrl").
 
+-define(PROTOCOL, amqp10).
 -define(HIBERNATE_AFTER, 6_000).
 -define(CREDIT_REPLY_TIMEOUT, 30_000).
 -define(UINT_OUTGOING_WINDOW, {uint, ?UINT_MAX}).
@@ -89,7 +90,8 @@
 -record(outgoing_link, {
           %% Although the source address of a link might be an exchange name and binding key
           %% or a topic filter, an outgoing link will always consume from a queue.
-          queue :: rabbit_misc:resource_name(),
+          queue_name :: rabbit_misc:resource_name(),
+          queue_type :: rabbit_queue_type:queue_type(),
           send_settled :: boolean(),
           max_message_size :: unlimited | pos_integer(),
           %% When credit API v1 is used, our session process holds the delivery-count
@@ -435,6 +437,7 @@ send_delivery_state_changes(State0 = #state{cfg = #cfg{writer_pid = Writer,
     send_dispositions(ReleasedIds, #'v1_0.released'{}, Writer, ChannelNum),
     AcceptedIds = AcceptedIds1 ++ AcceptedIds0,
     send_dispositions(AcceptedIds, #'v1_0.accepted'{}, Writer, ChannelNum),
+    rabbit_global_counters:messages_confirmed(?PROTOCOL, length(AcceptedIds)),
     %% Send DETACH frames after DISPOSITION frames such that
     %% clients can handle DISPOSITIONs before closing their links.
     lists:foreach(fun(Frame) ->
@@ -624,7 +627,7 @@ destroy_incoming_link(Handle, #incoming_link{queue = QNameBin}, QNameBin, {Frame
 destroy_incoming_link(_, _, _, Acc) ->
     Acc.
 
-destroy_outgoing_link(Handle, #outgoing_link{queue = QNameBin}, QNameBin, {Frames, Unsettled0, Links}) ->
+destroy_outgoing_link(Handle, #outgoing_link{queue_name = QNameBin}, QNameBin, {Frames, Unsettled0, Links}) ->
     {Unsettled, _RemovedMsgIds} = remove_link_from_outgoing_unsettled_map(Handle, Unsettled0),
     {[detach(Handle, ?V_1_0_AMQP_ERROR_RESOURCE_DELETED) | Frames],
      Unsettled,
@@ -744,6 +747,7 @@ handle_control(#'v1_0.attach'{role = ?RECV_ROLE,
             case rabbit_amqqueue:with(
                    QName,
                    fun(Q) ->
+                           QType = amqqueue:get_type(Q),
                            %% Whether credit API v1 or v2 is used is decided only here at link attachment time.
                            %% This decision applies to the whole life time of the link.
                            %% This means even when feature flag credit_api_v2 will be enabled later, this consumer will
@@ -756,7 +760,7 @@ handle_control(#'v1_0.attach'{role = ?RECV_ROLE,
                            %% flow control state. Hence, credit API mixed version isn't an issue for streams.
                            {Mode,
                             DeliveryCount} = case rabbit_feature_flags:is_enabled(credit_api_v2) orelse
-                                                  amqqueue:get_type(Q) =:= rabbit_stream_queue of
+                                                  QType =:= rabbit_stream_queue of
                                                  true ->
                                                      {{credited, ?INITIAL_DELIVERY_COUNT}, credit_api_v2};
                                                  false ->
@@ -799,7 +803,8 @@ handle_control(#'v1_0.attach'{role = ?RECV_ROLE,
                                                             %% maximum size imposed by the link endpoint."
                                                             unlimited
                                                     end,
-                                   Link = #outgoing_link{queue = QNameBin,
+                                   Link = #outgoing_link{queue_name = QNameBin,
+                                                         queue_type = QType,
                                                          send_settled = SndSettled,
                                                          max_message_size = MaxMessageSize,
                                                          delivery_count = DeliveryCount},
@@ -914,7 +919,7 @@ handle_control(#'v1_0.detach'{handle = Handle = ?UINT(HandleInt),
     %% TODO keep the state around depending on the lifetime
     {QStates, Unsettled, OutgoingLinks}
     = case maps:take(HandleInt, OutgoingLinks0) of
-          {#outgoing_link{queue = QNameBin}, OutgoingLinks1} ->
+          {#outgoing_link{queue_name = QNameBin}, OutgoingLinks1} ->
               QName = rabbit_misc:r(Vhost, queue, QNameBin),
               case rabbit_amqqueue:lookup(QName) of
                   {ok, Q} ->
@@ -1041,6 +1046,7 @@ handle_control(#'v1_0.disposition'{role = ?RECV_ROLE,
               fun({QName, Ctag}, MsgIds, {QS0, ActionsAcc}) ->
                       case rabbit_queue_type_settle(QName, SettleOp, Ctag, MsgIds, QS0) of
                           {ok, QS, Actions0} ->
+                              messages_acknowledged(SettleOp, QName, QS, MsgIds),
                               {QS, ActionsAcc ++ Actions0};
                           {protocol_error, _ErrorType, Reason, ReasonArgs} ->
                               protocol_error(?V_1_0_AMQP_ERROR_INTERNAL_ERROR,
@@ -1377,7 +1383,8 @@ handle_deliver(ConsumerTag, AckRequired,
                               cfg = #cfg{outgoing_max_frame_size = MaxFrameSize}}) ->
     Handle = ctag_to_handle(ConsumerTag),
     case OutgoingLinks of
-        #{Handle := #outgoing_link{send_settled = SendSettled,
+        #{Handle := #outgoing_link{queue_type = QType,
+                                   send_settled = SendSettled,
                                    max_message_size = MaxMessageSize}} ->
             %% "The delivery-tag MUST be unique amongst all deliveries that could be
             %% considered unsettled by either end of the link." [2.6.12]
@@ -1418,6 +1425,7 @@ handle_deliver(ConsumerTag, AckRequired,
                              TLen = iolist_size(amqp10_framing:encode_bin(Transfer)),
                              encode_frames(Transfer, Sections, MaxFrameSize - TLen, [])
                      end,
+            messages_delivered(Redelivered, QType),
             Del = #outgoing_unsettled{
                      msg_id = MsgId,
                      consumer_tag = ConsumerTag,
@@ -1533,7 +1541,7 @@ incoming_link_transfer(
     RoutingKeys = mc:get_annotation(routing_keys, Mc),
     RoutingKey = routing_key(RoutingKeys, XName),
     % Mc1 = rabbit_message_interceptor:intercept(Mc),
-    % rabbit_global_counters:messages_received(ProtoVer, 1),
+    messages_received(Settled),
     case rabbit_exchange:lookup(XName) of
         {ok, Exchange} ->
             check_write_permitted_on_topic(Exchange, User, RoutingKey),
@@ -1551,7 +1559,6 @@ incoming_link_transfer(
             case rabbit_queue_type:deliver(Qs, Mc, Opts, QStates0) of
                 {ok, QStates, Actions} ->
                     State1 = State0#state{queue_states = QStates},
-                    % rabbit_global_counters:messages_routed(ProtoVer, length(Qs)),
                     %% Confirms must be registered before processing actions
                     %% because actions may contain rejections of publishes.
                     {U, Reply0} = process_routing_confirm(
@@ -1582,16 +1589,18 @@ incoming_link_transfer(
     end.
 
 process_routing_confirm([], _SenderSettles = true, _, U) ->
-    % rabbit_global_counters:messages_unroutable_dropped(ProtoVer, 1),
+    rabbit_global_counters:messages_unroutable_dropped(?PROTOCOL, 1),
     {U, []};
 process_routing_confirm([], _SenderSettles = false, DeliveryId, U) ->
-    % rabbit_global_counters:messages_unroutable_returned(ProtoVer, 1),
+    rabbit_global_counters:messages_unroutable_returned(?PROTOCOL, 1),
     Disposition = released(DeliveryId),
     {U, [Disposition]};
 process_routing_confirm([_|_] = Qs, SenderSettles, DeliveryId, U0) ->
     QNames = rabbit_amqqueue:queue_names(Qs),
     false = maps:is_key(DeliveryId, U0),
-    U = U0#{DeliveryId => {maps:from_keys(QNames, ok), SenderSettles, false}},
+    Map = maps:from_keys(QNames, ok),
+    U = U0#{DeliveryId => {Map, SenderSettles, false}},
+    rabbit_global_counters:messages_routed(?PROTOCOL, map_size(Map)),
     {U, []}.
 
 released(DeliveryId) ->
@@ -1656,7 +1665,7 @@ ensure_target(#'v1_0.target'{address = Address,
     end.
 
 handle_outgoing_link_flow_control(
-  #outgoing_link{queue = QNameBin,
+  #outgoing_link{queue_name = QNameBin,
                  delivery_count = MaybeDeliveryCountSnd},
   #'v1_0.flow'{handle = ?UINT(HandleInt),
                delivery_count = MaybeDeliveryCountRcv,
@@ -2065,6 +2074,30 @@ routing_key(undefined, XName) ->
 routing_key([RoutingKey], _XName) ->
     RoutingKey.
 
+messages_received(Settled) ->
+    rabbit_global_counters:messages_received(?PROTOCOL, 1),
+    case Settled of
+        true -> ok;
+        false -> rabbit_global_counters:messages_received_confirm(?PROTOCOL, 1)
+    end.
+
+messages_delivered(Redelivered, QueueType) ->
+    rabbit_global_counters:messages_delivered(?PROTOCOL, QueueType, 1),
+    case Redelivered of
+        true -> rabbit_global_counters:messages_redelivered(?PROTOCOL, QueueType, 1);
+        false -> ok
+    end.
+
+messages_acknowledged(complete, QName, QS, MsgIds) ->
+    case rabbit_queue_type:module(QName, QS) of
+        {ok, QType} ->
+            rabbit_global_counters:messages_acknowledged(?PROTOCOL, QType, length(MsgIds));
+        _ ->
+            ok
+    end;
+messages_acknowledged(_, _, _, _) ->
+    ok.
+
 check_internal_exchange(#exchange{internal = true,
                                   name = XName}) ->
     protocol_error(?V_1_0_AMQP_ERROR_UNAUTHORIZED_ACCESS,