Restart global counters with our fresh understanding

This has gone through a few cycles, with @mkuratczyk & @dcorbacho
covering most of the ground. @dcorbacho had most of this in
#3045, but the main
branch went through a few changes in the meantime. Rather than resolving
all the conflicts, and then making the necessary changes, we (@gerhard +
@kjnilsson) took all learnings and started re-applying a lot of the
existing code from #3045. We are confident in this approach and would
like to see it through.

Expose global metrics in rabbitmq_prometheus. We don't want to keep
modifying the existing collector, but start with a new namespace,
rabbitmq_global_, and continue building on top of it. The idea is to
build in parallel, and slowly transition to the new metrics, because
semantically the changes are too big since streams, and we have been
discussing protocol-specific metrics with @kjnilsson, which makes me
think that this approach is least disruptive and... simple.

While at this, we removed redundant empty return value handling in the
channel. The function called no longer returns this.

Also removed all DONE / TODO & other comments - we'll handle them when
the time comes, no need to leave TODO reminders.

Pair @kjnilsson

Signed-off-by: Gerhard Lazu <gerhard@lazu.co.uk>

Author: gerhard

deps/rabbit/BUILD.bazel

@@ -154,6 +154,7 @@ RUNTIME_DEPS = [
     "@observer_cli//:bazel_erlang_lib",
     "@osiris//:bazel_erlang_lib",
     "@recon//:bazel_erlang_lib",
+    "@seshat//:bazel_erlang_lib",
     "@sysmon_handler//:bazel_erlang_lib",
     "@systemd//:bazel_erlang_lib",
 ]

deps/rabbit/Makefile

@@ -136,14 +136,15 @@ APPS_DIR := $(CURDIR)/apps
 LOCAL_DEPS = sasl rabbitmq_prelaunch os_mon inets compiler public_key crypto ssl syntax_tools xmerl
 
 BUILD_DEPS = rabbitmq_cli
-DEPS = ranch rabbit_common ra sysmon_handler stdout_formatter recon observer_cli osiris amqp10_common syslog systemd
+DEPS = ranch rabbit_common ra sysmon_handler stdout_formatter recon observer_cli osiris amqp10_common syslog systemd seshat
 TEST_DEPS = rabbitmq_ct_helpers rabbitmq_ct_client_helpers amqp_client meck proper
 
 PLT_APPS += mnesia
 
 dep_syslog = git https://github.com/schlagert/syslog 4.0.0
 dep_osiris = git https://github.com/rabbitmq/osiris master
 dep_systemd = hex 0.6.1
+dep_seshat = git https://github.com/rabbitmq/seshat main
 
 define usage_xml_to_erl
 $(subst __,_,$(patsubst $(DOCS_DIR)/rabbitmq%.1.xml, src/rabbit_%_usage.erl, $(subst -,_,$(1))))

deps/rabbit/src/rabbit.erl

@@ -127,6 +127,13 @@
                     {requires,    pre_boot},
                     {enables,     external_infrastructure}]}).
 
+-rabbit_boot_step({rabbit_global_counters,
+                   [{description, "global metrics storage"},
+                    {mfa,         {rabbit_global_counters, init,
+                                   []}},
+                    {requires,    pre_boot},
+                    {enables,     external_infrastructure}]}).
+
 -rabbit_boot_step({rabbit_event,
                    [{description, "statistics event manager"},
                     {mfa,         {rabbit_sup, start_restartable_child,

deps/rabbit/src/rabbit_channel.erl

@@ -1285,6 +1285,7 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
                                    confirm_enabled  = ConfirmEnabled,
                                    delivery_flow    = Flow
                                    }) ->
+    rabbit_global_counters:messages_received(all, 1),
     check_msg_size(Content, MaxMessageSize, GCThreshold),
     ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
     check_write_permitted(ExchangeName, User, AuthzContext),
@@ -1302,7 +1303,8 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
     {MsgSeqNo, State1} =
         case DoConfirm orelse Mandatory of
             false -> {undefined, State};
-            true  -> SeqNo = State#ch.publish_seqno,
+            true  -> rabbit_global_counters:messages_received_confirm(all, 1),
+                     SeqNo = State#ch.publish_seqno,
                      {SeqNo, State#ch{publish_seqno = SeqNo + 1}}
         end,
     case rabbit_basic:message(ExchangeName, RoutingKey, DecodedContent) of
@@ -1314,9 +1316,11 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
                                 Username, TraceState),
             DQ = {Delivery#delivery{flow = Flow}, QNames},
             {noreply, case Tx of
-                          none         -> deliver_to_queues(DQ, State1);
-                          {Msgs, Acks} -> Msgs1 = ?QUEUE:in(DQ, Msgs),
-                                          State1#ch{tx = {Msgs1, Acks}}
+                          none ->
+                              deliver_to_queues(DQ, State1);
+                          {Msgs, Acks} ->
+                              Msgs1 = ?QUEUE:in(DQ, Msgs),
+                              State1#ch{tx = {Msgs1, Acks}}
                       end};
         {error, Reason} ->
             precondition_failed("invalid message: ~p", [Reason])
@@ -1363,11 +1367,9 @@ handle_method(#'basic.get'{queue = QueueNameBin, no_ack = NoAck},
             handle_basic_get(WriterPid, DeliveryTag, NoAck, MessageCount, Msg,
                              State#ch{queue_states = QueueStates});
         {empty, QueueStates} ->
+            rabbit_global_counters:messages_get_empty(all, 1),
             ?INCR_STATS(queue_stats, QueueName, 1, get_empty, State),
             {reply, #'basic.get_empty'{}, State#ch{queue_states = QueueStates}};
-        empty ->
-            ?INCR_STATS(queue_stats, QueueName, 1, get_empty, State),
-            {reply, #'basic.get_empty'{}, State};
         {error, {unsupported, single_active_consumer}} ->
             rabbit_misc:protocol_error(
               resource_locked,
@@ -1692,9 +1694,9 @@ handle_method(#'tx.select'{}, _, State) ->
 handle_method(#'tx.commit'{}, _, #ch{tx = none}) ->
     precondition_failed("channel is not transactional");
 
-handle_method(#'tx.commit'{}, _, State = #ch{tx      = {Msgs, Acks},
+handle_method(#'tx.commit'{}, _, State = #ch{tx      = {Deliveries, Acks},
                                              limiter = Limiter}) ->
-    State1 = queue_fold(fun deliver_to_queues/2, State, Msgs),
+    State1 = queue_fold(fun deliver_to_queues/2, State, Deliveries),
     Rev = fun (X) -> lists:reverse(lists:sort(X)) end,
     {State2, Actions2} =
         lists:foldl(fun ({ack,     A}, {Acc, Actions}) ->
@@ -1964,15 +1966,28 @@ record_sent(Type, Tag, AckRequired,
                         unacked_message_q = UAMQ,
                         next_tag          = DeliveryTag
                        }) ->
-    ?INCR_STATS(queue_stats, QName, 1, case {Type, AckRequired} of
-                                           {get, true}      -> get;
-                                           {get, false}     -> get_no_ack;
-                                           {deliver, true}  -> deliver;
-                                           {deliver, false} -> deliver_no_ack
-                                       end, State),
+    rabbit_global_counters:messages_delivered(all, 1),
+    ?INCR_STATS(queue_stats, QName, 1,
+                case {Type, AckRequired} of
+                    {get, true} ->
+                        rabbit_global_counters:messages_delivered_get_manual_ack(all, 1),
+                        get;
+                    {get, false} ->
+                        rabbit_global_counters:messages_delivered_get_auto_ack(all, 1),
+                        get_no_ack;
+                    {deliver, true} ->
+                        rabbit_global_counters:messages_delivered_consume_manual_ack(all, 1),
+                        deliver;
+                    {deliver, false} ->
+                        rabbit_global_counters:messages_delivered_consume_auto_ack(all, 1),
+                        deliver_no_ack
+                end, State),
     case Redelivered of
-        true  -> ?INCR_STATS(queue_stats, QName, 1, redeliver, State);
-        false -> ok
+        true ->
+            rabbit_global_counters:messages_redelivered(all, 1),
+            ?INCR_STATS(queue_stats, QName, 1, redeliver, State);
+        false ->
+            ok
     end,
     DeliveredAt = os:system_time(millisecond),
     rabbit_trace:tap_out(Msg, ConnName, ChannelNum, Username, TraceState),
@@ -2036,6 +2051,7 @@ ack(Acked, State = #ch{queue_states = QueueStates0}) ->
 
 incr_queue_stats(QName, MsgIds, State) ->
     Count = length(MsgIds),
+    rabbit_global_counters:messages_acknowledged(all, Count),
     ?INCR_STATS(queue_stats, QName, Count, ack, State).
 
 %% {Msgs, Acks}
@@ -2110,6 +2126,7 @@ deliver_to_queues({#delivery{message   = #basic_message{exchange_name = XName},
                              mandatory = false},
                    _RoutedToQueueNames = []}, State) -> %% optimisation
     ?INCR_STATS(exchange_stats, XName, 1, publish, State),
+    rabbit_global_counters:messages_unroutable_dropped(all, 1),
     ?INCR_STATS(exchange_stats, XName, 1, drop_unroutable, State),
     State;
 deliver_to_queues({Delivery = #delivery{message    = Message = #basic_message{exchange_name = XName},
@@ -2128,6 +2145,7 @@ deliver_to_queues({Delivery = #delivery{message    = Message = #basic_message{ex
     Qs = rabbit_amqqueue:lookup(AllNames),
     case rabbit_queue_type:deliver(Qs, Delivery, QueueStates0) of
         {ok, QueueStates, Actions}  ->
+            rabbit_global_counters:messages_routed(all, length(Qs)),
             %% NB: the order here is important since basic.returns must be
             %% sent before confirms.
             ok = process_routing_mandatory(Mandatory, Qs, Message, State0),
@@ -2164,6 +2182,7 @@ deliver_to_queues({Delivery = #delivery{message    = Message = #basic_message{ex
          end,
     case rabbit_queue_type:deliver(Qs, Delivery, QueueStates0) of
         {ok, QueueStates, Actions}  ->
+            rabbit_global_counters:messages_routed(all, length(Qs)),
             %% NB: the order here is important since basic.returns must be
             %% sent before confirms.
             ok = process_routing_mandatory(Mandatory, Qs, Message, State0),
@@ -2213,11 +2232,13 @@ infer_extra_bcc(Qs) ->
 process_routing_mandatory(_Mandatory = true,
                           _RoutedToQs = [],
                           Msg, State) ->
+    rabbit_global_counters:messages_unroutable_returned(all, 1),
     ok = basic_return(Msg, State, no_route),
     ok;
 process_routing_mandatory(_Mandatory = false,
                           _RoutedToQs = [],
                           #basic_message{exchange_name = ExchangeName}, State) ->
+    rabbit_global_counters:messages_unroutable_dropped(all, 1),
     ?INCR_STATS(exchange_stats, ExchangeName, 1, drop_unroutable, State),
     ok;
 process_routing_mandatory(_, _, _, _) ->
@@ -2245,6 +2266,7 @@ send_confirms_and_nacks(State = #ch{tx = none, confirmed = C, rejected = R}) ->
     case rabbit_node_monitor:pause_partition_guard() of
         ok      ->
             Confirms = lists:append(C),
+            rabbit_global_counters:messages_confirmed(all, length(Confirms)),
             Rejects = lists:append(R),
             ConfirmMsgSeqNos =
                 lists:foldl(
@@ -2783,10 +2805,10 @@ get_operation_timeout_and_deadline() ->
     Deadline =  now_millis() + Timeout,
     {Timeout, Deadline}.
 
-queue_fold(Fun, Init, Q) ->
-    case ?QUEUE:out(Q) of
-        {empty, _Q}      -> Init;
-        {{value, V}, Q1} -> queue_fold(Fun, Fun(V, Init), Q1)
+queue_fold(Fun, Acc, Queue) ->
+    case ?QUEUE:out(Queue) of
+        {empty, _Queue}      -> Acc;
+        {{value, Item}, Queue1} -> queue_fold(Fun, Fun(Item, Acc), Queue1)
     end.
 
 evaluate_consumer_timeout(State0 = #ch{cfg = #conf{channel = Channel,

deps/rabbit/src/rabbit_global_counters.erl

@@ -0,0 +1,164 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2021 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbit_global_counters).
+-on_load(init/0).
+
+-export([
+         init/0,
+         overview/0,
+         prometheus_format/0,
+         messages_received/2,
+         messages_received_confirm/2,
+         messages_unroutable_dropped/2,
+         messages_unroutable_returned/2,
+         messages_routed/2,
+         messages_confirmed/2,
+         messages_delivered/2,
+         messages_delivered_consume_manual_ack/2,
+         messages_delivered_consume_auto_ack/2,
+         messages_delivered_get_manual_ack/2,
+         messages_delivered_get_auto_ack/2,
+         messages_get_empty/2,
+         messages_redelivered/2,
+         messages_acknowledged/2
+       ]).
+
+-define(MESSAGES_RECEIVED, 1).
+-define(MESSAGES_RECEIVED_CONFIRM, 2).
+-define(MESSAGES_UNROUTABLE_DROPPED, 3).
+-define(MESSAGES_UNROUTABLE_RETURNED, 4).
+-define(MESSAGES_ROUTED, 5).
+-define(MESSAGES_CONFIRMED, 6).
+-define(MESSAGES_DELIVERED, 7).
+-define(MESSAGES_DELIVERED_CONSUME_MANUAL_ACK, 8).
+-define(MESSAGES_DELIVERED_CONSUME_AUTO_ACK, 9).
+-define(MESSAGES_DELIVERED_GET_MANUAL_ACK, 10).
+-define(MESSAGES_DELIVERED_GET_AUTO_ACK, 11).
+-define(MESSAGES_GET_EMPTY, 12).
+-define(MESSAGES_REDELIVERED, 13).
+-define(MESSAGES_ACKNOWLEDGED, 14).
+
+-define(COUNTERS,
+            [
+                {
+                    messages_received_total, ?MESSAGES_RECEIVED, counter,
+                    "Total number of messages received from publishers"
+                },
+                {
+                    messages_received_confirm_total, ?MESSAGES_RECEIVED_CONFIRM, counter,
+                    "Total number of messages received from publishers expecting confirmations"
+                },
+                {
+                    messages_unroutable_dropped_total, ?MESSAGES_UNROUTABLE_DROPPED, counter,
+                    "Total number of messages published as non-mandatory into an exchange and dropped as unroutable"
+                },
+                {
+                    messages_unroutable_returned_total, ?MESSAGES_UNROUTABLE_RETURNED, counter,
+                   "Total number of messages published as mandatory into an exchange and returned to the publisher as unroutable"
+                },
+                {
+                    messages_routed_total, ?MESSAGES_ROUTED, counter,
+                    "Total number of messages routed to queues or streams"
+                },
+                {
+                    messages_confirmed_total, ?MESSAGES_CONFIRMED, counter,
+                    "Total number of messages confirmed to publishers"
+                },
+                {
+                    messages_delivered_total, ?MESSAGES_DELIVERED, counter,
+                    "Total number of messages delivered to consumers"
+                },
+                {
+                    messages_delivered_consume_manual_ack_total, ?MESSAGES_DELIVERED_CONSUME_MANUAL_ACK, counter,
+                    "Total number of messages delivered to consumers using basic.consume with manual acknowledgment"
+                },
+                {
+                    messages_delivered_consume_auto_ack_total, ?MESSAGES_DELIVERED_CONSUME_AUTO_ACK, counter,
+                    "Total number of messages delivered to consumers using basic.consume with automatic acknowledgment"
+                },
+                {
+                    messages_delivered_get_manual_ack_total, ?MESSAGES_DELIVERED_GET_MANUAL_ACK, counter,
+                    "Total number of messages delivered to consumers using basic.get with manual acknowledgment"
+                },
+                {
+                    messages_delivered_get_auto_ack_total, ?MESSAGES_DELIVERED_GET_AUTO_ACK, counter,
+                    "Total number of messages delivered to consumers using basic.get with automatic acknowledgment"
+                },
+                {
+                    messages_get_empty_total, ?MESSAGES_GET_EMPTY, counter,
+                    "Total number of times basic.get operations fetched no message"
+                },
+                {
+                    messages_redelivered_total, ?MESSAGES_REDELIVERED, counter,
+                    "Total number of messages redelivered to consumers"
+                },
+                {
+                    messages_acknowledged_total, ?MESSAGES_ACKNOWLEDGED, counter,
+                    "Total number of messages acknowledged by consumers"
+                }
+            ]).
+
+init() ->
+    _ = seshat_counters:new_group(global),
+    CategoryCounters = seshat_counters:new(global, all, ?COUNTERS),
+    persistent_term:put({?MODULE, all}, CategoryCounters),
+    ok.
+
+overview() ->
+    seshat_counters:overview(global).
+
+prometheus_format() ->
+    seshat_counters:prometheus_format(global).
+
+messages_received(Category, Num) ->
+    counters:add(fetch(Category), ?MESSAGES_RECEIVED, Num).
+
+messages_received_confirm(Category, Num) ->
+    counters:add(fetch(Category), ?MESSAGES_RECEIVED_CONFIRM, Num).
+
+messages_unroutable_dropped(Category, Num) ->
+    counters:add(fetch(Category), ?MESSAGES_UNROUTABLE_DROPPED, Num).
+
+messages_unroutable_returned(Category, Num) ->
+    counters:add(fetch(Category), ?MESSAGES_UNROUTABLE_RETURNED, Num).
+
+% Formerly known as queue_messages_published_total
+% This is a more accurate representation of what happens with messages that
+%   arrive on a channel or connection in the case of streams.
+messages_routed(Category, Num) ->
+    counters:add(fetch(Category), ?MESSAGES_ROUTED, Num).
+
+messages_confirmed(Category, Num) ->
+    counters:add(fetch(Category), ?MESSAGES_CONFIRMED, Num).
+
+messages_delivered(Category, Num) ->
+    counters:add(fetch(Category), ?MESSAGES_DELIVERED, Num).
+
+messages_delivered_consume_manual_ack(Category, Num) ->
+    counters:add(fetch(Category), ?MESSAGES_DELIVERED_CONSUME_MANUAL_ACK, Num).
+
+messages_delivered_consume_auto_ack(Category, Num) ->
+    counters:add(fetch(Category), ?MESSAGES_DELIVERED_CONSUME_AUTO_ACK, Num).
+
+messages_delivered_get_manual_ack(Category, Num) ->
+    counters:add(fetch(Category), ?MESSAGES_DELIVERED_GET_MANUAL_ACK, Num).
+
+messages_delivered_get_auto_ack(Category, Num) ->
+    counters:add(fetch(Category), ?MESSAGES_DELIVERED_GET_AUTO_ACK, Num).
+
+messages_get_empty(Category, Num) ->
+    counters:add(fetch(Category), ?MESSAGES_GET_EMPTY, Num).
+
+messages_redelivered(Category, Num) ->
+    counters:add(fetch(Category), ?MESSAGES_REDELIVERED, Num).
+
+messages_acknowledged(Category, Num) ->
+    counters:add(fetch(Category), ?MESSAGES_ACKNOWLEDGED, Num).
+
+fetch(Category) ->
+    persistent_term:get({?MODULE, Category}).