Make queue commands generic - i.e. make them work with any queue type

Author: ikavgo

deps/rabbit/src/rabbit_classic_queue.erl

@@ -1,5 +1,6 @@
 -module(rabbit_classic_queue).
 -behaviour(rabbit_queue_type).
+-behaviour(rabbit_queue_commands).
 -behaviour(rabbit_policy_validator).
 
 -include("amqqueue.hrl").
@@ -74,6 +75,15 @@
 
 -export([validate_policy/1]).
 
+%% commands
+-export([add_member/5,
+         list_with_local_promotable/0,
+         delete_member/3,
+         peek/2,
+         status/1,
+         reclaim_memory/1,
+         shrink_all/1]).
+
 -rabbit_boot_step(
    {rabbit_classic_queue_type,
     [{description, "Classic queue: queue type"},
@@ -730,3 +740,24 @@ queue_vm_stats_sups() ->
 %% Other as usual by substraction.
 queue_vm_ets() ->
     {[], []}.
+
+add_member(_VHost, _Name, _Node, _Membership, _Timeout) ->
+    {error, classic_queue_not_supported}.
+
+list_with_local_promotable() ->
+    {error, classic_queue_not_supported}.
+
+delete_member(_VHost, _Name, _Node) ->
+    {error, classic_queue_not_supported}.
+
+peek(_Pos, _QName) ->
+    {error, classic_queue_not_supported}.
+
+status(_QName) ->
+    {error, classic_queue_not_supported}.
+
+reclaim_memory(_QName) ->
+    {error, classic_queue_not_supported}.
+
+shrink_all(_Node) ->
+    {error, classic_queue_not_supported}.

deps/rabbit/src/rabbit_observer_cli_quorum_queues.erl

@@ -138,7 +138,7 @@ sheet_body(PrevState) ->
                                         empty_row(Name);
                                     _ ->
                                         QQCounters = maps:get({QName, node()}, ra_counters:overview()),
-                                        {ok, InternalName} = rabbit_queue_type_util:qname_to_internal_name(#resource{virtual_host = Vhost, name= Name}),
+                                        {ok, InternalName} = rabbit_queue_type_util:qname_to_internal_name(rabbit_misc:queue_resource(Vhost, Name)),
                                         [{_, CT, SnapIdx, LA, CI, LW, CL}] = ets:lookup(ra_metrics, R),
                                         [
                                          Pid,

deps/rabbit/src/rabbit_queue_commands.erl

@@ -0,0 +1,146 @@
+-module(rabbit_queue_commands).
+
+-export([add_member/5,
+         list_with_local_promotable/1,
+         delete_member/3,
+         grow/5,
+         peek/2,
+         status/1,
+         reclaim_memory/2,
+         shrink_all/2]).
+
+-include_lib("rabbit_common/include/resource.hrl").
+
+%% TODO: membership is a subset of ra_membership() with 'unknown' removed.
+-type membership() :: voter | non_voter | promotable.
+
+-callback add_member(VHost :: rabbit_types:vhost(), Name :: resource_name(), Node :: node(), Membership :: membership(), Timeout :: timeout()) -> ok | {error, term()}.
+
+-callback list_with_local_promotable() -> [amqqueue:amqqueue()] | {error, term()}.
+
+-callback delete_member(VHost :: rabbit_types:vhost(), Name :: resource_name(), Node :: node()) -> ok | {error, term()}.
+
+-callback peek(Pos :: non_neg_integer(), QName :: rabbit_amqqueue:name()) -> {ok, [{binary(), term()}]} | {error, term()}.
+
+-callback status(QName :: rabbit_amqqueue:name()) -> [[{binary(), term()}]] | {error, term()}.
+
+-callback reclaim_memory(QName :: rabbit_amqqueue:name()) ->  ok | {error, term()}.
+
+-callback shrink_all(Node :: node()) -> [{rabbit_amqqueue:name(),
+                                          {ok, pos_integer()} | {error, pos_integer(), term()}}] | {error, term()}.
+
+-spec add_member(rabbit_types:vhost(), resource_name(), node(), membership(), timeout()) -> ok | {error, term()}.
+add_member(VHost, Name, Node, Membership, Timeout) ->
+    case rabbit_amqqueue:lookup(rabbit_misc:queue_resource(VHost, Name)) of
+        {ok, Q} ->
+            Type = amqqueue:get_type(Q),
+            Type:add_member(VHost, Name, Node, Membership, Timeout);
+        {error, not_found} = E ->
+            E
+    end.
+
+-spec list_with_local_promotable(atom() | binary()) -> [amqqueue:amqqueue()] | {error, term()}.
+list_with_local_promotable(TypeDescriptor) ->
+    case rabbit_queue_type:lookup(TypeDescriptor) of
+        {ok, Type} ->
+            {ok, Type:list_with_local_promotable()};
+        {error, not_found}  ->
+            {error, {unknown_queue_type, TypeDescriptor}}
+    end.
+
+-spec delete_member(rabbit_types:vhost(), resource_name(), node()) -> ok | {error, term()}.
+delete_member(VHost, Name, Node) ->
+    case rabbit_amqqueue:lookup(rabbit_misc:queue_resource(VHost, Name)) of
+        {ok, Q} ->
+            Type = amqqueue:get_type(Q),
+            Type:delete_member(VHost, Name, Node);
+        {error, not_found} = E ->
+            E
+    end.
+
+-spec grow(node(), binary(), binary(), all | even, membership()) ->
+          [{rabbit_amqqueue:name(),
+            {ok, pos_integer()} | {error, pos_integer(), term()}}].
+grow(Node, VhostSpec, QueueSpec, Strategy, Membership) ->
+    Running = rabbit_nodes:list_running(),
+    [begin
+         Size = length(amqqueue:get_nodes(Q)),
+         QName = amqqueue:get_name(Q),
+         rabbit_log:info("~ts: adding a new member (replica) on node ~w",
+                         [rabbit_misc:rs(QName), Node]),
+         Type = amqqueue:get_type(Q),
+         case Type:add_member(Q, Node, Membership) of
+             ok ->
+                 {QName, {ok, Size + 1}};
+             {error, Err} ->
+                 rabbit_log:warning(
+                   "~ts: failed to add member (replica) on node ~w, error: ~w",
+                   [rabbit_misc:rs(QName), Node, Err]),
+                 {QName, {error, Size, Err}}
+         end
+     end
+     || Q <- rabbit_amqqueue:list(),
+        rabbit_queue_type:is_replicable(Q),
+        %% don't add a member if there is already one on the node
+        not lists:member(Node, amqqueue:get_nodes(Q)),
+        %% node needs to be running
+        lists:member(Node, Running),
+        matches_strategy(Strategy, amqqueue:get_nodes(Q)),
+        is_match(amqqueue:get_vhost(Q), VhostSpec) andalso
+        is_match(get_resource_name(amqqueue:get_name(Q)), QueueSpec) ].
+
+-spec peek(Pos :: non_neg_integer(), QName :: rabbit_amqqueue:name()) -> {ok, [{binary(), term()}]} | {error, term()}.
+peek(Pos, QName) ->
+    case rabbit_amqqueue:lookup(QName) of
+        {ok, Q} ->
+            Type = amqqueue:get_type(Q),
+            Type:peek(Pos, QName);
+        {error, not_found} = E ->
+            E
+    end.
+
+-spec status(QName :: rabbit_amqqueue:name()) -> [[{binary(), term()}]] | {error, term()}.
+status(QName) ->
+    case rabbit_amqqueue:lookup(QName) of
+        {ok, Q} ->
+            Type = amqqueue:get_type(Q),
+            Type:status(QName);
+        {error, not_found} = E ->
+            E
+    end.
+
+-spec reclaim_memory(rabbit_types:vhost(), Name :: rabbit_misc:resource_name()) -> ok | {error, term()}.
+reclaim_memory(VHost, QueueName) ->
+    QName = #resource{virtual_host = VHost, name = QueueName, kind = queue},
+    case rabbit_amqqueue:lookup(QName) of
+        {ok, Q} ->
+            Type = amqqueue:get_type(Q),
+            Type:reclaim_memory(QName);
+        {error, not_found} = E ->
+            E
+    end.
+
+-spec shrink_all(Node :: node(), TypeDescriptor :: rabbit_registry:type_descriptor()) ->
+    {ok, pos_integer()} |
+    {error, pos_integer(), term()} |
+    {error, {unknown_queue_type, rabbit_registry:type_descriptor()}} |
+    {error, atom()}. %% to cover not_supported
+shrink_all(Node, <<"all">>) ->
+    lists:flatten([Type:shrink_all(Node) || Type <- rabbit_queue_type:list_replicable()]);
+shrink_all(Node, TypeDescriptor) ->
+    case rabbit_queue_type:lookup(TypeDescriptor) of
+        {ok, Type} ->
+            {ok, Type:shrink_all(Node)};
+        {error, not_found}  ->
+            {error, {unknown_queue_type, TypeDescriptor}}
+    end.
+
+matches_strategy(all, _) -> true;
+matches_strategy(even, Members) ->
+    length(Members) rem 2 == 0.
+
+is_match(Subj, E) ->
+   nomatch /= re:run(Subj, E).
+
+get_resource_name(#resource{name  = Name}) ->
+    Name.

deps/rabbit/src/rabbit_queue_type.erl

@@ -19,6 +19,7 @@
          init/0,
          close/1,
          discover/1,
+         lookup/1,
          short_alias_of/1,
          default/0,
          default_alias/0,
@@ -64,6 +65,7 @@
          can_redeliver/2,
          rebalance_module/1,
          is_replicable/1,
+         list_replicable/0,
          stop/1,
          list_with_minimum_quorum/0,
          drain/1,
@@ -313,6 +315,14 @@ discover(TypeDescriptor) ->
     {ok, TypeModule} = rabbit_registry:lookup_type_module(queue, TypeDescriptor),
     TypeModule.
 
+-spec lookup(binary() | atom()) -> {ok, queue_type()} | {error, not_found}.
+lookup(<<"undefined">>) ->
+    fallback();
+lookup(undefined) ->
+    fallback();
+lookup(TypeDescriptor) ->
+     rabbit_registry:lookup_type_module(queue, TypeDescriptor).
+
 -spec short_alias_of(TypeDescriptor) -> Ret when
       TypeDescriptor :: {utf8, binary()} | atom() | binary(),
       Ret :: binary() | undefined.
@@ -906,12 +916,18 @@ rebalance_module(Q) ->
     Capabilities = TypeModule:capabilities(),
     maps:get(rebalance_module, Capabilities, undefined).
 
--spec is_replicable(amqqueue:amqqueue()) -> undefine | module().
-is_replicable(Q) ->
+-spec is_replicable(amqqueue:amqqueue() | queue_type()) -> false | module().
+is_replicable(Q) when ?is_amqqueue(Q) ->
     TypeModule = amqqueue:get_type(Q),
+    is_replicable(TypeModule);
+is_replicable(TypeModule) when is_atom(TypeModule) ->
     Capabilities = TypeModule:capabilities(),
     maps:get(is_replicable, Capabilities, false).
 
+list_replicable() ->
+    _ = [TypeModule ||
+            {_Type, TypeModule} <- rabbit_registry:lookup_all(queue), is_replicable(TypeModule)].
+
 -spec stop(rabbit_types:vhost()) -> ok.
 stop(VHost) ->
     %% original rabbit_amqqueue:stop doesn't do any catches or try after

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -9,6 +9,7 @@
 -feature(maybe_expr, enable).
 
 -behaviour(rabbit_queue_type).
+-behaviour(rabbit_queue_commands).
 -behaviour(rabbit_policy_validator).
 -behaviour(rabbit_policy_merge_strategy).
 
@@ -31,7 +32,7 @@
 -export([purge/1]).
 -export([deliver/3]).
 -export([dead_letter_publish/5]).
--export([cluster_state/1, status/2]).
+-export([cluster_state/1, status/1]).
 -export([update_consumer_handler/8, update_consumer/9]).
 -export([cancel_consumer_handler/2, cancel_consumer/3]).
 -export([become_leader/2, handle_tick/3, spawn_deleter/1]).
@@ -55,14 +56,13 @@
 -export([transfer_leadership/2, get_replicas/1, queue_length/1]).
 -export([list_with_minimum_quorum/0,
          list_with_local_promotable/0,
-         list_with_local_promotable_for_cli/0,
          filter_quorum_critical/3,
          all_replica_states/0]).
 -export([capabilities/0]).
 -export([repair_amqqueue_nodes/1,
          repair_amqqueue_nodes/2
          ]).
--export([reclaim_memory/2,
+-export([reclaim_memory/1,
          wal_force_roll_over/1]).
 -export([notify_decorators/1,
          notify_decorators/3,
@@ -501,11 +501,6 @@ list_with_local_promotable() ->
     #{node() := ReplicaStates} = get_replica_states([node()]),
     filter_promotable(Queues, ReplicaStates).
 
--spec list_with_local_promotable_for_cli() -> [#{binary() => any()}].
-list_with_local_promotable_for_cli() ->
-    Qs = list_with_local_promotable(),
-    lists:map(fun amqqueue:to_printable/1, Qs).
-
 -spec get_replica_states([node()]) -> #{node() => replica_states()}.
 get_replica_states(Nodes) ->
     maps:from_list(
@@ -1246,11 +1241,10 @@ key_metrics_rpc(ServerId) ->
     Metrics = ra:key_metrics(ServerId),
     Metrics#{machine_version => rabbit_fifo:version()}.
 
--spec status(rabbit_types:vhost(), Name :: rabbit_misc:resource_name()) ->
+-spec status(rabbit_types:r(queue)) ->
     [[{binary(), term()}]] | {error, term()}.
-status(Vhost, QueueName) ->
+status(QName) ->
     %% Handle not found queues
-    QName = #resource{virtual_host = Vhost, name = QueueName, kind = queue},
     case rabbit_amqqueue:lookup(QName) of
         {ok, Q} when ?amqqueue_is_classic(Q) ->
             {error, classic_queue_not_supported};
@@ -1618,9 +1612,8 @@ matches_strategy(even, Members) ->
 is_match(Subj, E) ->
    nomatch /= re:run(Subj, E).
 
--spec reclaim_memory(rabbit_types:vhost(), Name :: rabbit_misc:resource_name()) -> ok | {error, term()}.
-reclaim_memory(Vhost, QueueName) ->
-    QName = #resource{virtual_host = Vhost, name = QueueName, kind = queue},
+-spec reclaim_memory(rabbit_amqqueue:name()) -> ok | {error, term()}.
+reclaim_memory(QName) ->
     case rabbit_amqqueue:lookup(QName) of
         {ok, Q} when ?amqqueue_is_classic(Q) ->
             {error, classic_queue_not_supported};

deps/rabbit/src/rabbit_stream_queue.erl

@@ -68,6 +68,9 @@
          queue_vm_stats_sups/0,
          queue_vm_ets/0]).
 
+%% queues commands
+-export([shrink_all/1]).
+
 -include_lib("rabbit_common/include/rabbit.hrl").
 -include("amqqueue.hrl").
 
@@ -1471,3 +1474,6 @@ queue_vm_stats_sups() ->
 queue_vm_ets() ->
     {[],
      []}.
+
+shrink_all(Node) ->
+    delete_all_replicas(Node).

deps/rabbit/src/rabbit_upgrade_preparation.erl

@@ -8,7 +8,7 @@
 -module(rabbit_upgrade_preparation).
 
 -export([await_online_quorum_plus_one/1,
-         list_with_minimum_quorum_for_cli/0]).
+         with_minimum_quorum/0]).
 
 %%
 %% API
@@ -56,8 +56,8 @@ endangered_critical_components() ->
 do_await_safe_online_quorum(0) ->
     false;
 do_await_safe_online_quorum(IterationsLeft) ->
-    EndangeredQueues = rabbit_queue_type:list_with_minimum_quorum(),
-    case EndangeredQueues =:= [] andalso endangered_critical_components() =:= [] of
+    {EndangeredQueues, EndangeredCriticalComponents} = with_minimum_quorum(),
+    case EndangeredQueues =:= [] andalso EndangeredCriticalComponents =:= [] of
         true -> true;
         false ->
             case IterationsLeft rem ?LOGGING_FREQUENCY of
@@ -67,7 +67,7 @@ do_await_safe_online_quorum(IterationsLeft) ->
                         N -> rabbit_log:info("Waiting for ~p queues and streams to have quorum+1 replicas online. "
                                              "You can list them with `rabbitmq-diagnostics check_if_node_is_quorum_critical`", [N])
                     end,
-                    case endangered_critical_components() of
+                    case EndangeredCriticalComponents of
                         [] -> ok;
                         _ -> rabbit_log:info("Waiting for the following critical components to have quorum+1 replicas online: ~p.",
                                              [endangered_critical_components()])
@@ -79,13 +79,8 @@ do_await_safe_online_quorum(IterationsLeft) ->
             do_await_safe_online_quorum(IterationsLeft - 1)
     end.
 
--spec list_with_minimum_quorum_for_cli() -> [#{binary() => term()}].
-list_with_minimum_quorum_for_cli() ->
+-spec with_minimum_quorum() -> {[amqqueue:amqqueue()], [atom()]}.
+with_minimum_quorum() ->
     EndangeredQueues = rabbit_queue_type:list_with_minimum_quorum(),
-    [amqqueue:to_printable(Q) || Q <- EndangeredQueues] ++
-    [#{
-           <<"readable_name">> => C,
-           <<"name">> => C,
-           <<"virtual_host">> => <<"(not applicable)">>,
-           <<"type">> => process
-      } || C <- endangered_critical_components()].
+    EndangeredCriticalComponents = endangered_critical_components(),
+    {EndangeredQueues, EndangeredCriticalComponents}.