Implement rabbitmq-queues leader_health_check command for quorum queues

Author: Ayanda-D

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -82,6 +82,8 @@
          file_handle_other_reservation/0,
          file_handle_release_reservation/0]).
 
+-export([leader_health_check/2, run_leader_health_check/4]).
+
 -ifdef(TEST).
 -export([filter_promotable/2,
          ra_machine_config/1]).
@@ -144,6 +146,8 @@
 -define(SNAPSHOT_INTERVAL, 8192). %% the ra default is 4096
 % -define(UNLIMITED_PREFETCH_COUNT, 2000). %% something large for ra
 -define(MIN_CHECKPOINT_INTERVAL, 8192). %% the ra default is 16384
+-define(QQ_LEADER_HEALTH_CHECK_TIMEOUT, 1_000).
+-define(QQ_GLOBAL_LEADER_HEALTH_CHECK_TIMEOUT, 60_000).
 
 %%----------- QQ policies ---------------------------------------------------
 
@@ -2145,3 +2149,63 @@ file_handle_other_reservation() ->
 file_handle_release_reservation() ->
     ok.
 
+leader_health_check(QueueNameOrRegEx, VHost) ->
+    %% Set a process limit threshold to 40% of ErlangVM process limit, beyond which
+    %% we cannot spawn any new processes for executing QQ leader health checks.
+    ProcessLimitThreshold = round(0.4 * erlang:system_info(process_limit)),
+
+    HealthCheckRef = make_ref(),
+    HealthCheckPids =
+        lists:flatten(
+            [begin
+                {resource, VHost, queue, QueueName} = QResource = amqqueue:get_name(Q),
+                case check_process_limit_safety(ProcessLimitThreshold) of
+                    true ->
+                        case re:run(QueueName, QueueNameOrRegEx, [{capture, none}]) of
+                            match ->
+                                {ClusterName, _} = rabbit_amqqueue:pid_of(Q),
+                                _Pid = spawn(fun() -> run_leader_health_check(ClusterName, QResource, HealthCheckRef, self()) end);
+                            _ ->
+                                []
+                        end;
+                    false ->
+                        []
+                end
+            end || Q <- rabbit_amqqueue:list(VHost), amqqueue:get_type(Q) == ?MODULE]),
+    wait_for_leader_health_checks(HealthCheckRef, length(HealthCheckPids), []).
+
+run_leader_health_check(ClusterName, QResource, HealthCheckRef, From) ->
+    Leader = ra_leaderboard:lookup_leader(ClusterName),
+    case ra_server_proc:ping(Leader, ?QQ_LEADER_HEALTH_CHECK_TIMEOUT) of
+        {pong,leader} ->
+            From ! {ok, HealthCheckRef, QResource};
+        _ ->
+            From ! {error, HealthCheckRef, QResource}
+    end,
+    ok.
+
+wait_for_leader_health_checks(Ref, N, UnhealthyAcc) ->
+    receive
+        {ok, Ref, _QResource} when N == 1 ->
+            UnhealthyAcc;
+        {error, Ref, QResource} when N == 1 ->
+            [cli_format(QResource) | UnhealthyAcc];
+        {ok, Ref, _QResource} ->
+            wait_for_leader_health_checks(Ref, N - 1, UnhealthyAcc);
+        {error, Ref, QResource} ->
+            wait_for_leader_health_checks(Ref, N - 1, [cli_format(QResource) | UnhealthyAcc])
+    after
+        ?QQ_GLOBAL_LEADER_HEALTH_CHECK_TIMEOUT ->
+            UnhealthyAcc
+    end.
+
+check_process_limit_safety(ProcessLimitThreshold) ->
+    erlang:system_info(process_count) < ProcessLimitThreshold.
+
+cli_format(QResource = {resource, VHost, queue, QName}) ->
+    #{
+        <<"readable_name">> => rabbit_data_coercion:to_binary(rabbit_misc:rs(QResource)),
+        <<"name">> => QName,
+        <<"virtual_host">> => VHost,
+        <<"type">> => <<"quorum">>
+    }.

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

@@ -0,0 +1,78 @@
+## 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-2024 VMware, Inc. or its affiliates.  All rights reserved.
+
+defmodule RabbitMQ.CLI.Queues.Commands.LeaderHealthCheckCommand do
+  alias RabbitMQ.CLI.Core.DocGuide
+
+  @behaviour RabbitMQ.CLI.CommandBehaviour
+
+  import RabbitMQ.CLI.Core.Platform, only: [line_separator: 0]
+
+  def scopes(), do: [:queues]
+
+  use RabbitMQ.CLI.Core.AcceptsDefaultSwitchesAndTimeout
+
+  use RabbitMQ.CLI.Core.MergesDefaultVirtualHost
+
+  def validate(args, _) when length(args) < 1 do
+    {:validation_failure, :not_enough_args}
+  end
+
+  def validate(args, _) when length(args) > 1 do
+    {:validation_failure, :too_many_args}
+  end
+
+  use RabbitMQ.CLI.Core.RequiresRabbitAppRunning
+
+  def run([pattern] = _args, %{node: node_name, vhost: vhost}) do
+    case :rabbit_misc.rpc_call(node_name, :rabbit_quorum_queue, :leader_health_check, [pattern, vhost]) do
+      [] ->
+        :ok
+
+      unhealthy_queues ->
+        {:error, unhealthy_queues}
+    end
+  end
+
+  def output(:ok, %{formatter: "json"}) do
+    {:error, :check_passed}
+  end
+
+  def output({:error, unhealthy_queues}, %{vhost: vhost, formatter: "json"}) when is_list(unhealthy_queues) do
+    lines = queue_lines(unhealthy_queues)
+
+    {:error, :check_failed, Enum.join(lines, line_separator())}
+  end
+
+  def formatter(), do: RabbitMQ.CLI.Formatters.PrettyTable
+
+  def usage() do
+    "leader_health_check [--vhost <vhost>] <pattern>"
+  end
+
+  def usage_additional do
+    [
+      ["<pattern>", "regular expression pattern used to match quorum queues"]
+    ]
+  end
+
+  def help_section(), do: :observability_and_health_checks
+
+  def usage_doc_guides() do
+    [
+      DocGuide.quorum_queues()
+    ]
+  end
+
+  def description(), do: "Checks availability and health of quorum queue leader"
+
+  def banner([name], %{vhost: vhost}),
+    do: "Checking availability and health status of queue(s) matching #{name} in vhost #{vhost} ..."
+
+  def queue_lines(qs) do
+    for q <- qs, do: "Leader for #{q["readable_name"]} is unhealthy and unavailable"
+  end
+end