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

Track messages that were not routed anywhere and also not published as mandatory

Author: michaelklishin

src/rabbit_channel.erl

@@ -2111,8 +2111,9 @@ notify_limiter(Limiter, Acked) ->
 deliver_to_queues({#delivery{message   = #basic_message{exchange_name = XName},
                              confirm   = false,
                              mandatory = false},
-                   []}, State) -> %% optimisation
+                   _RoutedToQs = []}, State) -> %% optimisation
     ?INCR_STATS(exchange_stats, XName, 1, publish, State),
+    ?INCR_STATS(exchange_stats, XName, 1, drop_unroutable, State),
     State;
 deliver_to_queues({Delivery = #delivery{message    = Message = #basic_message{
                                                        exchange_name = XName},
@@ -2165,9 +2166,16 @@ deliver_to_queues({Delivery = #delivery{message    = Message = #basic_message{
     end,
     State2#ch{queue_states = QueueStates}.
 
-process_routing_mandatory(true, [], Msg, State) ->
+process_routing_mandatory(_Mandatory = true,
+                          _RoutedToQs = [],
+                          Msg, State) ->
     ok = basic_return(Msg, State, no_route),
     ok;
+process_routing_mandatory(_Mandatory = false,
+                          _RoutedToQs = [],
+                          #basic_message{exchange_name = ExchangeName}, State) ->
+    ?INCR_STATS(exchange_stats, ExchangeName, 1, drop_unroutable, State),
+    ok;
 process_routing_mandatory(_, _, _, _) ->
     ok.
 

src/rabbit_core_metrics_gc.erl

@@ -157,7 +157,7 @@ gc_process_and_entity(Table, GbSet) ->
     ets:foldl(fun({{Pid, Id} = Key, _, _, _, _, _, _, _, _}, none)
                   when Table == channel_queue_metrics ->
                       gc_process_and_entity(Id, Pid, Table, Key, GbSet);
-                 ({{Pid, Id} = Key, _, _, _, _}, none)
+                 ({{Pid, Id} = Key, _, _, _, _, _}, none)
                     when Table == channel_exchange_metrics ->
                       gc_process_and_entity(Id, Pid, Table, Key, GbSet);
                  ({{Id, Pid, _} = Key, _, _, _, _, _, _}, none)

test/rabbit_core_metrics_gc_SUITE.erl

@@ -177,6 +177,8 @@ channel_metrics(Config) ->
     amqp_channel:call(Ch, #'queue.declare'{queue = <<"queue_metrics">>}),
     amqp_channel:cast(Ch, #'basic.publish'{routing_key = <<"queue_metrics">>},
                       #amqp_msg{payload = <<"hello">>}),
+    amqp_channel:cast(Ch, #'basic.publish'{routing_key = <<"won't route $¢% anywhere">>},
+                      #amqp_msg{payload = <<"hello">>}),
     {#'basic.get_ok'{}, _} = amqp_channel:call(Ch, #'basic.get'{queue = <<"queue_metrics">>,
                                                                 no_ack=true}),
     timer:sleep(150),

test/unit_inbroker_non_parallel_SUITE.erl

@@ -510,7 +510,7 @@ channel_statistics1(_Config) ->
                      [{{Ch, QRes}, 1, 0, 0, 0, 0, 0, 0, 0}] = ets:lookup(
                                                                 channel_queue_metrics,
                                                                 {Ch, QRes}),
-                     [{{Ch, X}, 1, 0, 0, 0}] = ets:lookup(
+                     [{{Ch, X}, 1, 0, 0, 0, 0}] = ets:lookup(
                                                  channel_exchange_metrics,
                                                  {Ch, X}),
                      [{{Ch, {QRes, X}}, 1, 0}] = ets:lookup(
@@ -525,7 +525,7 @@ channel_statistics1(_Config) ->
                      [{{Ch, QRes}, 1, 0, 0, 0, 0, 0, 0, 1}] = ets:lookup(
                                                                 channel_queue_metrics,
                                                                 {Ch, QRes}),
-                 [{{Ch, X}, 1, 0, 0, 0}] = ets:lookup(
+                 [{{Ch, X}, 1, 0, 0, 0, 0}] = ets:lookup(
                                              channel_exchange_metrics,
                                              {Ch, X}),
                  [{{Ch, {QRes, X}}, 1, 1}] = ets:lookup(
@@ -538,7 +538,7 @@ channel_statistics1(_Config) ->
     force_metric_gc(),
     Check4 = fun() ->
                  [] = ets:lookup(channel_queue_metrics, {Ch, QRes}),
-                 [{{Ch, X}, 1, 0, 0, 0}] = ets:lookup(
+                 [{{Ch, X}, 1, 0, 0, 0, 0}] = ets:lookup(
                                              channel_exchange_metrics,
                                              {Ch, X}),
                  [] = ets:lookup(channel_queue_exchange_metrics,