Report connection/channel/queue churn

Rate metrics for monitoring of high connection churn scenarios
[#160943831]

3 files changed, 47 insertions, 10 deletions

diff --git a/deps/rabbitmq_management_agent/include/rabbit_mgmt_metrics.hrl b/deps/rabbitmq_management_agent/include/rabbit_mgmt_metrics.hrl
index e95111379a..0a6ed98b5f 100644
--- a/deps/rabbitmq_management_agent/include/rabbit_mgmt_metrics.hrl
+++ b/deps/rabbitmq_management_agent/include/rabbit_mgmt_metrics.hrl
@@ -53,7 +53,8 @@
                  {node_node_stats, set},
                  {node_node_coarse_stats, set},
                  {queue_msg_rates, set},
-                 {vhost_msg_rates, set}]).
+                 {vhost_msg_rates, set},
+                 {connection_churn_rates, set}]).
 
 -define(INDEX_TABLES, [consumer_stats_queue_index,
                        consumer_stats_channel_index,
@@ -140,7 +141,11 @@
 -define(queue_msg_rates(Disk_reads, Disk_writes), {Disk_reads, Disk_writes}).
 -define(vhost_msg_rates(Disk_reads, Disk_writes), {Disk_reads, Disk_writes}).
 -define(old_aggr_stats(Id, Stats), {Id, Stats}).
-
+-define(connection_churn_rates(Connection_created, Connection_closed, Channel_created,
+                               Channel_closed, Queue_declared, Queue_created,
+                               Queue_deleted),
+        {Connection_created, Connection_closed, Channel_created, Channel_closed,
+         Queue_declared, Queue_created, Queue_deleted}).
 
 -define(stats_per_table(Table),
         case Table of
@@ -186,6 +191,8 @@
                 [send_bytes, recv_bytes];
             T when T =:= queue_msg_rates;
                    T =:= vhost_msg_rates ->
-                [disk_reads, disk_writes]
+                [disk_reads, disk_writes];
+            T when T =:= connection_churn_rates ->
+                [connection_created, connection_closed, channel_created, channel_closed, queue_declared, queue_created, queue_deleted]
         end).
 %%------------------------------------------------------------------------------
diff --git a/deps/rabbitmq_management_agent/src/rabbit_mgmt_data.erl b/deps/rabbitmq_management_agent/src/rabbit_mgmt_data.erl
index 336e5892e1..78798e506f 100644
--- a/deps/rabbitmq_management_agent/src/rabbit_mgmt_data.erl
+++ b/deps/rabbitmq_management_agent/src/rabbit_mgmt_data.erl
@@ -119,14 +119,15 @@ node_data(Ranges, Id) ->
                         pick_range(coarse_node_stats, Ranges), Id),
        raw_message_data(node_persister_stats,
                         pick_range(coarse_node_stats, Ranges), Id),
-       {node_stats, lookup_element(node_stats, Id)}]).
+       {node_stats, lookup_element(node_stats, Id)}] ++
+      node_connection_churn_rates_data(Ranges)).
 
 overview_data(_Pid, User, Ranges, VHosts) ->
     Raw = [raw_all_message_data(vhost_msg_stats, pick_range(queue_msg_counts, Ranges), VHosts),
            raw_all_message_data(vhost_stats_fine_stats, pick_range(fine_stats, Ranges), VHosts),
            raw_all_message_data(vhost_msg_rates, pick_range(queue_msg_rates, Ranges), VHosts),
-           raw_all_message_data(vhost_stats_deliver_stats, pick_range(deliver_get, Ranges), VHosts)],
-
+           raw_all_message_data(vhost_stats_deliver_stats, pick_range(deliver_get, Ranges), VHosts),
+           raw_message_data(connection_churn_rates, pick_range(queue_msg_rates, Ranges), node())],
     maps:from_list(Raw ++
                    [{connections_count, count_created_stats(connection_created_stats, User)},
                     {channels_count, count_created_stats(channel_created_stats, User)},
@@ -173,6 +174,10 @@ node_raw_detail_stats_data(Ranges, Id) ->
                         pick_range(coarse_node_node_stats, Ranges), Key)
       || Key <- get_table_keys(node_node_coarse_stats, first(Id))].
 
+node_connection_churn_rates_data(Ranges) ->
+    [raw_message_data(connection_churn_rates,
+                      pick_range(churn_rates, Ranges), node())].
+
 exchange_raw_detail_stats_data(Ranges, Id) ->
      [raw_message_data2(channel_exchange_stats_fine_stats,
                         pick_range(fine_stats, Ranges), Key)
@@ -437,7 +442,8 @@ pick_range(K, {_RangeL, _RangeM, RangeD, _RangeN}) when K == coarse_conn_stats;
     RangeD;
 pick_range(K, {_RangeL, _RangeM, _RangeD, RangeN})
   when K == coarse_node_stats;
-       K == coarse_node_node_stats ->
+       K == coarse_node_node_stats;
+       K == churn_rates ->
     RangeN.
 
 first(Id)  ->
@@ -473,7 +479,9 @@ empty(Type, V) when Type =:= node_node_coarse_stats;
             Type =:= vhost_stats_coarse_conn_stats;
             Type =:= queue_msg_rates;
             Type =:= vhost_msg_rates ->
-    {V, V}.
+    {V, V};
+empty(connection_churn_rates, V) ->
+    {V, V, V, V, V, V, V}.
 
 retention_policy(connection_stats_coarse_conn_stats) ->
     basic;
@@ -518,6 +526,8 @@ retention_policy(node_coarse_stats) ->
 retention_policy(node_persister_stats) ->
     global;
 retention_policy(node_node_coarse_stats) ->
+    global;
+retention_policy(connection_churn_rates) ->
     global.
 
 format_resource(unknown) -> unknown;
diff --git a/deps/rabbitmq_management_agent/src/rabbit_mgmt_metrics_collector.erl b/deps/rabbitmq_management_agent/src/rabbit_mgmt_metrics_collector.erl
index 6255905362..0f89fe59f2 100644
--- a/deps/rabbitmq_management_agent/src/rabbit_mgmt_metrics_collector.erl
+++ b/deps/rabbitmq_management_agent/src/rabbit_mgmt_metrics_collector.erl
@@ -129,7 +129,8 @@ retention_policy(queue_coarse_metrics) -> basic;
 retention_policy(node_persister_metrics) -> global;
 retention_policy(node_coarse_metrics) -> global;
 retention_policy(node_metrics) -> basic;
-retention_policy(node_node_metrics) -> global.
+retention_policy(node_node_metrics) -> global;
+retention_policy(connection_churn_metrics) -> basic.
 
 take_smaller(Policies) ->
     Intervals = [I || {_, I} <- Policies],
@@ -516,7 +517,18 @@ aggregate_entry({Id, Metrics}, NextStats, Ops0,
                      Ops0),
     Ops = insert_entry_ops(node_node_coarse_stats, Id, false, Stats, Ops1,
                            GPolicies),
-    {NextStats, Ops, State}.
+    {NextStats, Ops, State};
+aggregate_entry({Id, ConnCreated, ConnClosed, ChCreated, ChClosed,
+                 QueueDeclared, QueueCreated, QueueDeleted}, NextStats, Ops0,
+                #state{table = connection_churn_metrics,
+                       policies = {_, _, GPolicies}} = State) ->
+    %% Id is the local node. There is only one entry on every ETS table.
+    Stats = ?connection_churn_rates(ConnCreated, ConnClosed, ChCreated, ChClosed,
+                                    QueueDeclared, QueueCreated, QueueDeleted),
+    Diff = get_difference(Id, Stats, State),
+    Ops = insert_entry_ops(connection_churn_rates, Id, true, Diff, Ops0,
+                           GPolicies),
+    {insert_old_aggr_stats(NextStats, Id, Stats), Ops, State}.
 
 insert_entry(Table, Id, TS, Entry, Size, Interval0, Incremental) ->
     Key = {Id, Interval0},
@@ -581,6 +593,10 @@ sum_entry({A0, A1}, {B0, B1}) ->
     {B0 + A0, B1 + A1};
 sum_entry({A0, A1, A2}, {B0, B1, B2}) ->
     {B0 + A0, B1 + A1, B2 + A2};
+sum_entry({A0, A1, A2, A3, A4, A5}, {B0, B1, B2, B3, B4, B5}) ->
+    {B0 + A0, B1 + A1, B2 + A2, B3 + A3, B4 + A4, B5 + A5};
+sum_entry({A0, A1, A2, A3, A4, A5, A6}, {B0, B1, B2, B3, B4, B5, B6}) ->
+    {B0 + A0, B1 + A1, B2 + A2, B3 + A3, B4 + A4, B5 + A5, B6 + A6};
 sum_entry({A0, A1, A2, A3, A4, A5, A6, A7}, {B0, B1, B2, B3, B4, B5, B6, B7}) ->
     {B0 + A0, B1 + A1, B2 + A2, B3 + A3, B4 + A4, B5 + A5, B6 + A6, B7 + A7}.
 
@@ -590,6 +606,10 @@ difference({A0, A1}, {B0, B1}) ->
     {B0 - A0, B1 - A1};
 difference({A0, A1, A2}, {B0, B1, B2}) ->
     {B0 - A0, B1 - A1, B2 - A2};
+difference({A0, A1, A2, A3, A4, A5}, {B0, B1, B2, B3, B4, B5}) ->
+    {B0 - A0, B1 - A1, B2 - A2, B3 - A3, B4 - A4, B5 - A5};
+difference({A0, A1, A2, A3, A4, A5, A6}, {B0, B1, B2, B3, B4, B5, B6}) ->
+    {B0 - A0, B1 - A1, B2 - A2, B3 - A3, B4 - A4, B5 - A5, B6 - A6};
 difference({A0, A1, A2, A3, A4, A5, A6, A7}, {B0, B1, B2, B3, B4, B5, B6, B7}) ->
     {B0 - A0, B1 - A1, B2 - A2, B3 - A3, B4 - A4, B5 - A5, B6 - A6, B7 - A7}.