Merge pull request #9357 from rabbitmq/mergify/bp/v3.12.x/pr-9356

Improve consumer metric cleanup when a channel goes down (backport #9356)

Author: michaelklishin

deps/rabbit/src/rabbit_amqqueue_process.erl

@@ -909,7 +909,7 @@ handle_ch_down(DownPid, State = #q{consumers                 = Consumers,
             {ok, State1};
         {ChAckTags, ChCTags, Consumers1} ->
             QName = qname(State1),
-            [emit_consumer_deleted(DownPid, CTag, QName, ?INTERNAL_USER) || CTag <- ChCTags],
+            [rabbit_core_metrics:consumer_deleted(DownPid, CTag, QName) || CTag <- ChCTags],
             Holder1 = new_single_active_consumer_after_channel_down(DownPid, Holder, SingleActiveConsumerOn, Consumers1),
             State2 = State1#q{consumers          = Consumers1,
                               active_consumer    = Holder1},

deps/rabbitmq_management_agent/src/rabbit_mgmt_metrics_gc.erl

@@ -136,6 +136,18 @@ delete_samples(Table, Id, Intervals) ->
     [ets:delete(Table, {Id, I}) || I <- Intervals],
     ok.
 
+index_delete(consumer_stats = Table, channel = Type, Id) ->
+    IndexTable = rabbit_mgmt_metrics_collector:index_table(Table, Type),
+    MatchPattern = {'_', Id, '_'},
+    %% Delete consumer_stats_queue_index
+    ets:match_delete(consumer_stats_queue_index,
+                     {'_', MatchPattern}),
+    %% Delete consumer_stats
+    ets:match_delete(consumer_stats,
+                     {MatchPattern,'_'}),
+    %% Delete consumer_stats_channel_index
+    ets:delete(IndexTable, Id),
+    ok;
 index_delete(Table, Type, Id) ->
     IndexTable = rabbit_mgmt_metrics_collector:index_table(Table, Type),
     Keys = ets:lookup(IndexTable, Id),