Extend QQ grow command to support target quorum cluster size

Author: Ayanda-D

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -1543,29 +1543,20 @@ shrink_all(Node) ->
             amqqueue:get_type(Q) == ?MODULE,
             lists:member(Node, get_nodes(Q))].
 
-
+-spec grow(node() | integer(), binary(), binary(), all | even) ->
+    [{rabbit_amqqueue:name(),
+      {ok, pos_integer()} | {error, pos_integer(), term()}}].
 grow(Node, VhostSpec, QueueSpec, Strategy) ->
     grow(Node, VhostSpec, QueueSpec, Strategy, promotable).
 
--spec grow(node(), binary(), binary(), all | even, membership()) ->
+-spec grow(node() | integer(), binary(), binary(), all | even, membership()) ->
     [{rabbit_amqqueue:name(),
       {ok, pos_integer()} | {error, pos_integer(), term()}}].
-grow(Node, VhostSpec, QueueSpec, Strategy, Membership) ->
+grow(Node, VhostSpec, QueueSpec, Strategy, Membership) when is_atom(Node) ->
     Running = rabbit_nodes:list_running(),
     [begin
          Size = length(get_nodes(Q)),
-         QName = amqqueue:get_name(Q),
-         ?LOG_INFO("~ts: adding a new member (replica) on node ~w",
-                         [rabbit_misc:rs(QName), Node]),
-         case add_member(Q, Node, Membership) of
-             ok ->
-                 {QName, {ok, Size + 1}};
-             {error, Err} ->
-                 ?LOG_WARNING(
-                   "~ts: failed to add member (replica) on node ~w, error: ~w",
-                   [rabbit_misc:rs(QName), Node, Err]),
-                 {QName, {error, Size, Err}}
-         end
+         maybe_grow(Q, Node, Membership, Size)
      end
      || Q <- rabbit_amqqueue:list(),
         amqqueue:get_type(Q) == ?MODULE,
@@ -1575,7 +1566,53 @@ grow(Node, VhostSpec, QueueSpec, Strategy, Membership) ->
         lists:member(Node, Running),
         matches_strategy(Strategy, get_nodes(Q)),
         is_match(amqqueue:get_vhost(Q), VhostSpec) andalso
-        is_match(get_resource_name(amqqueue:get_name(Q)), QueueSpec) ].
+        is_match(get_resource_name(amqqueue:get_name(Q)), QueueSpec) ];
+
+grow(QuorumClusterSize, VhostSpec, QueueSpec, Strategy, Membership)
+  when is_integer(QuorumClusterSize) ->
+    Running = rabbit_nodes:list_running(),
+    TotalRunning = length(Running),
+
+    TargetQuorumClusterSize =
+        if QuorumClusterSize > TotalRunning ->
+            %% we cant grow beyond total running nodes
+            TotalRunning;
+        true ->
+            QuorumClusterSize
+        end,
+
+    lists:flatten(
+        [begin
+            QNodes = get_nodes(Q),
+            case length(QNodes) of
+                Size when Size < TargetQuorumClusterSize ->
+                    TargetAvailableNodes = Running -- QNodes,
+                    Node = hd(TargetAvailableNodes),
+                    maybe_grow(Q, Node, Membership, Size);
+                _ ->
+                    []
+            end
+        end
+        ||  _ <- lists:seq(1, TargetQuorumClusterSize),
+            Q <- rabbit_amqqueue:list(),
+            amqqueue:get_type(Q) == ?MODULE,
+            matches_strategy(Strategy, get_nodes(Q)),
+            is_match(amqqueue:get_vhost(Q), VhostSpec) andalso
+            is_match(get_resource_name(amqqueue:get_name(Q)), QueueSpec)]).
+
+maybe_grow(Q, Node, Membership, Size) ->
+    QName = amqqueue:get_name(Q),
+    ?LOG_INFO("~ts: adding a new member (replica) on node ~w",
+                    [rabbit_misc:rs(QName), Node]),
+    case add_member(Q, Node, Membership) of
+        ok ->
+            {QName, {ok, Size + 1}};
+        {error, Err} ->
+            ?LOG_WARNING(
+            "~ts: failed to add member (replica) on node ~w, error: ~w",
+            [rabbit_misc:rs(QName), Node, Err]),
+            {QName, {error, Size, Err}}
+    end.
 
 -spec transfer_leadership(amqqueue:amqqueue(), node()) -> {migrated, node()} | {not_migrated, atom()}.
 transfer_leadership(Q, Destination) ->

deps/rabbitmq_cli/lib/rabbitmq/cli/queues/commands/grow_command.ex

@@ -39,6 +39,11 @@ defmodule RabbitMQ.CLI.Queues.Commands.GrowCommand do
     {:validation_failure, "strategy '#{s}' is not recognised."}
   end
 
+  def validate([n, _], _)
+      when (is_integer(n) and n <= 0) do
+    {:validation_failure, "target quorum cluster size '#{n}' must be greater than 0."}
+  end
+
   def validate(_, %{membership: m})
       when not (m == "promotable" or
                   m == "non_voter" or
@@ -60,14 +65,22 @@ defmodule RabbitMQ.CLI.Queues.Commands.GrowCommand do
     )
   end
 
-  def run([node, strategy], %{
+  def run([node_or_quorum_cluster_size, strategy], %{
         node: node_name,
         vhost_pattern: vhost_pat,
         queue_pattern: queue_pat,
         membership: membership,
         errors_only: errors_only
       }) do
-    args = [to_atom(node), vhost_pat, queue_pat, to_atom(strategy)]
+
+    node_or_quorum_cluster_size =
+      if is_integer(node_or_quorum_cluster_size) do
+        node_or_quorum_cluster_size
+      else
+        to_atom(node_or_quorum_cluster_size)
+      end
+
+    args = [node_or_quorum_cluster_size, vhost_pat, queue_pat, to_atom(strategy)]
 
     args =
       case to_atom(membership) do
@@ -108,11 +121,11 @@ defmodule RabbitMQ.CLI.Queues.Commands.GrowCommand do
 
   def usage,
     do:
-      "grow <node> <all | even> [--vhost-pattern <pattern>] [--queue-pattern <pattern>] [--membership <promotable|voter>]"
+      "grow <node | quorum_cluster_size> <all | even> [--vhost-pattern <pattern>] [--queue-pattern <pattern>] [--membership <promotable|voter>]"
 
   def usage_additional do
     [
-      ["<node>", "node name to place replicas on"],
+      ["<node | quorum_cluster_size>", "node name to place replicas on or desired quorum cluster size"],
       [
         "<all | even>",
         "add a member for all matching queues or just those whose membership count is an even number"