extend CLI & HTTP API to pass membership

Author: Alex Valiushko

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -48,7 +48,7 @@
 -export([format_ra_event/3]).
 -export([cleanup_data_dir/0]).
 -export([shrink_all/1,
-         grow/4]).
+         grow/5]).
 -export([transfer_leadership/2, get_replicas/1, queue_length/1]).
 -export([file_handle_leader_reservation/1,
          file_handle_other_reservation/0]).
@@ -86,6 +86,7 @@
 -type msg_id() :: non_neg_integer().
 -type qmsg() :: {rabbit_types:r('queue'), pid(), msg_id(), boolean(),
                  mc:state()}.
+-type membership() :: voter | non_voter | promotable.  %% see ra_membership() in Ra.
 
 -define(RA_SYSTEM, quorum_queues).
 -define(RA_WAL_NAME, ra_log_wal).
@@ -1179,10 +1180,10 @@ add_member(VHost, Name, Node, VoterStatus, Timeout) ->
 add_member(Q, Node) ->
     add_member(Q, Node, promotable).
 
-add_member(Q, Node, VoterStatus) ->
-    add_member(Q, Node, VoterStatus, ?ADD_MEMBER_TIMEOUT).
+add_member(Q, Node, Membership) ->
+    add_member(Q, Node, Membership, ?ADD_MEMBER_TIMEOUT).
 
-add_member(Q, Node, VoterStatus, Timeout) when ?amqqueue_is_quorum(Q) ->
+add_member(Q, Node, Membership, Timeout) when ?amqqueue_is_quorum(Q) ->
     {RaName, _} = amqqueue:get_pid(Q),
     QName = amqqueue:get_name(Q),
     %% TODO parallel calls might crash this, or add a duplicate in quorum_nodes
@@ -1192,7 +1193,7 @@ add_member(Q, Node, VoterStatus, Timeout) when ?amqqueue_is_quorum(Q) ->
                                       ?TICK_TIMEOUT),
     SnapshotInterval = application:get_env(rabbit, quorum_snapshot_interval,
                                            ?SNAPSHOT_INTERVAL),
-    Conf = make_ra_conf(Q, ServerId, TickTimeout, SnapshotInterval, VoterStatus),
+    Conf = make_ra_conf(Q, ServerId, TickTimeout, SnapshotInterval, Membership),
     case ra:start_server(?RA_SYSTEM, Conf) of
         ok ->
             ServerIdSpec = maps:with([id, uid, membership], Conf),
@@ -1304,17 +1305,17 @@ shrink_all(Node) ->
             amqqueue:get_type(Q) == ?MODULE,
             lists:member(Node, get_nodes(Q))].
 
--spec grow(node(), binary(), binary(), all | even) ->
+-spec grow(node(), binary(), binary(), all | even, membership()) ->
     [{rabbit_amqqueue:name(),
       {ok, pos_integer()} | {error, pos_integer(), term()}}].
- grow(Node, VhostSpec, QueueSpec, Strategy) ->
+ grow(Node, VhostSpec, QueueSpec, Membership, Strategy) ->
     Running = rabbit_nodes:list_running(),
     [begin
          Size = length(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]),
-         case add_member(Q, Node) of
+         case add_member(Q, Node, Membership) of
              ok ->
                  {QName, {ok, Size + 1}};
              {error, Err} ->

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

@@ -10,21 +10,45 @@ defmodule RabbitMQ.CLI.Queues.Commands.AddMemberCommand do
 
   @behaviour RabbitMQ.CLI.CommandBehaviour
 
-  @default_timeout 5_000
+  defp default_opts, do: %{vhost: "/",
+                           membership: "promotable",
+                           timeout: 5_000}
 
   def merge_defaults(args, opts) do
-    timeout =
-      case opts[:timeout] do
-        nil -> @default_timeout
-        :infinity -> @default_timeout
-        other -> other
-      end
-
-    {args, Map.merge(%{vhost: "/", timeout: timeout}, opts)}
+    default = default_opts()
+    opts = Map.update(opts, :timeout, :infinity,
+                       &(case &1 do
+                         :infinity -> default.timeout
+                         other -> other
+                       end))
+    {args, Map.merge(default, opts)}
   end
 
-  use RabbitMQ.CLI.Core.AcceptsDefaultSwitchesAndTimeout
-  use RabbitMQ.CLI.Core.AcceptsTwoPositionalArguments
+  def switches(),
+    do: [
+      timeout: :integer,
+      membership: :string
+    ]
+
+  def aliases(), do: [t: :timeout]
+
+  def validate(args, _) when length(args) < 2 do
+    {:validation_failure, :not_enough_args}
+  end
+
+  def validate(args, _) when length(args) > 2 do
+    {:validation_failure, :too_many_args}
+  end
+
+  def validate(_, %{membership: m}) when not (m == "promotable" or
+                                              m == "non_voter" or
+                                              m == "voter") do
+    {:validation_failure, "voter status '#{m}' is not recognised."}
+  end
+
+  def validate(_, _) do
+    :ok
+  end
 
   def validate_execution_environment(args, opts) do
     Validators.chain(
@@ -39,11 +63,13 @@ defmodule RabbitMQ.CLI.Queues.Commands.AddMemberCommand do
     )
   end
 
-  def run([name, node] = _args, %{vhost: vhost, node: node_name, timeout: timeout}) do
+  def run([name, node] = _args,
+          %{vhost: vhost, node: node_name, timeout: timeout, membership: membership}) do
     case :rabbit_misc.rpc_call(node_name, :rabbit_quorum_queue, :add_member, [
            vhost,
            name,
            to_atom(node),
+           to_atom(membership),
            timeout
          ]) do
       {:error, :classic_queue_not_supported} ->
@@ -59,12 +85,13 @@ defmodule RabbitMQ.CLI.Queues.Commands.AddMemberCommand do
 
   use RabbitMQ.CLI.DefaultOutput
 
-  def usage, do: "add_member [--vhost <vhost>] <queue> <node>"
+  def usage, do: "add_member [--vhost <vhost>] <queue> <node> [--membership <promotable|voter>]"
 
   def usage_additional do
     [
       ["<queue>", "quorum queue name"],
-      ["<node>", "node to add a new replica on"]
+      ["<node>", "node to add a new replica on"],
+      ["--membership <promotable|voter>", "add a promotable non-voter (default) or full voter"]
     ]
   end
 

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

@@ -10,12 +10,16 @@ defmodule RabbitMQ.CLI.Queues.Commands.GrowCommand do
 
   @behaviour RabbitMQ.CLI.CommandBehaviour
 
-  defp default_opts, do: %{vhost_pattern: ".*", queue_pattern: ".*", errors_only: false}
+  defp default_opts, do: %{vhost_pattern: ".*",
+                           queue_pattern: ".*",
+                           membership: "promotable",
+                           errors_only: false}
 
   def switches(),
     do: [
       vhost_pattern: :string,
       queue_pattern: :string,
+      membership: :string,
       errors_only: :boolean
     ]
 
@@ -31,17 +35,19 @@ defmodule RabbitMQ.CLI.Queues.Commands.GrowCommand do
     {:validation_failure, :too_many_args}
   end
 
-  def validate([_, s], _) do
-    case s do
-      "all" ->
-        :ok
+  def validate([_, s], _) when not (s == "all" or
+                                    s == "even") do
+    {:validation_failure, "strategy '#{s}' is not recognised."}
+  end
 
-      "even" ->
-        :ok
+  def validate(_, %{membership: m}) when not (m == "promotable" or
+                                              m == "non_voter" or
+                                              m == "voter") do
+    {:validation_failure, "voter status '#{m}' is not recognised."}
+  end
 
-      _ ->
-        {:validation_failure, "strategy '#{s}' is not recognised."}
-    end
+  def validate(_, _) do
+    :ok
   end
 
   def validate_execution_environment(args, opts) do
@@ -58,12 +64,14 @@ defmodule RabbitMQ.CLI.Queues.Commands.GrowCommand do
         node: node_name,
         vhost_pattern: vhost_pat,
         queue_pattern: queue_pat,
+        membership: membership,
         errors_only: errors_only
       }) do
     case :rabbit_misc.rpc_call(node_name, :rabbit_quorum_queue, :grow, [
            to_atom(node),
            vhost_pat,
            queue_pat,
+           to_atom(membership),
            to_atom(strategy)
          ]) do
       {:error, _} = error ->
@@ -97,7 +105,7 @@ defmodule RabbitMQ.CLI.Queues.Commands.GrowCommand do
   def formatter(), do: RabbitMQ.CLI.Formatters.Table
 
   def usage,
-    do: "grow <node> <all | even> [--vhost-pattern <pattern>] [--queue-pattern <pattern>]"
+    do: "grow <node> <all | even> [--vhost-pattern <pattern>] [--queue-pattern <pattern>] [--membership <promotable|voter>]"
 
   def usage_additional do
     [
@@ -108,6 +116,7 @@ defmodule RabbitMQ.CLI.Queues.Commands.GrowCommand do
       ],
       ["--queue-pattern <pattern>", "regular expression to match queue names"],
       ["--vhost-pattern <pattern>", "regular expression to match virtual host names"],
+      ["--membership <promotable|voter>", "add a promotable non-voter (default) or full voter"],
       ["--errors-only", "only list queues which reported an error"]
     ]
   end

deps/rabbitmq_cli/test/queues/add_member_command_test.exs

@@ -20,6 +20,8 @@ defmodule RabbitMQ.CLI.Queues.Commands.AddMemberCommandTest do
     {:ok,
      opts: %{
        node: get_rabbit_hostname(),
+       membership: "voter",
+       vhost: "/",
        timeout: context[:test_timeout] || 30000
      }}
   end
@@ -42,17 +44,35 @@ defmodule RabbitMQ.CLI.Queues.Commands.AddMemberCommandTest do
            ) == {:validation_failure, :too_many_args}
   end
 
+  test "validate: when membership promotable is provided, returns a success" do
+    assert @command.validate(["quorum-queue-a", "rabbit@new-node"], %{membership: "promotable"}) == :ok
+  end
+
+  test "validate: when membership voter is provided, returns a success" do
+    assert @command.validate(["quorum-queue-a", "rabbit@new-node"], %{membership: "voter"}) == :ok
+  end
+
+  test "validate: when membership non_voter is provided, returns a success" do
+    assert @command.validate(["quorum-queue-a", "rabbit@new-node"], %{membership: "non_voter"}) == :ok
+  end
+
+  test "validate: when wrong membership is provided, returns failure" do
+    assert @command.validate(["quorum-queue-a", "rabbit@new-node"], %{membership: "banana"}) ==
+             {:validation_failure, "voter status 'banana' is not recognised."}
+  end
+
+
   test "validate: treats two positional arguments and default switches as a success" do
     assert @command.validate(["quorum-queue-a", "rabbit@new-node"], %{}) == :ok
   end
 
   @tag test_timeout: 3000
-  test "run: targeting an unreachable node throws a badrpc" do
+  test "run: targeting an unreachable node throws a badrpc", context do
     assert match?(
              {:badrpc, _},
              @command.run(
                ["quorum-queue-a", "rabbit@new-node"],
-               %{node: :jake@thedog, vhost: "/", timeout: 200}
+               Map.merge(context[:opts], %{node: :jake@thedog})
              )
            )
   end

deps/rabbitmq_cli/test/queues/grow_command_test.exs

@@ -23,13 +23,14 @@ defmodule RabbitMQ.CLI.Queues.Commands.GrowCommandTest do
        timeout: context[:test_timeout] || 30000,
        vhost_pattern: ".*",
        queue_pattern: ".*",
+       membership: "promotable",
        errors_only: false
      }}
   end
 
   test "merge_defaults: defaults to reporting complete results" do
     assert @command.merge_defaults([], %{}) ==
-             {[], %{vhost_pattern: ".*", queue_pattern: ".*", errors_only: false}}
+             {[], %{vhost_pattern: ".*", queue_pattern: ".*", errors_only: false, membership: "promotable"}}
   end
 
   test "validate: when no arguments are provided, returns a failure" do
@@ -58,13 +59,30 @@ defmodule RabbitMQ.CLI.Queues.Commands.GrowCommandTest do
              {:validation_failure, :too_many_args}
   end
 
+  test "validate: when membership promotable is provided, returns a success" do
+    assert @command.validate(["quorum-queue-a", "all"], %{membership: "promotable"}) == :ok
+  end
+
+  test "validate: when membership voter is provided, returns a success" do
+    assert @command.validate(["quorum-queue-a", "all"], %{membership: "voter"}) == :ok
+  end
+
+  test "validate: when membership non_voter is provided, returns a success" do
+    assert @command.validate(["quorum-queue-a", "all"], %{membership: "non_voter"}) == :ok
+  end
+
+  test "validate: when wrong membership is provided, returns failure" do
+    assert @command.validate(["quorum-queue-a", "all"], %{membership: "banana"}) ==
+             {:validation_failure, "voter status 'banana' is not recognised."}
+  end
+
   @tag test_timeout: 3000
   test "run: targeting an unreachable node throws a badrpc", context do
     assert match?(
              {:badrpc, _},
              @command.run(
                ["quorum-queue-a", "all"],
-               Map.merge(context[:opts], %{node: :jake@thedog, timeout: 200})
+               Map.merge(context[:opts], %{node: :jake@thedog})
              )
            )
   end

deps/rabbitmq_management/src/rabbit_mgmt_wm_quorum_queue_replicas_add_member.erl

@@ -38,11 +38,12 @@ accept_content(ReqData, Context) ->
   QName = rabbit_mgmt_util:id(queue, ReqData),
   Res = rabbit_mgmt_util:with_decode(
     [node], ReqData, Context,
-    fun([NewReplicaNode], _Body, _ReqData) ->
+    fun([NewReplicaNode], Body, _ReqData) ->
+      Membership = maps:get(<<"membership">>, Body, promotable),
       rabbit_amqqueue:with(
         rabbit_misc:r(VHost, queue, QName),
         fun(_Q) ->
-          rabbit_quorum_queue:add_member(VHost, QName, rabbit_data_coercion:to_atom(NewReplicaNode), promotable, ?TIMEOUT)
+          rabbit_quorum_queue:add_member(VHost, QName, rabbit_data_coercion:to_atom(NewReplicaNode), Membership, ?TIMEOUT)
         end)
     end),
   case Res of

deps/rabbitmq_management/src/rabbit_mgmt_wm_quorum_queue_replicas_grow.erl

@@ -39,11 +39,13 @@ accept_content(ReqData, Context) ->
   NewReplicaNode = rabbit_mgmt_util:id(node, ReqData),
   rabbit_mgmt_util:with_decode(
     [vhost_pattern, queue_pattern, strategy], ReqData, Context,
-    fun([VHPattern, QPattern, Strategy], _Body, _ReqData) ->
+    fun([VHPattern, QPattern, Strategy], Body, _ReqData) ->
+      Membership = maps:get(<<"membership">>, Body, promotable),
       rabbit_quorum_queue:grow(
         rabbit_data_coercion:to_atom(NewReplicaNode),
         VHPattern,
         QPattern,
+        rabbit_data_coercion:to_atom(Membership),
         rabbit_data_coercion:to_atom(Strategy))
     end),
   {true, ReqData, Context}.