Expand health checks for readiness to serve clients

This is a squash of the commits from `main` which add new API health
checks in connection to
<https://redirect.github.com/rabbitmq/rabbitmq-server/issues/13782>.

    GET /api/health/checks/is-in-service
    GET /api/health/checks/below-node-connection-limit
    GET /api/health/checks/ready-to-serve-clients

Additionally

    GET /api/health/checks/protocol-listener/{protocol}

Now accepts multiple protocol names, comma-separated. When listeners are
available for all requested protocols then the endpoint returns 200 OK.

Squashed commits:

0d692fa Prefer node-local listeners helper in protocol-listener health check

This is a minor change that avoids a cluster-wide query for active
listeners. The old code called `rabbit_networking:active_listeners/0`
and then filtered the results by ones available on the local node. This
caused an RPC and concatenation of all other cluster members' listeners
and then in the next line filtered down to local nodes. Equivalently we
can use `rabbit_networking:node_listeners(node())` which dumps a local
ETS table.

This is not a very impactful change but it's nice to keep the latency of
the health-check handlers low and reduce some unnecessary cluster noise.

5d319be Accept multiple protocols in protocol listener health check

This updates the health check for protocol listeners to accept a set of
protocols, comma-separated. The check only returns 200 OK when all
requested protocols have active listeners.

07fe630 Add an 'is-in-service' health check wrapping `rabbit:is_serving/0`

This is useful for a load balancer, for example, to be able to avoid
sending new connections to a node which is running and has listeners
bound to TCP ports but is being drained for maintenance.

028b692 Add a health check for testing the node connection limit

67bdc01 Add a health check for testing readiness to serve clients

Author: the-mikedavis

deps/rabbitmq_management/priv/www/api/index.html

@@ -1054,10 +1054,12 @@ <h2>Reference</h2>
         <td></td>
         <td></td>
         <td></td>
-        <td class="path">/api/health/checks/protocol-listener/<i>protocol</i></td>
+        <td class="path">/api/health/checks/protocol-listener/<i>protocols</i></td>
         <td>
-          Responds a 200 OK if there is an active listener for the given protocol,
-          otherwise responds with a 503 Service Unavailable. Valid protocol names are: amqp091, amqp10, mqtt, stomp, web-mqtt, web-stomp.
+          Responds a 200 OK if all given protocols have active listeners,
+          otherwise responds with a 503 Service Unavailable. Multiple protocols
+          may be provided by separating the names with commas. Valid protocol
+          names are: amqp091, amqp10, mqtt, stomp, web-mqtt, web-stomp.
         </td>
       </tr>
       <tr>
@@ -1097,6 +1099,54 @@ <h2>Reference</h2>
           otherwise responds with a 503 Service Unavailable.
         </td>
       </tr>
+      <tr>
+        <td>X</td>
+        <td></td>
+        <td></td>
+        <td></td>
+        <td class="path">/api/health/checks/is-in-service</td>
+        <td>
+          Responds a 200 OK if the target node is booted, running, and ready to
+          serve clients, otherwise responds with a 503 Service Unavailable. If the
+          target node is being drained for maintenance then this check returns 503
+          Service Unavailable.
+        </td>
+      </tr>
+      <tr>
+        <td>X</td>
+        <td></td>
+        <td></td>
+        <td></td>
+        <td class="path">/api/health/checks/below-node-connection-limit</td>
+        <td>
+          Responds a 200 OK if the target node has fewer connections to the AMQP
+          and AMQPS ports than the configured maximum, otherwise responds with a
+          503 Service Unavailable.
+        </td>
+      </tr>
+      <tr>
+        <td>X</td>
+        <td></td>
+        <td></td>
+        <td></td>
+        <td class="path">/api/health/checks/ready-to-serve-clients</td>
+        <td>
+          <p>
+            Responds a 200 OK if the target node is ready to serve clients, otherwise
+            responds with a 503 Service Unavailable. This check combines:
+          </p>
+          <ol>
+            <li>/api/health/checks/is-in-service</li>
+            <li>/api/health/checks/protocol-listener/amqp or /api/health/checks/protocol-listener/amqps</li>
+            <li>/api/health/checks/below-node-connection-limit</li>
+          </ol>
+          <p>
+            So this check will only return 200 OK if the target node is in service,
+            an AMQP or AMQPS listener is available and the target node has fewer active
+            AMQP and AMQPS connections that its configured limit.
+          </p>
+        </td>
+      </tr>
       <tr>
         <td>X</td>
         <td></td>

deps/rabbitmq_management/src/rabbit_mgmt_dispatcher.erl

@@ -187,10 +187,13 @@ dispatcher() ->
      {"/health/checks/local-alarms",                           rabbit_mgmt_wm_health_check_local_alarms, []},
      {"/health/checks/certificate-expiration/:within/:unit",   rabbit_mgmt_wm_health_check_certificate_expiration, []},
      {"/health/checks/port-listener/:port",                    rabbit_mgmt_wm_health_check_port_listener, []},
-     {"/health/checks/protocol-listener/:protocol",            rabbit_mgmt_wm_health_check_protocol_listener, []},
+     {"/health/checks/protocol-listener/:protocols",           rabbit_mgmt_wm_health_check_protocol_listener, []},
      {"/health/checks/virtual-hosts",                          rabbit_mgmt_wm_health_check_virtual_hosts, []},
      {"/health/checks/node-is-mirror-sync-critical",           rabbit_mgmt_wm_health_check_node_is_mirror_sync_critical, []},
      {"/health/checks/node-is-quorum-critical",                rabbit_mgmt_wm_health_check_node_is_quorum_critical, []},
+     {"/health/checks/is-in-service",                          rabbit_mgmt_wm_health_check_is_in_service, []},
+     {"/health/checks/below-node-connection-limit",            rabbit_mgmt_wm_health_check_below_node_connection_limit, []},
+     {"/health/checks/ready-to-serve-clients",                 rabbit_mgmt_wm_health_check_ready_to_serve_clients, []},
      {"/reset",                                                rabbit_mgmt_wm_reset, []},
      {"/reset/:node",                                          rabbit_mgmt_wm_reset, []},
      {"/rebalance/queues",                                     rabbit_mgmt_wm_rebalance_queues, [{queues, all}]},

deps/rabbitmq_management/src/rabbit_mgmt_wm_health_check_below_node_connection_limit.erl

@@ -0,0 +1,63 @@
+%% 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) 2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+%%
+
+-module(rabbit_mgmt_wm_health_check_below_node_connection_limit).
+
+-export([init/2]).
+-export([to_json/2, content_types_provided/2]).
+-export([variances/2]).
+
+-include("rabbit_mgmt.hrl").
+-include_lib("rabbitmq_management_agent/include/rabbit_mgmt_records.hrl").
+
+init(Req, _State) ->
+    Req1 = rabbit_mgmt_headers:set_no_cache_headers(
+             rabbit_mgmt_headers:set_common_permission_headers(
+               Req, ?MODULE), ?MODULE),
+    {cowboy_rest, Req1, #context{}}.
+
+variances(Req, Context) ->
+    {[<<"accept-encoding">>, <<"origin">>], Req, Context}.
+
+content_types_provided(ReqData, Context) ->
+   {rabbit_mgmt_util:responder_map(to_json), ReqData, Context}.
+
+to_json(ReqData, Context) ->
+    ActiveConns = lists:foldl(
+                    fun(Protocol, Acc) ->
+                            Acc + protocol_connection_count(Protocol)
+                    end, 0, [amqp, 'amqp/ssl']),
+    Limit = rabbit_misc:get_env(rabbit, connection_max, infinity),
+    case ActiveConns < Limit of
+        true ->
+            rabbit_mgmt_util:reply(
+              #{status => ok,
+                limit => Limit,
+                connections => ActiveConns}, ReqData, Context);
+        false ->
+            Body = #{
+                status => failed,
+                reason => <<"node connection limit is reached">>,
+                limit => Limit,
+                connections => ActiveConns
+            },
+            {Response, ReqData1, Context1} = rabbit_mgmt_util:reply(
+                                               Body, ReqData, Context),
+            {stop,
+             cowboy_req:reply(
+               ?HEALTH_CHECK_FAILURE_STATUS, #{}, Response, ReqData1),
+             Context1}
+    end.
+
+protocol_connection_count(Protocol) ->
+    case rabbit_networking:ranch_ref_of_protocol(Protocol) of
+        undefined ->
+            0;
+        RanchRef ->
+            #{active_connections := Count} = ranch:info(RanchRef),
+            Count
+    end.

deps/rabbitmq_management/src/rabbit_mgmt_wm_health_check_is_in_service.erl

@@ -0,0 +1,44 @@
+%% 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) 2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+%%
+
+-module(rabbit_mgmt_wm_health_check_is_in_service).
+
+-export([init/2]).
+-export([to_json/2, content_types_provided/2]).
+-export([variances/2]).
+
+-include("rabbit_mgmt.hrl").
+-include_lib("rabbitmq_management_agent/include/rabbit_mgmt_records.hrl").
+
+init(Req, _State) ->
+    Req1 = rabbit_mgmt_headers:set_no_cache_headers(
+             rabbit_mgmt_headers:set_common_permission_headers(
+               Req, ?MODULE), ?MODULE),
+    {cowboy_rest, Req1, #context{}}.
+
+variances(Req, Context) ->
+    {[<<"accept-encoding">>, <<"origin">>], Req, Context}.
+
+content_types_provided(ReqData, Context) ->
+   {rabbit_mgmt_util:responder_map(to_json), ReqData, Context}.
+
+to_json(ReqData, Context) ->
+    case rabbit:is_serving() of
+        true ->
+            rabbit_mgmt_util:reply(#{status => ok}, ReqData, Context);
+        false ->
+            Msg = "this rabbit node is not currently available to serve",
+            failure(Msg, ReqData, Context)
+    end.
+
+failure(Message, ReqData, Context) ->
+    Body = #{
+        status => failed,
+        reason => rabbit_data_coercion:to_binary(Message)
+    },
+    {Response, ReqData1, Context1} = rabbit_mgmt_util:reply(Body, ReqData, Context),
+    {stop, cowboy_req:reply(?HEALTH_CHECK_FAILURE_STATUS, #{}, Response, ReqData1), Context1}.

deps/rabbitmq_management/src/rabbit_mgmt_wm_health_check_protocol_listener.erl

@@ -27,38 +27,46 @@ content_types_provided(ReqData, Context) ->
    {rabbit_mgmt_util:responder_map(to_json), ReqData, Context}.
 
 resource_exists(ReqData, Context) ->
-    {case protocol(ReqData) of
+    {case protocols(ReqData) of
          none -> false;
          _ -> true
      end, ReqData, Context}.
 
 to_json(ReqData, Context) ->
-    Protocol = normalize_protocol(protocol(ReqData)),
-    Listeners = rabbit_networking:active_listeners(),
-    Local = [L || #listener{node = N} = L <- Listeners, N == node()],
-    ProtoListeners = [L || #listener{protocol = P} = L <- Local, atom_to_list(P) == Protocol],
-    case ProtoListeners of
+    Protocols = string:split(protocols(ReqData), ",", all),
+    RequestedProtocols = sets:from_list(
+                           [normalize_protocol(P) || P <- Protocols],
+                           [{version, 2}]),
+    Listeners = rabbit_networking:node_listeners(node()),
+    ActiveProtocols = sets:from_list(
+                        [atom_to_list(P) || #listener{protocol = P} <- Listeners],
+                        [{version, 2}]),
+    MissingProtocols = sets:to_list(sets:subtract(RequestedProtocols, ActiveProtocols)),
+    case MissingProtocols of
         [] ->
-            Msg = <<"No active listener">>,
-            failure(Msg, Protocol, [P || #listener{protocol = P} <- Local], ReqData, Context);
+            Body = #{status    => ok,
+                     protocols => [list_to_binary(P) || P <- sets:to_list(ActiveProtocols)]},
+            rabbit_mgmt_util:reply(Body, ReqData, Context);
         _ ->
-            rabbit_mgmt_util:reply([{status, ok},
-                                    {protocol, list_to_binary(Protocol)}], ReqData, Context)
+            Msg = <<"No active listener">>,
+            failure(Msg, MissingProtocols, sets:to_list(ActiveProtocols), ReqData, Context)
     end.
 
 failure(Message, Missing, Protocols, ReqData, Context) ->
-    {Response, ReqData1, Context1} = rabbit_mgmt_util:reply([{status, failed},
-                                                             {reason, Message},
-                                                             {missing, list_to_binary(Missing)},
-                                                             {protocols, Protocols}],
-                                                            ReqData, Context),
+    Body = #{
+        status    => failed,
+        reason    => Message,
+        missing   => [list_to_binary(P) || P <- Missing],
+        protocols => [list_to_binary(P) || P <- Protocols]
+    },
+    {Response, ReqData1, Context1} = rabbit_mgmt_util:reply(Body, ReqData, Context),
     {stop, cowboy_req:reply(503, #{}, Response, ReqData1), Context1}.
 
 is_authorized(ReqData, Context) ->
     rabbit_mgmt_util:is_authorized(ReqData, Context).
 
-protocol(ReqData) ->
-    rabbit_mgmt_util:id(protocol, ReqData).
+protocols(ReqData) ->
+    rabbit_mgmt_util:id(protocols, ReqData).
 
 normalize_protocol(Protocol) ->
     case string:lowercase(binary_to_list(Protocol)) of

deps/rabbitmq_management/src/rabbit_mgmt_wm_health_check_ready_to_serve_clients.erl

@@ -0,0 +1,81 @@
+%% 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) 2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+%%
+
+%% A composite health check that combines:
+%% * GET /api/health/checks/is-in-service
+%% * GET /api/health/checks/protocol-listener/amqp
+%% * GET /api/health/checks/below-node-connection-limit
+
+-module(rabbit_mgmt_wm_health_check_ready_to_serve_clients).
+
+-export([init/2]).
+-export([to_json/2, content_types_provided/2]).
+-export([variances/2]).
+
+-include("rabbit_mgmt.hrl").
+-include_lib("rabbitmq_management_agent/include/rabbit_mgmt_records.hrl").
+
+init(Req, _State) ->
+    Req1 = rabbit_mgmt_headers:set_no_cache_headers(
+             rabbit_mgmt_headers:set_common_permission_headers(
+               Req, ?MODULE), ?MODULE),
+    {cowboy_rest, Req1, #context{}}.
+
+variances(Req, Context) ->
+    {[<<"accept-encoding">>, <<"origin">>], Req, Context}.
+
+content_types_provided(ReqData, Context) ->
+   {rabbit_mgmt_util:responder_map(to_json), ReqData, Context}.
+
+to_json(ReqData, Context) ->
+    case check() of
+        {ok, Body} ->
+            rabbit_mgmt_util:reply(Body, ReqData, Context);
+        {error, Body} ->
+            {Response, ReqData1, Context1} = rabbit_mgmt_util:reply(
+                                               Body, ReqData, Context),
+            {stop,
+             cowboy_req:reply(
+               ?HEALTH_CHECK_FAILURE_STATUS, #{}, Response, ReqData1),
+             Context1}
+    end.
+
+check() ->
+    case rabbit:is_serving() of
+        true ->
+            RanchRefs0 = [
+                rabbit_networking:ranch_ref_of_protocol(amqp),
+                rabbit_networking:ranch_ref_of_protocol('amqp/ssl')
+            ],
+            RanchRefs = [R || R <- RanchRefs0, R =/= undefined],
+            case RanchRefs of
+                [_ | _] ->
+                    ActiveConns = lists:foldl(
+                      fun(RanchRef, Acc) ->
+                              #{active_connections := Count} = ranch:info(RanchRef),
+                              Acc + Count
+                      end, 0, RanchRefs),
+                    Limit = rabbit_misc:get_env(rabbit, connection_max, infinity),
+                    case ActiveConns < Limit of
+                        true ->
+                            {ok, #{status => ok,
+                                   limit => Limit,
+                                   connections => ActiveConns}};
+                        false ->
+                            {error, #{status => failed,
+                                      reason => <<"node connection limit is reached">>,
+                                      limit => Limit,
+                                      connections => ActiveConns}}
+                    end;
+                [] ->
+                    {error, #{status => failed,
+                              reason => <<"no active listeners for AMQP/AMQPS">>}}
+            end;
+        false ->
+            {error, #{status => failed,
+                      reason => <<"the rabbit node is not currently available to serve">>}}
+    end.