Merge pull request #1761 from rabbitmq/quorum-data-cleanup

Data cleanup for quorum queues

Author: michaelklishin

src/rabbit_amqqueue.erl

@@ -30,6 +30,7 @@
          emit_info_all/5, list_local/1, info_local/1,
 	 emit_info_local/4, emit_info_down/4]).
 -export([list_down/1, count/1, list_names/0, list_local_names/0]).
+-export([list_by_type/1]).
 -export([notify_policy_changed/1]).
 -export([consumers/1, consumers_all/1,  emit_consumers_all/4, consumer_info_keys/0]).
 -export([basic_get/6, basic_consume/12, basic_cancel/6, notify_decorators/1]).
@@ -124,6 +125,7 @@
 -spec list(rabbit_types:vhost()) -> [rabbit_types:amqqueue()].
 -spec list_names() -> [rabbit_amqqueue:name()].
 -spec list_down(rabbit_types:vhost()) -> [rabbit_types:amqqueue()].
+-spec list_by_type(atom()) -> [rabbit_types:amqqueue()].
 -spec info_keys() -> rabbit_types:info_keys().
 -spec info(rabbit_types:amqqueue()) -> rabbit_types:infos().
 -spec info(rabbit_types:amqqueue(), rabbit_types:info_keys()) ->
@@ -741,6 +743,15 @@ list_local_names() ->
     [ Q#amqqueue.name || #amqqueue{state = State, pid = QPid} = Q <- list(),
            State =/= crashed, is_local_to_node(QPid, node())].
 
+list_by_type(Type) ->
+    {atomic, Qs} =
+        mnesia:sync_transaction(
+          fun () ->
+                  mnesia:match_object(rabbit_durable_queue,
+                                      #amqqueue{_ = '_', type = Type}, read)
+          end),
+    Qs.
+
 list_local_followers() ->
     [ Q#amqqueue.name
       || #amqqueue{state = State, type = quorum, pid = {_, Leader},

src/rabbit_quorum_queue.erl

@@ -34,6 +34,7 @@
 -export([add_member/3]).
 -export([delete_member/3]).
 -export([requeue/3]).
+-export([cleanup_data_dir/0]).
 
 -include_lib("rabbit_common/include/rabbit.hrl").
 -include_lib("stdlib/include/qlc.hrl").
@@ -240,7 +241,7 @@ recover(Queues) ->
                  case ra:start_server(Name, {Name, node()}, Machine, RaNodes) of
                      ok -> ok;
                      Err ->
-                         rabbit_log:warning("recover: Quorum queue ~w could not"
+                         rabbit_log:warning("recover: quorum queue ~w could not"
                                             " be started ~w", [Name, Err]),
                          ok
                  end;
@@ -251,7 +252,7 @@ recover(Queues) ->
                  ok;
              Err ->
                  %% catch all clause to avoid causing the vhost not to start
-                 rabbit_log:warning("recover: Quorum queue ~w could not be "
+                 rabbit_log:warning("recover: quorum queue ~w could not be "
                                     "restarted ~w", [Name, Err]),
                  ok
          end,
@@ -281,7 +282,7 @@ delete(#amqqueue{ type = quorum, pid = {Name, _}, name = QName, quorum_nodes = Q
             end,
             rpc:call(LeaderNode, rabbit_core_metrics, queue_deleted, [QName]),
             {ok, Msgs};
-        {error, {no_more_nodes_to_try, Errs}} = Err ->
+        {error, {no_more_servers_to_try, Errs}} ->
             case lists:all(fun({{error, noproc}, _}) -> true;
                               (_) -> false
                            end, Errs) of
@@ -291,7 +292,10 @@ delete(#amqqueue{ type = quorum, pid = {Name, _}, name = QName, quorum_nodes = Q
                     rabbit_core_metrics:queue_deleted(QName),
                     {ok, Msgs};
                 false ->
-                    Err
+                    rabbit_misc:protocol_error(
+                      internal_error,
+                      "Cannot delete quorum queue '~s', not enough nodes online to reach a quorum: ~255p",
+                      [rabbit_misc:rs(QName), Errs])
             end
     end.
 
@@ -386,6 +390,26 @@ purge(Node) ->
 requeue(ConsumerTag, MsgIds, FState) ->
     rabbit_fifo_client:return(quorum_ctag(ConsumerTag), MsgIds, FState).
 
+cleanup_data_dir() ->
+    Names = [Name || #amqqueue{pid = {Name, _}, quorum_nodes = Nodes}
+                         <- rabbit_amqqueue:list_by_type(quorum),
+                     lists:member(node(), Nodes)],
+    Registered = ra_directory:list_registered(),
+    [maybe_delete_data_dir(UId) || {Name, UId} <- Registered,
+                                   not lists:member(Name, Names)],
+    ok.
+
+maybe_delete_data_dir(UId) ->
+    Dir = ra_env:server_data_dir(UId),
+    {ok, Config} = ra_log:read_config(Dir),
+    case maps:get(machine, Config) of
+        {module, rabbit_fifo, _} ->
+            ra_lib:recursive_delete(Dir),
+            ra_directory:unregister_name(UId);
+        _ ->
+            ok
+    end.
+
 cluster_state(Name) ->
     case whereis(Name) of
         undefined -> down;

test/quorum_queue_SUITE.erl

@@ -36,6 +36,7 @@ groups() ->
                         {cluster_size_2, [], [add_member]}
                        ]},
      {clustered, [], [
+                      {cluster_size_2, [], [cleanup_data_dir]},
                       {cluster_size_2, [], [add_member_not_running,
                                             add_member_classic,
                                             add_member_already_a_member,
@@ -1586,6 +1587,34 @@ delete_member(Config) ->
                  rpc:call(Server, rabbit_quorum_queue, delete_member,
                           [<<"/">>, QQ, Server])).
 
+cleanup_data_dir(Config) ->
+    %% This test is slow, but also checks that we handle properly errors when
+    %% trying to delete a queue in minority. A case clause there had gone
+    %% previously unnoticed.
+
+    [Server1, Server2] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    Ch = rabbit_ct_client_helpers:open_channel(Config, Server1),
+    QQ = ?config(queue_name, Config),
+    ?assertEqual({'queue.declare_ok', QQ, 0, 0},
+                 declare(Ch, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}])),
+    timer:sleep(100),
+
+    [{_, UId}] = rpc:call(Server1, ra_directory, list_registered, []),
+    DataDir = rpc:call(Server1, ra_env, server_data_dir, [UId]),
+    ?assert(filelib:is_dir(DataDir)),
+
+    ok = rabbit_ct_broker_helpers:stop_node(Config, Server2),
+
+    ?assertExit({{shutdown,
+                  {connection_closing, {server_initiated_close, 541, _}}}, _},
+                amqp_channel:call(Ch, #'queue.delete'{queue = QQ})),
+    ?assert(filelib:is_dir(DataDir)),
+
+    ?assertEqual(ok,
+                 rpc:call(Server1, rabbit_quorum_queue, cleanup_data_dir,
+                          [])),
+    ?assert(not filelib:is_dir(DataDir)).
+
 basic_recover(Config) ->
     [Server | _] = Servers = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),