Merge pull request #10426 from rabbitmq/amazon-mq-global-quorum-critical

New upgrade time QQ health check: add check_if_new_quorum_queue_replicas_have_finished_initial_sync by @illotum (plus a test)

Author: michaelklishin

deps/rabbit/src/amqqueue.erl

@@ -84,6 +84,7 @@
          reset_mirroring_and_decorators/1,
          set_immutable/1,
          qnode/1,
+         to_printable/1,
          macros/0]).
 
 -define(record_version, amqqueue_v2).
@@ -641,6 +642,14 @@ qnode(none) ->
 qnode({_, Node}) ->
     Node.
 
+-spec to_printable(amqqueue()) -> #{binary() => any()}.
+to_printable(#amqqueue{name = QName = #resource{name = Name},
+                       vhost = VHost, type = Type}) ->
+     #{<<"readable_name">> => rabbit_data_coercion:to_binary(rabbit_misc:rs(QName)),
+       <<"name">> => Name,
+       <<"virtual_host">> => VHost,
+       <<"type">> => Type}.
+
 % private
 
 macros() ->

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -55,7 +55,8 @@
          file_handle_other_reservation/0]).
 -export([file_handle_release_reservation/0]).
 -export([list_with_minimum_quorum/0,
-         filter_quorum_critical/1,
+         list_with_local_promotable/0,
+         list_with_local_promotable_for_cli/0,
          filter_quorum_critical/3,
          all_replica_states/0]).
 -export([capabilities/0]).
@@ -77,6 +78,10 @@
 -export([force_shrink_member_to_current_member/2,
          force_all_queues_shrink_member_to_current_member/0]).
 
+-ifdef(TEST).
+-export([filter_promotable/2]).
+-endif.
+
 -import(rabbit_queue_type_util, [args_policy_lookup/3,
                                  qname_to_internal_name/1,
                                  erpc_call/5]).
@@ -89,6 +94,8 @@
 -type qmsg() :: {rabbit_types:r('queue'), pid(), msg_id(), boolean(),
                  mc:state()}.
 -type membership() :: voter | non_voter | promotable.  %% see ra_membership() in Ra.
+-type replica_states() :: #{atom() => replica_state()}.
+-type replica_state() :: leader | follower | non_voter | promotable.
 
 -define(RA_SYSTEM, quorum_queues).
 -define(RA_WAL_NAME, ra_log_wal).
@@ -405,35 +412,37 @@ all_replica_states() ->
 
 -spec list_with_minimum_quorum() -> [amqqueue:amqqueue()].
 list_with_minimum_quorum() ->
-    filter_quorum_critical(
-      rabbit_amqqueue:list_local_quorum_queues()).
-
--spec filter_quorum_critical([amqqueue:amqqueue()]) -> [amqqueue:amqqueue()].
-filter_quorum_critical(Queues) ->
-    %% Example map of QQ replica states:
-    %%    #{rabbit@warp10 =>
-    %%      #{'%2F_qq.636' => leader,'%2F_qq.243' => leader,
-    %%        '%2F_qq.1939' => leader,'%2F_qq.1150' => leader,
-    %%        '%2F_qq.1109' => leader,'%2F_qq.1654' => leader,
-    %%        '%2F_qq.1679' => leader,'%2F_qq.1003' => leader,
-    %%        '%2F_qq.1593' => leader,'%2F_qq.1765' => leader,
-    %%        '%2F_qq.933' => leader,'%2F_qq.38' => leader,
-    %%        '%2F_qq.1357' => leader,'%2F_qq.1345' => leader,
-    %%        '%2F_qq.1694' => leader,'%2F_qq.994' => leader,
-    %%        '%2F_qq.490' => leader,'%2F_qq.1704' => leader,
-    %%        '%2F_qq.58' => leader,'%2F_qq.564' => leader,
-    %%        '%2F_qq.683' => leader,'%2F_qq.386' => leader,
-    %%        '%2F_qq.753' => leader,'%2F_qq.6' => leader,
-    %%        '%2F_qq.1590' => leader,'%2F_qq.1363' => leader,
-    %%        '%2F_qq.882' => leader,'%2F_qq.1161' => leader,...}}
-    ReplicaStates = maps:from_list(
-                        rabbit_misc:append_rpc_all_nodes(rabbit_nodes:list_running(),
-                            ?MODULE, all_replica_states, [])),
+    Queues = rabbit_amqqueue:list_local_quorum_queues(),
+    ReplicaStates = get_replica_states(rabbit_nodes:list_running()),
     filter_quorum_critical(Queues, ReplicaStates, node()).
 
--spec filter_quorum_critical([amqqueue:amqqueue()], #{node() => #{atom() => atom()}}, node()) ->
+-spec list_with_local_promotable() -> [amqqueue:amqqueue()].
+list_with_local_promotable() ->
+    Queues = rabbit_amqqueue:list_local_quorum_queues(),
+    #{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(
+      rabbit_misc:append_rpc_all_nodes(Nodes, ?MODULE, all_replica_states, [])).
+
+-spec filter_promotable([amqqueue:amqqueue()], replica_states()) ->
     [amqqueue:amqqueue()].
+filter_promotable(Queues, ReplicaStates) ->
+    lists:filter(fun (Q) ->
+                    {RaName, _Node} = amqqueue:get_pid(Q),
+                    State = maps:get(RaName, ReplicaStates),
+                    State == promotable
+                 end, Queues).
 
+-spec filter_quorum_critical([amqqueue:amqqueue()], #{node() => replica_states()}, node()) ->
+    [amqqueue:amqqueue()].
 filter_quorum_critical(Queues, ReplicaStates, Self) ->
     lists:filter(fun (Q) ->
                     MemberNodes = get_nodes(Q),

deps/rabbit/src/rabbit_upgrade_preparation.erl

@@ -85,15 +85,7 @@ list_with_minimum_quorum_for_cli() ->
     EndangeredQueues = lists:append(
                          rabbit_quorum_queue:list_with_minimum_quorum(),
                          rabbit_stream_queue:list_with_minimum_quorum()),
-    [begin
-         #resource{name = Name} = QName = amqqueue:get_name(Q),
-         #{
-           <<"readable_name">> => rabbit_data_coercion:to_binary(rabbit_misc:rs(QName)),
-           <<"name">> => Name,
-           <<"virtual_host">> => amqqueue:get_vhost(Q),
-           <<"type">> => amqqueue:get_type(Q)
-          }
-     end || Q <- EndangeredQueues] ++
+    [amqqueue:to_printable(Q) || Q <- EndangeredQueues] ++
     [#{
            <<"readable_name">> => C,
            <<"name">> => C,

deps/rabbit/test/unit_quorum_queue_SUITE.erl

@@ -5,6 +5,7 @@
 all() ->
     [
      all_replica_states_includes_nonvoters,
+     filter_nonvoters,
      filter_quorum_critical_accounts_nonvoters
     ].
 
@@ -27,6 +28,24 @@ filter_quorum_critical_accounts_nonvoters(_Config) ->
     [Q1] = rabbit_quorum_queue:filter_quorum_critical(Qs, Ss, test@follower2),
     ok.
 
+filter_nonvoters(_Config) ->
+    Qs = [_, _, _, Q4] =
+           [amqqueue:new(rabbit_misc:r(<<"/">>, queue, <<"q1">>),
+                        {q1, test@leader},
+                        false, false, none, [], undefined, #{}),
+            amqqueue:new(rabbit_misc:r(<<"/">>, queue, <<"q2">>),
+                        {q2, test@leader},
+                        false, false, none, [], undefined, #{}),
+            amqqueue:new(rabbit_misc:r(<<"/">>, queue, <<"q3">>),
+                        {q3, test@leader},
+                        false, false, none, [], undefined, #{}),
+            amqqueue:new(rabbit_misc:r(<<"/">>, queue, <<"q4">>),
+                        {q4, test@leader},
+                        false, false, none, [], undefined, #{})],
+    Ss = #{q1 => leader, q2 => follower, q3 => non_voter, q4 => promotable},
+    [Q4] = rabbit_quorum_queue:filter_promotable(Qs, Ss),
+    ok.
+
 all_replica_states_includes_nonvoters(_Config) ->
     ets:new(ra_state, [named_table, public, {write_concurrency, true}]),
     ets:insert(ra_state, [

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

@@ -0,0 +1,95 @@
+## This Source Code Form is subject to the terms of the Mozilla Public
+## License, v. 2.0. If a copy of the MPL was not distributed with this
+## file, You can obtain one at https://mozilla.org/MPL/2.0/.
+##
+## Copyright (c) 2007-2023 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries.  All rights reserved.
+
+defmodule RabbitMQ.CLI.Queues.Commands.CheckIfNewQuorumQueueReplicasHaveFinishedInitialSyncCommand do
+  @moduledoc """
+  Exits with a non-zero code if there are quorum queues
+  that run "non-voter" (not yet done with their initial sync, promotable to voters)
+  replicas on the current node.
+
+  This command is used to verify if a new cluster node hosts only
+  fully synchronized.
+  """
+
+  @behaviour RabbitMQ.CLI.CommandBehaviour
+
+  import RabbitMQ.CLI.Core.Platform, only: [line_separator: 0]
+
+  def scopes(), do: [:diagnostics, :queues]
+
+  use RabbitMQ.CLI.Core.AcceptsDefaultSwitchesAndTimeout
+  use RabbitMQ.CLI.Core.MergesNoDefaults
+  use RabbitMQ.CLI.Core.AcceptsNoPositionalArguments
+  use RabbitMQ.CLI.Core.RequiresRabbitAppRunning
+
+  def run([], %{node: node_name, timeout: timeout}) do
+    case :rabbit_misc.rpc_call(
+           node_name,
+           :rabbit_quorum_queue,
+           :list_with_local_promotable_for_cli,
+           [],
+           timeout
+         ) do
+      [] -> {:ok, []}
+      qs when is_list(qs) -> {:ok, qs}
+      other -> other
+    end
+  end
+
+  def output({:ok, []}, %{formatter: "json"}) do
+    {:ok, %{"result" => "ok"}}
+  end
+
+  def output({:ok, []}, %{silent: true}) do
+    {:ok, :check_passed}
+  end
+
+  def output({:ok, []}, %{node: node_name}) do
+    {:ok, "Node #{node_name} reported no queues with promotable replicas"}
+  end
+
+  def output({:ok, qs}, %{node: node_name, formatter: "json"}) when is_list(qs) do
+    {:error, :check_failed,
+     %{
+       "result" => "error",
+       "queues" => qs,
+       "message" => "Node #{node_name} reported local queues promotable replicas"
+     }}
+  end
+
+  def output({:ok, qs}, %{silent: true}) when is_list(qs) do
+    {:error, :check_failed}
+  end
+
+  def output({:ok, qs}, %{node: node_name}) when is_list(qs) do
+    lines = queue_lines(qs, node_name)
+
+    {:error, :check_failed, Enum.join(lines, line_separator())}
+  end
+
+  use RabbitMQ.CLI.DefaultOutput
+
+  def help_section(), do: :observability_and_health_checks
+
+  def description() do
+    "Health check that exits with a non-zero code if there are queues " <>
+      "that run promotable replicas on the current node."
+  end
+
+  def usage, do: "check_if_new_quorum_queue_replicas_have_finished_initial_sync"
+
+  def banner([], %{node: node_name}) do
+    "Checking if node #{node_name} runs promotable replicas of any queues ..."
+  end
+
+  #
+  # Implementation
+  #
+
+  def queue_lines(qs, node_name) do
+    for q <- qs, do: "#{q["readable_name"]} hasn't finished synchronization with #{node_name}."
+  end
+end

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

@@ -2,7 +2,7 @@
 ## License, v. 2.0. If a copy of the MPL was not distributed with this
 ## file, You can obtain one at https://mozilla.org/MPL/2.0/.
 ##
-## Copyright (c) 2007-2023 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries.  All rights reserved.
+## Copyright (c) 2007-2024 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries.  All rights reserved.
 
 defmodule RabbitMQ.CLI.Queues.Commands.CheckIfNodeIsMirrorSyncCriticalCommand do
   @moduledoc """

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

@@ -2,7 +2,7 @@
 ## License, v. 2.0. If a copy of the MPL was not distributed with this
 ## file, You can obtain one at https://mozilla.org/MPL/2.0/.
 ##
-## Copyright (c) 2007-2023 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries.  All rights reserved.
+## Copyright (c) 2007-2024 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries.  All rights reserved.
 
 defmodule RabbitMQ.CLI.Queues.Commands.CheckIfNodeIsQuorumCriticalCommand do
   @moduledoc """

deps/rabbitmq_cli/test/queues/check_if_new_quorum_queue_replicas_have_finished_initial_sync_test.exs

@@ -0,0 +1,45 @@
+## This Source Code Form is subject to the terms of the Mozilla Public
+## License, v. 2.0. If a copy of the MPL was not distributed with this
+## file, You can obtain one at https://mozilla.org/MPL/2.0/.
+##
+## Copyright (c) 2007-2023 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries.  All rights reserved.
+
+defmodule RabbitMQ.CLI.Queues.Commands.CheckIfNewQuorumQueueReplicasHaveFinishedInitialSyncCommandTest do
+  use ExUnit.Case, async: false
+  import TestHelper
+
+  @command RabbitMQ.CLI.Queues.Commands.CheckIfNewQuorumQueueReplicasHaveFinishedInitialSyncCommand
+
+  setup_all do
+    RabbitMQ.CLI.Core.Distribution.start()
+
+    :ok
+  end
+
+  setup context do
+    {:ok,
+     opts: %{
+       node: get_rabbit_hostname(),
+       timeout: context[:test_timeout] || 30000
+     }}
+  end
+
+  test "validate: accepts no positional arguments" do
+    assert @command.validate([], %{}) == :ok
+  end
+
+  test "validate: any positional arguments fail validation" do
+    assert @command.validate(["quorum-queue-a"], %{}) == {:validation_failure, :too_many_args}
+
+    assert @command.validate(["quorum-queue-a", "two"], %{}) ==
+             {:validation_failure, :too_many_args}
+
+    assert @command.validate(["quorum-queue-a", "two", "three"], %{}) ==
+             {:validation_failure, :too_many_args}
+  end
+
+  @tag test_timeout: 3000
+  test "run: targeting an unreachable node throws a badrpc" do
+    assert match?({:badrpc, _}, @command.run([], %{node: :jake@thedog, vhost: "/", timeout: 200}))
+  end
+end