Merge pull request #7659 from rabbitmq/rin/queue-deleted-events-include-queue-type

Include the queue type in the queue_deleted rabbit_event

Author: michaelklishin

deps/rabbit/src/rabbit_amqqueue.erl

@@ -133,14 +133,6 @@ recover(VHost) ->
 filter_pid_per_type(QPids) ->
     lists:partition(fun(QPid) -> ?IS_CLASSIC(QPid) end, QPids).
 
-filter_resource_per_type(Resources) ->
-    Queues = [begin
-                  {ok, Q} = lookup(Resource),
-                  QPid = amqqueue:get_pid(Q),
-                  {Resource, QPid}
-              end || Resource <- Resources],
-    lists:partition(fun({_Resource, QPid}) -> ?IS_CLASSIC(QPid) end, Queues).
-
 -spec stop(rabbit_types:vhost()) -> 'ok'.
 stop(VHost) ->
     %% Classic queues
@@ -1506,11 +1498,17 @@ delete_immediately(QPids) ->
     end.
 
 delete_immediately_by_resource(Resources) ->
-    {Classic, Quorum} = filter_resource_per_type(Resources),
-    [gen_server2:cast(QPid, delete_immediately) || {_, QPid} <- Classic],
-    [rabbit_quorum_queue:delete_immediately(Resource, QPid)
-     || {Resource, QPid} <- Quorum],
-    ok.
+    lists:foreach(
+      fun(Resource) ->
+              {ok, Q} = lookup(Resource),
+              QPid = amqqueue:get_pid(Q),
+              case ?IS_CLASSIC(QPid) of
+                  true ->
+                      gen_server2:cast(QPid, delete_immediately);
+                  _ ->
+                      rabbit_quorum_queue:delete_immediately(Q)
+              end
+      end, Resources).
 
 -spec delete
         (amqqueue:amqqueue(), 'false', 'false', rabbit_types:username()) ->
@@ -1676,12 +1674,13 @@ notify_sent_queue_down(QPid) ->
 resume(QPid, ChPid) -> delegate:invoke_no_result(QPid, {gen_server2, cast,
                                                         [{resume, ChPid}]}).
 
--spec internal_delete(name(), rabbit_types:username()) -> 'ok'.
+-spec internal_delete(amqqueue:amqqueue(), rabbit_types:username()) -> 'ok'.
 
-internal_delete(QueueName, ActingUser) ->
-    internal_delete(QueueName, ActingUser, normal).
+internal_delete(Queue, ActingUser) ->
+    internal_delete(Queue, ActingUser, normal).
 
-internal_delete(QueueName, ActingUser, Reason) ->
+internal_delete(Queue, ActingUser, Reason) ->
+    QueueName = amqqueue:get_name(Queue),
     case rabbit_db_queue:delete(QueueName, Reason) of
         ok ->
             ok;
@@ -1691,6 +1690,7 @@ internal_delete(QueueName, ActingUser, Reason) ->
             rabbit_core_metrics:queue_deleted(QueueName),
             ok = rabbit_event:notify(queue_deleted,
                                      [{name, QueueName},
+                                      {type, amqqueue:get_type(Queue)},
                                       {user_who_performed_action, ActingUser}])
     end.
 
@@ -1874,7 +1874,7 @@ on_node_down(Node) ->
             end,
             notify_queue_binding_deletions(Deletions),
             rabbit_core_metrics:queues_deleted(QueueNames),
-            notify_queues_deleted(QueueNames),
+            notify_transient_queues_deleted(QueueNames),
             ok
     end.
 
@@ -1897,11 +1897,12 @@ notify_queue_binding_deletions(QueueDeletions) ->
     Deletions = rabbit_binding:process_deletions(QueueDeletions),
     rabbit_binding:notify_deletions(Deletions, ?INTERNAL_USER).
 
-notify_queues_deleted(QueueDeletions) ->
+notify_transient_queues_deleted(QueueDeletions) ->
     lists:foreach(
       fun(Queue) ->
               ok = rabbit_event:notify(queue_deleted,
                                        [{name, Queue},
+                                        {kind, rabbit_classic_queue},
                                         {user, ?INTERNAL_USER}])
       end,
       QueueDeletions).

deps/rabbit/src/rabbit_amqqueue_process.erl

@@ -314,7 +314,6 @@ terminate_delete(EmitStats, Reason0,
                  State = #q{q = Q,
                             backing_queue = BQ,
                             status = Status}) ->
-    QName = amqqueue:get_name(Q),
     ActingUser = terminated_by(Status),
     fun (BQS) ->
         Reason = case Reason0 of
@@ -330,7 +329,7 @@ terminate_delete(EmitStats, Reason0,
         %% logged.
         try
             %% don't care if the internal delete doesn't return 'ok'.
-            rabbit_amqqueue:internal_delete(QName, ActingUser, Reason0)
+            rabbit_amqqueue:internal_delete(Q, ActingUser, Reason0)
         catch
             {error, ReasonE} -> error(ReasonE)
         end,

deps/rabbit/src/rabbit_classic_queue.erl

@@ -466,10 +466,9 @@ delete_crashed(Q, ActingUser) ->
                   [Q, ActingUser]).
 
 delete_crashed_internal(Q, ActingUser) ->
-    QName = amqqueue:get_name(Q),
     {ok, BQ} = application:get_env(rabbit, backing_queue_module),
     BQ:delete_crashed(Q),
-    ok = rabbit_amqqueue:internal_delete(QName, ActingUser).
+    ok = rabbit_amqqueue:internal_delete(Q, ActingUser).
 
 recover_durable_queues(QueuesAndRecoveryTerms) ->
     {Results, Failures} =

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -20,7 +20,7 @@
          restart_server/1,
          stop_server/1,
          delete/4,
-         delete_immediately/2]).
+         delete_immediately/1]).
 -export([state_info/1, info/2, stat/1, infos/1]).
 -export([settle/5, dequeue/5, consume/3, cancel/5]).
 -export([credit/5]).
@@ -223,20 +223,20 @@ start_cluster(Q) ->
                                           ActingUser}]),
                     {new, NewQ};
                 {error, Error} ->
-                    declare_queue_error(Error, QName, Leader, ActingUser)
+                    declare_queue_error(Error, NewQ, Leader, ActingUser)
             catch
                 error:Error ->
-                    declare_queue_error(Error, QName, Leader, ActingUser)
+                    declare_queue_error(Error, NewQ, Leader, ActingUser)
             end;
         {existing, _} = Ex ->
             Ex
     end.
 
-declare_queue_error(Error, QName, Leader, ActingUser) ->
-    _ = rabbit_amqqueue:internal_delete(QName, ActingUser),
+declare_queue_error(Error, Queue, Leader, ActingUser) ->
+    _ = rabbit_amqqueue:internal_delete(Queue, ActingUser),
     {protocol_error, internal_error,
      "Cannot declare quorum ~ts on node '~ts' with leader on node '~ts': ~255p",
-     [rabbit_misc:rs(QName), node(), Leader, Error]}.
+     [rabbit_misc:rs(amqqueue:get_name(Queue)), node(), Leader, Error]}.
 
 ra_machine(Q) ->
     {module, rabbit_fifo, ra_machine_config(Q)}.
@@ -660,7 +660,7 @@ delete(Q, _IfUnused, _IfEmpty, ActingUser) when ?amqqueue_is_quorum(Q) ->
                     ok = force_delete_queue(Servers)
             end,
             notify_decorators(QName, shutdown),
-            ok = delete_queue_data(QName, ActingUser),
+            ok = delete_queue_data(Q, ActingUser),
             _ = erpc:call(LeaderNode, rabbit_core_metrics, queue_deleted, [QName],
                           ?RPC_TIMEOUT),
             {ok, ReadyMsgs};
@@ -671,7 +671,7 @@ delete(Q, _IfUnused, _IfEmpty, ActingUser) when ?amqqueue_is_quorum(Q) ->
                 true ->
                     %% If all ra nodes were already down, the delete
                     %% has succeed
-                    delete_queue_data(QName, ActingUser),
+                    delete_queue_data(Q, ActingUser),
                     {ok, ReadyMsgs};
                 false ->
                     %% attempt forced deletion of all servers
@@ -682,7 +682,7 @@ delete(Q, _IfUnused, _IfEmpty, ActingUser) when ?amqqueue_is_quorum(Q) ->
                       [rabbit_misc:rs(QName), Errs]),
                     ok = force_delete_queue(Servers),
                     notify_decorators(QName, shutdown),
-                    delete_queue_data(QName, ActingUser),
+                    delete_queue_data(Q, ActingUser),
                     {ok, ReadyMsgs}
             end
     end.
@@ -701,15 +701,15 @@ force_delete_queue(Servers) ->
      end || S <- Servers],
     ok.
 
-delete_queue_data(QName, ActingUser) ->
-    _ = rabbit_amqqueue:internal_delete(QName, ActingUser),
+delete_queue_data(Queue, ActingUser) ->
+    _ = rabbit_amqqueue:internal_delete(Queue, ActingUser),
     ok.
 
 
-delete_immediately(Resource, {_Name, _} = QPid) ->
-    _ = rabbit_amqqueue:internal_delete(Resource, ?INTERNAL_USER),
-    {ok, _} = ra:delete_cluster([QPid]),
-    rabbit_core_metrics:queue_deleted(Resource),
+delete_immediately(Queue) ->
+    _ = rabbit_amqqueue:internal_delete(Queue, ?INTERNAL_USER),
+    {ok, _} = ra:delete_cluster([amqqueue:get_pid(Queue)]),
+    rabbit_core_metrics:queue_deleted(amqqueue:get_name(Queue)),
     ok.
 
 settle(_QName, complete, CTag, MsgIds, QState) ->

deps/rabbit/src/rabbit_stream_coordinator.erl

@@ -177,8 +177,7 @@ delete_stream(Q, ActingUser)
     #{name := StreamId} = amqqueue:get_type_state(Q),
     case process_command({delete_stream, StreamId, #{}}) of
         {ok, ok, _} ->
-            QName = amqqueue:get_name(Q),
-              _ = rabbit_amqqueue:internal_delete(QName, ActingUser),
+            _ = rabbit_amqqueue:internal_delete(Q, ActingUser),
             {ok, {ok, 0}};
         Err ->
             Err

deps/rabbit/src/rabbit_stream_queue.erl

@@ -160,8 +160,7 @@ create_stream(Q0) ->
                                           ActingUser}]),
                     {new, Q};
                 Error ->
-
-                    _ = rabbit_amqqueue:internal_delete(QName, ActingUser),
+                    _ = rabbit_amqqueue:internal_delete(Q, ActingUser),
                     {protocol_error, internal_error, "Cannot declare a queue '~ts' on node '~ts': ~255p",
                      [rabbit_misc:rs(QName), node(), Error]}
             end;

deps/rabbit/src/rabbit_vhost.erl

@@ -413,8 +413,7 @@ assert_benign({error, not_found}, _) -> ok;
 assert_benign({error, {absent, Q, _}}, ActingUser) ->
     %% Removing the database entries here is safe. If/when the down node
     %% restarts, it will clear out the on-disk storage of the queue.
-    QName = amqqueue:get_name(Q),
-    rabbit_amqqueue:internal_delete(QName, ActingUser).
+    rabbit_amqqueue:internal_delete(Q, ActingUser).
 
 -spec exists(vhost:name()) -> boolean().
 

deps/rabbit/test/backing_queue_SUITE.erl

@@ -911,7 +911,7 @@ bq_queue_recover1(Config) ->
                   rabbit_variable_queue:fetch(true, VQ1),
               CountMinusOne = rabbit_variable_queue:len(VQ2),
               _VQ3 = rabbit_variable_queue:delete_and_terminate(shutdown, VQ2),
-              ok = rabbit_amqqueue:internal_delete(QName, <<"acting-user">>)
+              ok = rabbit_amqqueue:internal_delete(Q1, <<"acting-user">>)
       end),
     passed.
 

deps/rabbitmq_mqtt/src/rabbit_mqtt_qos0_queue.erl

@@ -97,7 +97,7 @@ declare(Q0, _Node) ->
 delete(Q, _IfUnused, _IfEmpty, ActingUser) ->
     QName = amqqueue:get_name(Q),
     log_delete(QName, amqqueue:get_exclusive_owner(Q)),
-    ok = rabbit_amqqueue:internal_delete(QName, ActingUser),
+    ok = rabbit_amqqueue:internal_delete(Q, ActingUser),
     {ok, 0}.
 
 -spec deliver([{amqqueue:amqqueue(), stateless}], Delivery :: term()) ->
@@ -164,7 +164,7 @@ recover(_VHost, Queues) ->
               true = is_recoverable(Q),
               QName = amqqueue:get_name(Q),
               log_delete(QName, amqqueue:get_exclusive_owner(Q)),
-              rabbit_amqqueue:internal_delete(QName, ?INTERNAL_USER)
+              rabbit_amqqueue:internal_delete(Q, ?INTERNAL_USER)
       end, Queues),
     %% We mark the queue recovery as failed because these queues are not really
     %% recovered, but deleted.