Test and clean up metrics dependent on both process and entity

It leaked consumer_created when the queue still existed, now both
entity and process must exists to keep the metrics.

2 files changed, 36 insertions, 12 deletions

diff --git a/src/rabbit_core_metrics_gc.erl b/src/rabbit_core_metrics_gc.erl
index 99a8058bde..3141fdc301 100644
--- a/src/rabbit_core_metrics_gc.erl
+++ b/src/rabbit_core_metrics_gc.erl
@@ -64,7 +64,6 @@ gc_connections() ->
     gc_process(connection_coarse_metrics).
 
 gc_channels() ->
-    %% TODO channel stats
     gc_process(channel_created),
     gc_process(channel_metrics),
     gc_process(channel_process_metrics),
@@ -107,7 +106,6 @@ gc_process(Pid, Table, Key) ->
         true ->
             none;
         false ->
-            %% TODO catch?
             ets:delete(Table, Key),
             none
     end.
@@ -128,7 +126,6 @@ gc_entity(Id, Table, Key, GbSet) ->
         true ->
             none;
         false ->
-            %% TODO catch?
             ets:delete(Table, Key),
             none
     end.
@@ -136,26 +133,22 @@ gc_entity(Id, Table, Key, GbSet) ->
 gc_process_and_entity(Table, GbSet) ->
     ets:foldl(fun({{Pid, Id} = Key, _, _, _, _, _, _, _}, none)
                   when Table == channel_queue_metrics ->
-                      gc_entity(Id, Table, Key, GbSet),
-                      gc_process(Pid, Table, Key);
+                      gc_process_and_entity(Id, Pid, Table, Key, GbSet);
                  ({{Pid, Id} = Key, _, _, _, _}, none)
                     when Table == channel_exchange_metrics ->
-                      gc_entity(Id, Table, Key, GbSet),
-                      gc_process(Pid, Table, Key);
+                      gc_process_and_entity(Id, Pid, Table, Key, GbSet);
                  ({{Id, Pid, _} = Key, _, _, _, _}, none)
                     when Table == consumer_created ->
-                      gc_entity(Id, Table, Key, GbSet),
-                      gc_process(Pid, Table, Key);
+                      gc_process_and_entity(Id, Pid, Table, Key, GbSet);
                  ({{{Pid, Id}, _} = Key, _, _, _, _}, none) ->
                       gc_process_and_entity(Id, Pid, Table, Key, GbSet)
               end, none, Table).
 
 gc_process_and_entity(Id, Pid, Table, Key, GbSet) ->
-    case rabbit_misc:is_process_alive(Pid) orelse gb_sets:is_member(Id, GbSet) of
+    case rabbit_misc:is_process_alive(Pid) andalso gb_sets:is_member(Id, GbSet) of
         true ->
             none;
         false ->
-            %% TODO catch?
             ets:delete(Table, Key),
             none
     end.
diff --git a/test/rabbit_core_metrics_gc_SUITE.erl b/test/rabbit_core_metrics_gc_SUITE.erl
index c3997f31e2..6e7607eea6 100644
--- a/test/rabbit_core_metrics_gc_SUITE.erl
+++ b/test/rabbit_core_metrics_gc_SUITE.erl
@@ -34,7 +34,8 @@ groups() ->
         connection_metrics,
         channel_metrics,
         node_metrics,
-        gen_server2_metrics
+        gen_server2_metrics,
+        consumer_metrics
       ]
      }
     ].
@@ -286,6 +287,36 @@ gen_server2_metrics(Config) ->
 
     ok.
 
+consumer_metrics(Config) ->
+    A = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
+    Ch = rabbit_ct_client_helpers:open_channel(Config, A),
+
+    amqp_channel:call(Ch, #'queue.declare'{queue = <<"queue_metrics">>}),
+    amqp_channel:call(Ch, #'basic.consume'{queue = <<"queue_metrics">>}),
+    timer:sleep(200),
+
+    DeadPid = rabbit_ct_broker_helpers:rpc(Config, A, ?MODULE, dead_pid, []),
+
+    QName = q(<<"queue_metrics">>),
+    CTag = <<"tag">>,
+    rabbit_ct_broker_helpers:rpc(Config, A, rabbit_core_metrics,
+                                 consumer_created, [DeadPid, CTag, true, true,
+                                                    QName, 1, []]),
+    Id = {QName, DeadPid, CTag},
+    [_] = rabbit_ct_broker_helpers:rpc(Config, A, ets, lookup, [consumer_created, Id]),
+
+    %% Trigger gc. When the gen_server:call returns, the gc has already finished.
+    rabbit_ct_broker_helpers:rpc(Config, A, erlang, send, [rabbit_core_metrics_gc, start_gc]),
+    rabbit_ct_broker_helpers:rpc(Config, A, gen_server, call, [rabbit_core_metrics_gc, test]),
+
+    [_|_] = rabbit_ct_broker_helpers:rpc(Config, A, ets, tab2list, [consumer_created]),
+    [] = rabbit_ct_broker_helpers:rpc(Config, A, ets, lookup, [consumer_created, Id]),
+
+    amqp_channel:call(Ch, #'queue.delete'{queue = <<"queue_metrics">>}),
+    rabbit_ct_client_helpers:close_channel(Ch),
+
+    ok.
+
 dead_pid() ->
     spawn(fun() -> ok end).