Delete queue metrics after sync.

This ensures that the core metrics for a queue is cleared up after a
queue master might have moved to another node.

[#147753285]

Author: kjnilsson

src/rabbit_amqqueue_process.erl

@@ -258,11 +258,13 @@ init_with_backing_queue_state(Q = #amqqueue{exclusive_owner = Owner}, BQ, BQS,
     State3.
 
 terminate(shutdown = R,      State = #q{backing_queue = BQ}) ->
+    rabbit_core_metrics:queue_deleted(qname(State)),
     terminate_shutdown(fun (BQS) -> BQ:terminate(R, BQS) end, State);
 terminate({shutdown, missing_owner} = Reason, State) ->
     %% if the owner was missing then there will be no queue, so don't emit stats
     terminate_shutdown(terminate_delete(false, Reason, State), State);
 terminate({shutdown, _} = R, State = #q{backing_queue = BQ}) ->
+    rabbit_core_metrics:queue_deleted(qname(State)),
     terminate_shutdown(fun (BQS) -> BQ:terminate(R, BQS) end, State);
 terminate(normal,            State) -> %% delete case
     terminate_shutdown(terminate_delete(true, normal, State), State);

src/rabbit_core_metrics_gc.erl

@@ -54,9 +54,13 @@ terminate(_Reason, #state{timer = TRef}) ->
 code_change(_OldVsn, State, _Extra) ->
     {ok, State}.
 
+start_timer(Interval, #state{timer = TRef0} = St) ->
+    timer:cancel(TRef0),
+    TRef1 = erlang:send_after(Interval, self(), start_gc),
+    St#state{timer = TRef1}.
+
 start_timer(#state{interval = Interval} = St) ->
-    TRef = erlang:send_after(Interval, self(), start_gc),
-    St#state{timer = TRef}.
+    start_timer(Interval, St).
 
 gc_connections() ->
     gc_process(connection_created),

src/rabbit_mirror_queue_sync.erl

@@ -263,9 +263,17 @@ syncer_loop(Ref, MPid, SPids) ->
             %% want.
             ok;
         {done, Ref} ->
+            cleanup_metrics(),
             [SPid ! {sync_complete, Ref} || SPid <- SPids]
     end.
 
+cleanup_metrics() ->
+    case get(process_name) of
+        {_Module, Queue} ->
+            rabbit_core_metrics:queue_deleted(Queue);
+        _ -> ok
+    end.
+
 broadcast(SPids, Msg) ->
     [begin
          credit_flow:send(SPid),

test/rabbit_core_metrics_gc_SUITE.erl

@@ -343,11 +343,6 @@ cluster_queue_metrics(Config) ->
     Node0 = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
     Node1 = rabbit_ct_broker_helpers:get_node_config(Config, 1, nodename),
 
-    rabbit_ct_broker_helpers:rpc(Config, Node0, erlang, send, [rabbit_core_metrics_gc, start_gc]),
-    rabbit_ct_broker_helpers:rpc(Config, Node0, gen_server, call, [rabbit_core_metrics_gc, test]),
-    rabbit_ct_broker_helpers:rpc(Config, Node1, erlang, send, [rabbit_core_metrics_gc, start_gc]),
-    rabbit_ct_broker_helpers:rpc(Config, Node1, gen_server, call, [rabbit_core_metrics_gc, test]),
-
     Ch = rabbit_ct_client_helpers:open_channel(Config, Node0),
 
     Node0Name = rabbit_data_coercion:to_binary(Node0),
@@ -370,24 +365,21 @@ cluster_queue_metrics(Config) ->
     % Synchronize
     Name = rabbit_misc:r(VHost, queue, QueueName),
     [#amqqueue{pid = QPid}] = rabbit_ct_broker_helpers:rpc(Config, Node0,
-                                                           ets, lookup, [rabbit_queue, Name]),
-    ok = rabbit_ct_broker_helpers:rpc(Config, Node0, rabbit_amqqueue, sync_mirrors, [QPid]),
-
-    timer:sleep(1000),
+                                                           ets, lookup,
+                                                           [rabbit_queue, Name]),
+    ok = rabbit_ct_broker_helpers:rpc(Config, Node0, rabbit_amqqueue,
+                                      sync_mirrors, [QPid]),
 
-    rabbit_ct_broker_helpers:rpc(Config, Node0, erlang, send, [rabbit_core_metrics_gc, start_gc]),
-    rabbit_ct_broker_helpers:rpc(Config, Node0, gen_server, call, [rabbit_core_metrics_gc, test]),
-    rabbit_ct_broker_helpers:rpc(Config, Node1, erlang, send, [rabbit_core_metrics_gc, start_gc]),
-    rabbit_ct_broker_helpers:rpc(Config, Node1, gen_server, call, [rabbit_core_metrics_gc, test]),
+    timer:sleep(1500),
 
     % Check ETS table for data
     % rabbit_core_metrics:queue_stats
-    % {Name, MessagesReady, MessagesUnacknowledge, Messages, Reductions}
-    % [{{resource,<<"/">>,queue,<<"cluster_queue_metrics">>}, 1,0,1,10524}]
-    [] = rabbit_ct_broker_helpers:rpc(Config, Node0, ets, tab2list, [queue_coarse_metrics]),
+    [] = rabbit_ct_broker_helpers:rpc(Config, Node0, ets, tab2list,
+                                      [queue_coarse_metrics]),
 
-    EtsData1_0 = rabbit_ct_broker_helpers:rpc(Config, Node1, ets, tab2list, [queue_coarse_metrics]),
-    [{Name, 1, 0, 1, _}] = EtsData1_0,
+    [{Name, 1, 0, 1, _}] = rabbit_ct_broker_helpers:rpc(Config, Node1, ets,
+                                                        tab2list,
+                                                        [queue_coarse_metrics]),
 
     amqp_channel:call(Ch, #'queue.delete'{queue=QueueName}),
     rabbit_ct_client_helpers:close_channel(Ch),