Support global flag to run leader health check for

all queues in all vhosts on local node

Author: Ayanda-D

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -2154,12 +2154,22 @@ leader_health_check(QueueNameOrRegEx, VHost) ->
     %% we cannot spawn any new processes for executing QQ leader health checks.
     ProcessLimitThreshold = round(0.4 * erlang:system_info(process_limit)),
 
+    leader_health_check(QueueNameOrRegEx, VHost, ProcessLimitThreshold).
+
+leader_health_check(QueueNameOrRegEx, VHost, ProcessLimitThreshold) ->
+    Qs =
+        case VHost of
+            global ->
+                rabbit_amqqueue:list();
+            VHost when is_binary(VHost) ->
+                rabbit_amqqueue:list(VHost)
+        end,
     ParentPID = self(),
     HealthCheckRef = make_ref(),
     HealthCheckPids =
         lists:flatten(
             [begin
-                {resource, VHost, queue, QueueName} = QResource = amqqueue:get_name(Q),
+                {resource, _VHostN, queue, QueueName} = QResource = amqqueue:get_name(Q),
                 case check_process_limit_safety(ProcessLimitThreshold) of
                     true ->
                         case re:run(QueueName, QueueNameOrRegEx, [{capture, none}]) of
@@ -2173,7 +2183,7 @@ leader_health_check(QueueNameOrRegEx, VHost) ->
                         rabbit_log:warning("Leader health check failed from exceeded process limit threshold"),
                         throw({error, leader_health_check_process_limit_exceeded})
                 end
-            end || Q <- rabbit_amqqueue:list(VHost), amqqueue:get_type(Q) == ?MODULE]),
+            end || Q <- Qs, amqqueue:get_type(Q) == ?MODULE]),
     wait_for_leader_health_checks(HealthCheckRef, length(HealthCheckPids), []).
 
 run_leader_health_check(ClusterName, QResource, HealthCheckRef, From) ->

deps/rabbit/test/quorum_queue_SUITE.erl

@@ -4147,52 +4147,119 @@ amqpl_headers(Config) ->
                                             multiple = true}).
 
 leader_health_check(Config) ->
-    [Server | _] = _Servers = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
-    Ch = rabbit_ct_client_helpers:open_channel(Config, Server),
+    VHost1 = <<"vhost1">>,
+    VHost2 = <<"vhost2">>,
+
+    set_up_vhost(Config, VHost1),
+    set_up_vhost(Config, VHost2),
+
+    Conn1 = rabbit_ct_client_helpers:open_unmanaged_connection(Config, 0, VHost1),
+    {ok, Ch1} = amqp_connection:open_channel(Conn1),
+
+    Conn2 = rabbit_ct_client_helpers:open_unmanaged_connection(Config, 0, VHost2),
+    {ok, Ch2} = amqp_connection:open_channel(Conn2),
+
+    Qs1 = [<<"Q.1">>, <<"Q.2">>, <<"Q.3">>],
+    Qs2 = [<<"Q.4">>, <<"Q.5">>, <<"Q.6">>],
+
+    %% in vhost1
     [?assertEqual({'queue.declare_ok', Q, 0, 0},
-                 declare(Ch, Q, [{<<"x-queue-type">>, longstr, <<"quorum">>}]))
-        || Q <- [<<"Q.1">>, <<"Q.2">>, <<"Q.3">>]],
+                 declare(Ch1, Q, [{<<"x-queue-type">>, longstr, <<"quorum">>}]))
+        || Q <- Qs1],
+
+    %% in vhost2
+    [?assertEqual({'queue.declare_ok', Q, 0, 0},
+                 declare(Ch2, Q, [{<<"x-queue-type">>, longstr, <<"quorum">>}]))
+        || Q <- Qs2],
+
+    %% test sucessful health checks in vhost1, vhost2, global
     ?assertEqual([], rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue, leader_health_check,
-                                      [<<".*">>, <<"/">>])),
+                                      [<<".*">>, VHost1])),
     ?assertEqual([], rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue, leader_health_check,
-                                      [<<"Q.*">>, <<"/">>])),
+                                      [<<"Q.*">>, VHost1])),
+    [?assertEqual([], rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue, leader_health_check,
+                                      [Q, VHost1])) || Q <- Qs1],
+
     ?assertEqual([], rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue, leader_health_check,
-                                      [<<"Q.1">>, <<"/">>])),
+                                      [<<".*">>, VHost2])),
     ?assertEqual([], rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue, leader_health_check,
-                                      [<<"Q.2">>, <<"/">>])),
+                                      [<<"Q.*">>, VHost2])),
+    [?assertEqual([], rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue, leader_health_check,
+                                      [Q, VHost2])) || Q <- Qs2],
+
+    ?assertEqual([], rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue, leader_health_check,
+                                      [<<".*">>, global])),
     ?assertEqual([], rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue, leader_health_check,
-                                      [<<"Q.3">>, <<"/">>])),
+                                      [<<"Q.*">>, global])),
 
+    %% clear leaderboard
     Qs = rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_amqqueue, list, []),
 
-    [{Q1_ClusterName, Q1Res}, {Q2_ClusterName, Q2Res}, {Q3_ClusterName, Q3Res}] =
+    [{_Q1_ClusterName, _Q1Res},
+     {_Q2_ClusterName, _Q2Res},
+     {_Q3_ClusterName, _Q3Res},
+     {_Q4_ClusterName, _Q4Res},
+     {_Q5_ClusterName, _Q5Res},
+     {_Q6_ClusterName, _Q6Res}] = QQ_Clusters =
         lists:usort(
             [begin
                 {ClusterName, _} = amqqueue:get_pid(Q),
                 {ClusterName, amqqueue:get_name(Q)}
             end
                 || Q <- Qs, amqqueue:get_type(Q) == rabbit_quorum_queue]),
 
-    rabbit_ct_broker_helpers:rpc(Config, 0, ra_leaderboard, clear, [Q1_ClusterName]),
-    rabbit_ct_broker_helpers:rpc(Config, 0, ra_leaderboard, clear, [Q2_ClusterName]),
-    rabbit_ct_broker_helpers:rpc(Config, 0, ra_leaderboard, clear, [Q3_ClusterName]),
-
-    Q1Data = amqqueue:to_printable(Q1Res, rabbit_quorum_queue),
-    Q2Data = amqqueue:to_printable(Q2Res, rabbit_quorum_queue),
-    Q3Data = amqqueue:to_printable(Q3Res, rabbit_quorum_queue),
+    [Q1Data, Q2Data, Q3Data, Q4Data, Q5Data, Q6Data] = QQ_Data =
+        [begin
+            rabbit_ct_broker_helpers:rpc(Config, 0, ra_leaderboard, clear, [Q_ClusterName]),
+            _QData = amqqueue:to_printable(Q_Res, rabbit_quorum_queue)
+         end
+            || {Q_ClusterName, Q_Res} <- QQ_Clusters],
 
+    %% test failed health checks in vhost1, vhost2, global
     ?assertEqual([Q1Data], rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue, leader_health_check,
-                                      [<<"Q.1">>, <<"/">>])),
+                                      [<<"Q.1">>, VHost1])),
     ?assertEqual([Q2Data], rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue, leader_health_check,
-                                      [<<"Q.2">>, <<"/">>])),
+                                      [<<"Q.2">>, VHost1])),
     ?assertEqual([Q3Data], rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue, leader_health_check,
-                                      [<<"Q.3">>, <<"/">>])),
+                                      [<<"Q.3">>, VHost1])),
     ?assertEqual([Q1Data, Q2Data, Q3Data],
         lists:usort(rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue, leader_health_check,
-                        [<<".*">>, <<"/">>]))),
+                        [<<".*">>, VHost1]))),
     ?assertEqual([Q1Data, Q2Data, Q3Data],
         lists:usort(rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue, leader_health_check,
-                        [<<"Q.*">>, <<"/">>]))).
+                        [<<"Q.*">>, VHost1]))),
+
+    ?assertEqual([Q4Data], rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue, leader_health_check,
+                                      [<<"Q.4">>, VHost2])),
+    ?assertEqual([Q5Data], rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue, leader_health_check,
+                                      [<<"Q.5">>, VHost2])),
+    ?assertEqual([Q6Data], rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue, leader_health_check,
+                                      [<<"Q.6">>, VHost2])),
+    ?assertEqual([Q4Data, Q5Data, Q6Data],
+        lists:usort(rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue, leader_health_check,
+                        [<<".*">>, VHost2]))),
+    ?assertEqual([Q4Data, Q5Data, Q6Data],
+        lists:usort(rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue, leader_health_check,
+                        [<<"Q.*">>, VHost2]))),
+
+    ?assertEqual(QQ_Data,
+        lists:usort(rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue, leader_health_check,
+                        [<<"Q.*">>, global]))),
+    ?assertEqual(QQ_Data,
+        lists:usort(rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue, leader_health_check,
+                        [<<"Q.*">>, global]))),
+
+    %% cleanup
+    [?assertMatch(#'queue.delete_ok'{},
+                 amqp_channel:call(Ch1, #'queue.delete'{queue = Q}))
+        || Q <- Qs1],
+    [?assertMatch(#'queue.delete_ok'{},
+                 amqp_channel:call(Ch1, #'queue.delete'{queue = Q}))
+        || Q <- Qs2],
+
+    amqp_connection:close(Conn1),
+    amqp_connection:close(Conn2).
+
 
 leader_locator_client_local(Config) ->
     [Server1 | _] = Servers = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
@@ -4514,6 +4581,11 @@ declare_passive(Ch, Q, Args) ->
                                            auto_delete = false,
                                            passive = true,
                                            arguments = Args}).
+
+set_up_vhost(Config, VHost) ->
+    rabbit_ct_broker_helpers:add_vhost(Config, VHost),
+    rabbit_ct_broker_helpers:set_full_permissions(Config, <<"guest">>, VHost).
+
 assert_queue_type(Server, Q, Expected) ->
     assert_queue_type(Server, <<"/">>, Q, Expected).
 

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

@@ -11,20 +11,27 @@ defmodule RabbitMQ.CLI.Queues.Commands.LeaderHealthCheckCommand do
 
   import RabbitMQ.CLI.Core.Platform, only: [line_separator: 0]
 
+  def switches(), do: [global: :boolean]
+
   def scopes(), do: [:queues]
 
+  def merge_defaults(args, opts) do
+    {args, Map.merge(%{global: false, vhost: "/"}, opts)}
+  end
+
   use RabbitMQ.CLI.Core.AcceptsDefaultSwitchesAndTimeout
   use RabbitMQ.CLI.Core.AcceptsOnePositionalArgument
-  use RabbitMQ.CLI.Core.MergesDefaultVirtualHost
   use RabbitMQ.CLI.Core.RequiresRabbitAppRunning
 
-  def run([pattern] = _args, %{node: node_name, vhost: vhost}) do
+  def run([pattern] = _args, %{node: node_name, vhost: vhost, global: global_opt}) do
+    vhost = if global_opt, do: :global, else: vhost
+
     case :rabbit_misc.rpc_call(node_name, :rabbit_quorum_queue, :leader_health_check, [pattern, vhost]) do
       [] ->
         :ok
 
-      unhealthy_queues ->
-        {:error, unhealthy_queues}
+      unhealthy_queues_or_error ->
+        {:error, unhealthy_queues_or_error}
     end
   end
 
@@ -67,12 +74,13 @@ defmodule RabbitMQ.CLI.Queues.Commands.LeaderHealthCheckCommand do
   def formatter(), do: RabbitMQ.CLI.Formatters.PrettyTable
 
   def usage() do
-    "leader_health_check [--vhost <vhost>] <pattern>"
+    "leader_health_check [--vhost <vhost>] [--global] <pattern>"
   end
 
   def usage_additional do
     [
-      ["<pattern>", "regular expression pattern used to match quorum queues"]
+      ["<pattern>", "regular expression pattern used to match quorum queues"],
+      ["--global", "run leader health check for all queues in all virtual hosts on the node"]
     ]
   end
 

deps/rabbitmq_cli/test/queues/leader_health_check_test.exs

@@ -46,7 +46,7 @@ defmodule RabbitMQ.CLI.Queues.Commands.LeaderHealthCheckCommandTest do
             {:error, {:badrpc, :nodedown}},
              @command.run(
                ["quorum.queue.*"],
-               %{node: :jake@thedog, vhost: "/", timeout: 200}
+               %{node: :jake@thedog, vhost: "/", global: false, timeout: 200}
              )
            )
   end