Merge pull request #79 from rabbitmq/rabbitmq-server-1904

New channel metric: dropped unroutable messages

6 files changed, 88 insertions, 20 deletions

diff --git a/deps/rabbitmq_management_agent/include/rabbit_mgmt_metrics.hrl b/deps/rabbitmq_management_agent/include/rabbit_mgmt_metrics.hrl
index 9b100baed5..8a1bcbe2bb 100644
--- a/deps/rabbitmq_management_agent/include/rabbit_mgmt_metrics.hrl
+++ b/deps/rabbitmq_management_agent/include/rabbit_mgmt_metrics.hrl
@@ -84,16 +84,16 @@
 -define(channel_consumer_created_stats(Queue, ChPid, ConsumerTag),
         {Queue, {ChPid, ConsumerTag}}).
 -define(channel_stats(Id, Props), {Id, Props}).
--define(channel_stats_fine_stats(Publish, Confirm, Return_unroutable),
-        {Publish, Confirm, Return_unroutable}).
--define(channel_exchange_stats_fine_stats(Publish, Confirm, Return_unroutable),
-        {Publish, Confirm, Return_unroutable}).
+-define(channel_stats_fine_stats(Publish, Confirm, ReturnUnroutable, DropUnroutable),
+        {Publish, Confirm, ReturnUnroutable, DropUnroutable}).
+-define(channel_exchange_stats_fine_stats(Publish, Confirm, ReturnUnroutable, DropUnroutable),
+        {Publish, Confirm, ReturnUnroutable, DropUnroutable}).
 -define(channel_queue_stats_deliver_stats(Get, Get_no_ack, Deliver, Deliver_no_ack,
                                           Redeliver, Ack, Deliver_get, Get_empty),
         {Get, Get_no_ack, Deliver, Deliver_no_ack, Redeliver, Ack, Deliver_get,
          Get_empty}).
--define(vhost_stats_fine_stats(Publish, Confirm, Return_unroutable),
-        {Publish, Confirm, Return_unroutable}).
+-define(vhost_stats_fine_stats(Publish, Confirm, ReturnUnroutable, DropUnroutable),
+        {Publish, Confirm, ReturnUnroutable, DropUnroutable}).
 -define(queue_stats_deliver_stats(Get, Get_no_ack, Deliver, Deliver_no_ack,
                                   Redeliver, Ack, Deliver_get, Get_empty),
         {Get, Get_no_ack, Deliver, Deliver_no_ack, Redeliver, Ack, Deliver_get,
@@ -156,7 +156,7 @@
             T when T =:= channel_stats_fine_stats;
                    T =:= channel_exchange_stats_fine_stats;
                    T =:= vhost_stats_fine_stats ->
-                [publish, confirm, return_unroutable];
+                [publish, confirm, return_unroutable, drop_unroutable];
             T when T =:= channel_queue_stats_deliver_stats;
                    T =:= queue_stats_deliver_stats;
                    T =:= vhost_stats_deliver_stats;
diff --git a/deps/rabbitmq_management_agent/src/exometer_slide.erl b/deps/rabbitmq_management_agent/src/exometer_slide.erl
index 9bcc1c1436..5af252803c 100644
--- a/deps/rabbitmq_management_agent/src/exometer_slide.erl
+++ b/deps/rabbitmq_management_agent/src/exometer_slide.erl
@@ -214,6 +214,12 @@ add_to_total({A0, A1}, {B0, B1}) ->
     {B0 + A0, B1 + A1};
 add_to_total({A0, A1, A2}, {B0, B1, B2}) ->
     {B0 + A0, B1 + A1, B2 + A2};
+add_to_total({A0, A1, A2, A3}, {B0, B1, B2, B3}) ->
+    {B0 + A0, B1 + A1, B2 + A2, B3 + A3};
+add_to_total({A0, A1, A2, A3, A4}, {B0, B1, B2, B3, B4}) ->
+    {B0 + A0, B1 + A1, B2 + A2, B3 + A3, B4 + A4};
+add_to_total({A0, A1, A2, A3, A4, A5}, {B0, B1, B2, B3, B4, B5}) ->
+    {B0 + A0, B1 + A1, B2 + A2, B3 + A3, B4 + A4, B5 + A5};
 add_to_total({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};
 add_to_total({A0, A1, A2, A3, A4, A5, A6, A7}, {B0, B1, B2, B3, B4, B5, B6, B7}) ->
@@ -234,6 +240,12 @@ is_zeros({0, 0}) ->
     true;
 is_zeros({0, 0, 0}) ->
     true;
+is_zeros({0, 0, 0, 0}) ->
+    true;
+is_zeros({0, 0, 0, 0, 0}) ->
+    true;
+is_zeros({0, 0, 0, 0, 0, 0}) ->
+    true;
 is_zeros({0, 0, 0, 0, 0, 0, 0}) ->
     true;
 is_zeros({0, 0, 0, 0, 0, 0, 0, 0, 0}) ->
diff --git a/deps/rabbitmq_management_agent/src/rabbit_mgmt_data.erl b/deps/rabbitmq_management_agent/src/rabbit_mgmt_data.erl
index 01b48a0a5c..bbeb6c57d6 100644
--- a/deps/rabbitmq_management_agent/src/rabbit_mgmt_data.erl
+++ b/deps/rabbitmq_management_agent/src/rabbit_mgmt_data.erl
@@ -377,6 +377,19 @@ lookup_all(Table, Ids, SecondKey) ->
     end.
 
 maybe_convert_for_compatibility(Table, Slide)
+  when Table =:= channel_stats_fine_stats orelse
+       Table =:= channel_exchange_stats_fine_stats orelse
+       Table =:= vhost_stats_fine_stats ->
+     ConversionNeeded = rabbit_feature_flags:is_disabled(
+                          drop_unroutable_metric),
+     case ConversionNeeded of
+         false ->
+             Slide;
+         true ->
+             %% drop_drop because the metric is named "drop_unroutable"
+             rabbit_mgmt_data_compat:drop_drop_unroutable_metric(Slide)
+     end;
+maybe_convert_for_compatibility(Table, Slide)
   when Table =:= channel_queue_stats_deliver_stats orelse
        Table =:= channel_stats_deliver_stats orelse
        Table =:= queue_stats_deliver_stats orelse
@@ -472,12 +485,13 @@ second(Id) ->
     {'_', Id}.
 
 empty(Type, V) when Type =:= connection_stats_coarse_conn_stats;
-            Type =:= channel_stats_fine_stats;
-            Type =:= channel_exchange_stats_fine_stats;
-            Type =:= vhost_stats_fine_stats;
             Type =:= queue_msg_stats;
             Type =:= vhost_msg_stats ->
     {V, V, V};
+empty(Type, V) when Type =:= channel_stats_fine_stats;
+            Type =:= channel_exchange_stats_fine_stats;
+            Type =:= vhost_stats_fine_stats ->
+    {V, V, V, V};
 empty(Type, V) when Type =:= channel_queue_stats_deliver_stats;
             Type =:= queue_stats_deliver_stats;
             Type =:= vhost_stats_deliver_stats;
diff --git a/deps/rabbitmq_management_agent/src/rabbit_mgmt_data_compat.erl b/deps/rabbitmq_management_agent/src/rabbit_mgmt_data_compat.erl
index 22ef01fbd9..a851f23c4e 100644
--- a/deps/rabbitmq_management_agent/src/rabbit_mgmt_data_compat.erl
+++ b/deps/rabbitmq_management_agent/src/rabbit_mgmt_data_compat.erl
@@ -18,7 +18,9 @@
 
 -export([fill_get_empty_queue_metric/1,
          drop_get_empty_queue_metric/1,
-         fill_consumer_active_fields/1]).
+         fill_consumer_active_fields/1,
+         fill_drop_unroutable_metric/1,
+         drop_drop_unroutable_metric/1]).
 
 fill_get_empty_queue_metric(Slide) ->
     exometer_slide:map(
@@ -26,8 +28,7 @@ fill_get_empty_queue_metric(Slide) ->
           (Value) when is_tuple(Value) andalso size(Value) =:= 8 ->
               Value;
           (Value) when is_tuple(Value) andalso size(Value) =:= 7 ->
-              %% We want to remote the last element, which is
-              %% the count of basic.get on empty queues.
+              %% Inject a 0 for the new metric
               list_to_tuple(
                 tuple_to_list(Value) ++ [0]);
           (Value) ->
@@ -38,7 +39,7 @@ drop_get_empty_queue_metric(Slide) ->
     exometer_slide:map(
       fun
           (Value) when is_tuple(Value) andalso size(Value) =:= 8 ->
-              %% We want to remote the last element, which is
+              %% We want to remove the last element, which is
               %% the count of basic.get on empty queues.
               list_to_tuple(
                 lists:sublist(
@@ -49,6 +50,33 @@ drop_get_empty_queue_metric(Slide) ->
               Value
       end, Slide).
 
+fill_drop_unroutable_metric(Slide) ->
+    exometer_slide:map(
+      fun
+          (Value) when is_tuple(Value) andalso size(Value) =:= 4 ->
+              Value;
+          (Value) when is_tuple(Value) andalso size(Value) =:= 3 ->
+              %% Inject a 0
+              list_to_tuple(
+                tuple_to_list(Value) ++ [0]);
+          (Value) ->
+              Value
+      end, Slide).
+
+drop_drop_unroutable_metric(Slide) ->
+    exometer_slide:map(
+      fun
+          (Value) when is_tuple(Value) andalso size(Value) =:= 4 ->
+              %% Remove the last element.
+              list_to_tuple(
+                lists:sublist(
+                  tuple_to_list(Value), size(Value) - 1));
+          (Value) when is_tuple(Value) andalso size(Value) =:= 3 ->
+              Value;
+          (Value) ->
+              Value
+      end, Slide).
+
 fill_consumer_active_fields(ConsumersStats) ->
     [case proplists:get_value(active, ConsumerStats) of
          undefined ->
diff --git a/deps/rabbitmq_management_agent/src/rabbit_mgmt_ff.erl b/deps/rabbitmq_management_agent/src/rabbit_mgmt_ff.erl
index c1290b6095..46af69077a 100644
--- a/deps/rabbitmq_management_agent/src/rabbit_mgmt_ff.erl
+++ b/deps/rabbitmq_management_agent/src/rabbit_mgmt_ff.erl
@@ -21,3 +21,9 @@
     #{desc          => "Count AMQP `basic.get` on empty queues in stats",
       stability     => stable
      }}).
+
+-rabbit_feature_flag(
+  {drop_unroutable_metric,
+   #{desc          => "Count unroutable publishes to be dropped in stats",
+     stability     => stable
+    }}).
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 27c233a7d7..217e1939ba 100644
--- a/deps/rabbitmq_management_agent/src/rabbit_mgmt_metrics_collector.erl
+++ b/deps/rabbitmq_management_agent/src/rabbit_mgmt_metrics_collector.erl
@@ -254,14 +254,14 @@ aggregate_entry({Id, Metrics}, NextStats, Ops0,
     Entry = ?channel_stats(Id, Ftd),
     Ops = insert_op(channel_stats, Id, Entry, Ops0),
     {NextStats, Ops, State};
-aggregate_entry({{Ch, X} = Id, Publish0, Confirm, ReturnUnroutable, 0},
+aggregate_entry({{Ch, X} = Id, Publish0, Confirm, ReturnUnroutable, DropUnroutable, 0},
                 NextStats, Ops0,
                 #state{table = channel_exchange_metrics,
                        policies = {BPolicies, DPolicies, GPolicies},
                        rates_mode = RatesMode,
                        lookup_exchange = ExchangeFun} = State) ->
-    Stats = ?channel_stats_fine_stats(Publish0, Confirm, ReturnUnroutable),
-    {Publish, _, _} = Diff = get_difference(Id, Stats, State),
+    Stats = ?channel_stats_fine_stats(Publish0, Confirm, ReturnUnroutable, DropUnroutable),
+    {Publish, _, _, _} = Diff = get_difference(Id, Stats, State),
 
     Ops1 = insert_entry_ops(channel_stats_fine_stats, Ch, true, Diff, Ops0,
                             BPolicies),
@@ -282,13 +282,13 @@ aggregate_entry({{Ch, X} = Id, Publish0, Confirm, ReturnUnroutable, 0},
                    Ops2
            end,
     {insert_old_aggr_stats(NextStats, Id, Stats), Ops3, State};
-aggregate_entry({{_Ch, X} = Id, Publish0, Confirm, ReturnUnroutable, 1},
+aggregate_entry({{_Ch, X} = Id, Publish0, Confirm, ReturnUnroutable, DropUnroutable, 1},
                 NextStats, Ops0,
                 #state{table = channel_exchange_metrics,
                        policies = {_BPolicies, DPolicies, GPolicies},
                        lookup_exchange = ExchangeFun} = State) ->
-    Stats = ?channel_stats_fine_stats(Publish0, Confirm, ReturnUnroutable),
-    {Publish, _, _} = Diff = get_difference(Id, Stats, State),
+    Stats = ?channel_stats_fine_stats(Publish0, Confirm, ReturnUnroutable, DropUnroutable),
+    {Publish, _, _, _} = Diff = get_difference(Id, Stats, State),
     Ops1 = insert_entry_ops(vhost_stats_fine_stats, vhost(X), true, Diff, Ops0,
                             GPolicies),
     Ops2 = case ExchangeFun(X) of
@@ -612,6 +612,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}, {B0, B1, B2, B3}) ->
+    {B0 + A0, B1 + A1, B2 + A2, B3 + A3};
+sum_entry({A0, A1, A2, A3, A4}, {B0, B1, B2, B3, B4}) ->
+    {B0 + A0, B1 + A1, B2 + A2, B3 + A3, B4 + A4};
 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}) ->
@@ -625,6 +629,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}, {B0, B1, B2, B3}) ->
+    {B0 - A0, B1 - A1, B2 - A2, B3 - A3};
+difference({A0, A1, A2, A3, A4}, {B0, B1, B2, B3, B4}) ->
+    {B0 - A0, B1 - A1, B2 - A2, B3 - A3, B4 - A4};
 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}) ->