rabbit_nodes: Add list functions to clarify which nodes we are interested in

So far, we had the following functions to list nodes in a RabbitMQ
cluster:
* `rabbit_mnesia:cluster_nodes/1` to get members of the Mnesia cluster;
  the argument was used to select members (all members or only those
  running Mnesia and participating in the cluster)
* `rabbit_nodes:all/0` to get all members of the Mnesia cluster
* `rabbit_nodes:all_running/0` to get all members who currently run
  Mnesia

Basically:
* `rabbit_nodes:all/0` calls `rabbit_mnesia:cluster_nodes(all)
* `rabbit_nodes:all_running/0` calls `rabbit_mnesia:cluster_nodes(running)`

We also have:
* `rabbit_node_monitor:alive_nodes/1` which filters the given list of
  nodes to only select those currently running Mnesia
* `rabbit_node_monitor:alive_rabbit_nodes/1` which filters the given
  list of nodes to only select those currently running RabbitMQ

Most of the code uses `rabbit_mnesia:cluster_nodes/1` or the
`rabbit_nodes:all*/0` functions. `rabbit_mnesia:cluster_nodes(running)`
or `rabbit_nodes:all_running/0` is often used as a close approximation
of "all cluster members running RabbitMQ". This list might be incorrect
in times where a node is joining the clustered or is being worked on
(i.e. Mnesia is running but not RabbitMQ).

With Khepri, there won't be the same possible approximation because we
will try to keep Khepri/Ra running even if RabbitMQ is stopped to
expand/shrink the cluster.

So in order to clarify what we want when we query a list of nodes, this
patch introduces the following functions:
* `rabbit_nodes:list_members/0` to get all cluster members, regardless
  of their state
* `rabbit_nodes:list_reachable/0` to get all cluster members we can
  reach using Erlang distribution, regardless of the state of RabbitMQ
* `rabbit_nodes:list_running/0` to get all cluster members who run
  RabbitMQ, regardless of the maintenance state
* `rabbit_nodes:list_serving/0` to get all cluster members who run
  RabbitMQ and are accepting clients

In addition to the list functions, there are the corresponding
`rabbit_nodes:filter_*(Nodes)` filtering functions.

The code is modified to use these new functions. One possible
significant change is that the new list functions will perform RPC calls
to query the nodes' state, unlike `rabbit_mnesia:cluster_nodes(running)`.

Author: dumbbell

deps/rabbit/src/rabbit_amqqueue.erl

@@ -441,7 +441,7 @@ rebalance(Type, VhostSpec, QueueSpec) ->
 maybe_rebalance({true, Id}, Type, VhostSpec, QueueSpec) ->
     rabbit_log:info("Starting queue rebalance operation: '~ts' for vhosts matching '~ts' and queues matching '~ts'",
                     [Type, VhostSpec, QueueSpec]),
-    Running = rabbit_maintenance:filter_out_drained_nodes_consistent_read(rabbit_nodes:all_running()),
+    Running = rabbit_maintenance:filter_out_drained_nodes_consistent_read(rabbit_nodes:list_running()),
     NumRunning = length(Running),
     ToRebalance = [Q || Q <- rabbit_amqqueue:list(),
                         filter_per_type(Type, Q),
@@ -1121,7 +1121,7 @@ list() ->
 
 do_list() ->
     All = mnesia:dirty_match_object(rabbit_queue, amqqueue:pattern_match_all()),
-    NodesRunning = rabbit_nodes:all_running(),
+    NodesRunning = rabbit_nodes:list_running(),
     lists:filter(fun (Q) ->
                          Pid = amqqueue:get_pid(Q),
                          St = amqqueue:get_state(Q),
@@ -1307,7 +1307,7 @@ is_in_virtual_host(Q, VHostName) ->
 -spec list(vhost:name()) -> [amqqueue:amqqueue()].
 list(VHostPath) ->
     All = list(VHostPath, rabbit_queue),
-    NodesRunning = rabbit_nodes:all_running(),
+    NodesRunning = rabbit_nodes:list_running(),
     lists:filter(fun (Q) ->
                          Pid = amqqueue:get_pid(Q),
                          St = amqqueue:get_state(Q),
@@ -1368,7 +1368,7 @@ list_down(VHostPath) ->
         true  ->
             Alive = sets:from_list([amqqueue:get_name(Q) || Q <- list(VHostPath)]),
             Durable = list(VHostPath, rabbit_durable_queue),
-            NodesRunning = rabbit_nodes:all_running(),
+            NodesRunning = rabbit_nodes:list_running(),
             lists:filter(fun (Q) ->
                                  N = amqqueue:get_name(Q),
                                  Pid = amqqueue:get_pid(Q),
@@ -1488,7 +1488,7 @@ emit_info_all(Nodes, VHostPath, Items, Ref, AggregatorPid) ->
     rabbit_control_misc:await_emitters_termination(Pids).
 
 collect_info_all(VHostPath, Items) ->
-    Nodes = rabbit_nodes:all_running(),
+    Nodes = rabbit_nodes:list_running(),
     Ref = make_ref(),
     Pids = [ spawn_link(Node, rabbit_amqqueue, emit_info_local, [VHostPath, Items, Ref, self()]) || Node <- Nodes ],
     rabbit_control_misc:await_emitters_termination(Pids),
@@ -1904,10 +1904,8 @@ forget_node_for_queue(DeadNode, [H|T], Q) when ?is_amqqueue(Q) ->
 node_permits_offline_promotion(Node) ->
     case node() of
         Node -> not rabbit:is_running(); %% [1]
-        _    -> All = rabbit_nodes:all(),
-                Running = rabbit_nodes:all_running(),
-                lists:member(Node, All) andalso
-                    not lists:member(Node, Running) %% [2]
+        _    -> NotRunning = rabbit_nodes:list_not_running(),
+                lists:member(Node, NotRunning) %% [2]
     end.
 %% [1] In this case if we are a real running node (i.e. rabbitmqctl
 %% has RPCed into us) then we cannot allow promotion. If on the other

deps/rabbit/src/rabbit_channel.erl

@@ -371,7 +371,7 @@ send_drained(Pid, CTagCredit) ->
 -spec list() -> [pid()].
 
 list() ->
-    Nodes = rabbit_nodes:all_running(),
+    Nodes = rabbit_nodes:list_running(),
     rabbit_misc:append_rpc_all_nodes(Nodes, rabbit_channel, list_local, [], ?RPC_TIMEOUT).
 
 -spec list_local() -> [pid()].

deps/rabbit/src/rabbit_channel_tracking.erl

@@ -262,7 +262,7 @@ list() ->
     lists:foldl(
       fun (Node, Acc) ->
               Acc ++ list_on_node(Node)
-      end, [], rabbit_nodes:all_running()).
+      end, [], rabbit_nodes:list_running()).
 
 -spec list_of_user(rabbit_types:username()) -> [rabbit_types:tracked_channel()].
 
@@ -307,7 +307,7 @@ list_on_node_mnesia(Node) ->
           #tracked_channel{_ = '_'})
     catch exit:{aborted, {no_exists, _}} ->
             %% The table might not exist yet (or is already gone)
-            %% between the time rabbit_nodes:all_running() runs and
+            %% between the time rabbit_nodes:list_running() runs and
             %% returns a specific node, and
             %% mnesia:dirty_match_object() is called for that node's
             %% table.

deps/rabbit/src/rabbit_connection_tracking.erl

@@ -434,7 +434,7 @@ get_all_tracked_connection_table_names_for_node(Node) ->
 -spec lookup(rabbit_types:connection_name()) -> rabbit_types:tracked_connection() | 'not_found'.
 
 lookup(Name) ->
-    Nodes = rabbit_nodes:all_running(),
+    Nodes = rabbit_nodes:list_running(),
     lookup(Name, Nodes).
 
 lookup(_, []) ->
@@ -469,15 +469,15 @@ list() ->
     lists:foldl(
       fun (Node, Acc) ->
               Acc ++ list_on_node(Node)
-      end, [], rabbit_nodes:all_running()).
+      end, [], rabbit_nodes:list_running()).
 
 -spec count() -> non_neg_integer().
 
 count() ->
     lists:foldl(
       fun (Node, Acc) ->
               count_on_node(Node) + Acc
-      end, 0, rabbit_nodes:all_running()).
+      end, 0, rabbit_nodes:list_running()).
 
 count_on_node(Node) ->
     case rabbit_feature_flags:is_enabled(tracking_records_in_ets) of
@@ -550,7 +550,7 @@ list_on_node_mnesia(Node) ->
           #tracked_connection{_ = '_'})
     catch exit:{aborted, {no_exists, _}} ->
             %% The table might not exist yet (or is already gone)
-            %% between the time rabbit_nodes:all_running() runs and
+            %% between the time rabbit_nodes:list_running() runs and
             %% returns a specific node, and
             %% mnesia:dirty_match_object() is called for that node's
             %% table.

deps/rabbit/src/rabbit_direct.erl

@@ -44,7 +44,7 @@ list_local() ->
 -spec list() -> [pid()].
 
 list() ->
-    Nodes = rabbit_nodes:all_running(),
+    Nodes = rabbit_nodes:list_running(),
     rabbit_misc:append_rpc_all_nodes(Nodes, rabbit_direct, list_local, [], ?RPC_TIMEOUT).
 
 %%----------------------------------------------------------------------------

deps/rabbit/src/rabbit_maintenance.erl

@@ -341,7 +341,7 @@ stop_local_quorum_queue_followers() ->
 
 -spec primary_replica_transfer_candidate_nodes() -> [node()].
 primary_replica_transfer_candidate_nodes() ->
-    filter_out_drained_nodes_consistent_read(rabbit_nodes:all_running() -- [node()]).
+    filter_out_drained_nodes_consistent_read(rabbit_nodes:list_running() -- [node()]).
 
 -spec random_primary_replica_transfer_candidate_node([node()], [node()]) -> {ok, node()} | undefined.
 random_primary_replica_transfer_candidate_node([], _Preferred) ->

deps/rabbit/src/rabbit_mirror_queue_misc.erl

@@ -154,7 +154,7 @@ remove_from_queue(QueueName, Self, DeadGMPids) ->
 slaves_to_start_on_failure(Q, DeadGMPids) ->
     %% In case Mnesia has not caught up yet, filter out nodes we know
     %% to be dead..
-    ClusterNodes = rabbit_nodes:all_running() --
+    ClusterNodes = rabbit_nodes:list_running() --
         [node(P) || P <- DeadGMPids],
     {_, OldNodes, _} = actual_queue_nodes(Q),
     {_, NewNodes} = suggested_queue_nodes(Q, ClusterNodes),
@@ -321,7 +321,7 @@ store_updated_slaves(Q0) when ?is_amqqueue(Q0) ->
 %% a long time without being removed.
 update_recoverable(SPids, RS) ->
     SNodes = [node(SPid) || SPid <- SPids],
-    RunningNodes = rabbit_nodes:all_running(),
+    RunningNodes = rabbit_nodes:list_running(),
     AddNodes = SNodes -- RS,
     DelNodes = RunningNodes -- SNodes, %% i.e. running with no slave
     (RS -- DelNodes) ++ AddNodes.
@@ -375,17 +375,17 @@ promote_slave([SPid | SPids]) ->
 -spec initial_queue_node(amqqueue:amqqueue(), node()) -> node().
 
 initial_queue_node(Q, DefNode) ->
-    {MNode, _SNodes} = suggested_queue_nodes(Q, DefNode, rabbit_nodes:all_running()),
+    {MNode, _SNodes} = suggested_queue_nodes(Q, DefNode, rabbit_nodes:list_running()),
     MNode.
 
 -spec suggested_queue_nodes(amqqueue:amqqueue()) ->
           {node(), [node()]}.
 
-suggested_queue_nodes(Q)      -> suggested_queue_nodes(Q, rabbit_nodes:all_running()).
+suggested_queue_nodes(Q)      -> suggested_queue_nodes(Q, rabbit_nodes:list_running()).
 suggested_queue_nodes(Q, All) -> suggested_queue_nodes(Q, node(), All).
 
 %% The third argument exists so we can pull a call to
-%% rabbit_nodes:all_running() out of a loop or transaction
+%% rabbit_nodes:list_running() out of a loop or transaction
 %% or both.
 suggested_queue_nodes(Q, DefNode, All) when ?is_amqqueue(Q) ->
     Owner = amqqueue:get_exclusive_owner(Q),

deps/rabbit/src/rabbit_mnesia_rename.erl

@@ -100,7 +100,7 @@ prepare(Node, NodeMapList) ->
 
     %% Check that we are in the cluster, all old nodes are in the
     %% cluster, and no new nodes are.
-    Nodes = rabbit_nodes:all(),
+    Nodes = rabbit_nodes:list_members(),
     case {FromNodes -- Nodes, ToNodes -- (ToNodes -- Nodes),
           lists:member(Node, Nodes ++ ToNodes)} of
         {[], [], true}  -> ok;
@@ -130,7 +130,7 @@ restore_backup(Backup) ->
 -spec maybe_finish() -> ok.
 
 maybe_finish() ->
-    AllNodes = rabbit_nodes:all(),
+    AllNodes = rabbit_nodes:list_members(),
     maybe_finish(AllNodes).
 
 -spec maybe_finish([node()]) -> 'ok'.
@@ -144,7 +144,7 @@ maybe_finish(AllNodes) ->
 finish(FromNode, ToNode, AllNodes) ->
     case node() of
         ToNode ->
-            case rabbit_nodes:filter_nodes_running_rabbitmq(AllNodes) of
+            case rabbit_nodes:filter_running(AllNodes) of
                 [] -> finish_primary(FromNode, ToNode);
                 _  -> finish_secondary(FromNode, ToNode, AllNodes)
             end;
@@ -257,8 +257,8 @@ update_term(_NodeMap, Term) ->
     Term.
 
 rename_in_running_mnesia(FromNode, ToNode) ->
-    All = rabbit_nodes:all(),
-    Running = rabbit_nodes:all_running(),
+    All = rabbit_nodes:list_members(),
+    Running = rabbit_nodes:list_running(),
     case {lists:member(FromNode, Running), lists:member(ToNode, All)} of
         {false, true}  -> ok;
         {true,  _}     -> exit({old_node_running,        FromNode});

deps/rabbit/src/rabbit_networking.erl

@@ -517,7 +517,7 @@ unregister_connection(Pid) -> pg_local:leave(rabbit_connections, Pid).
 -spec connections() -> [rabbit_types:connection()].
 
 connections() ->
-    Nodes = rabbit_nodes:all_running(),
+    Nodes = rabbit_nodes:list_running(),
     rabbit_misc:append_rpc_all_nodes(Nodes, rabbit_networking, connections_local, [], ?RPC_TIMEOUT).
 
 -spec local_connections() -> [rabbit_types:connection()].
@@ -540,7 +540,7 @@ unregister_non_amqp_connection(Pid) -> pg_local:leave(rabbit_non_amqp_connection
 -spec non_amqp_connections() -> [rabbit_types:connection()].
 
 non_amqp_connections() ->
-  Nodes = rabbit_nodes:all_running(),
+  Nodes = rabbit_nodes:list_running(),
   rabbit_misc:append_rpc_all_nodes(Nodes, rabbit_networking, local_non_amqp_connections, [], ?RPC_TIMEOUT).
 
 -spec local_non_amqp_connections() -> [rabbit_types:connection()].

deps/rabbit/src/rabbit_node_monitor.erl

@@ -163,7 +163,7 @@ notify_node_up() ->
 
 notify_joined_cluster() ->
     NewMember = node(),
-    Nodes = rabbit_nodes:all_running() -- [NewMember],
+    Nodes = rabbit_nodes:list_running() -- [NewMember],
     gen_server:abcast(Nodes, ?SERVER,
                       {joined_cluster, node(), rabbit_db_cluster:node_type()}),
 
@@ -172,7 +172,7 @@ notify_joined_cluster() ->
 -spec notify_left_cluster(node()) -> 'ok'.
 
 notify_left_cluster(Node) ->
-    Nodes = rabbit_nodes:all_running(),
+    Nodes = rabbit_nodes:list_running(),
     gen_server:abcast(Nodes, ?SERVER, {left_cluster, Node}),
     ok.
 
@@ -409,7 +409,7 @@ handle_call(_Request, _From, State) ->
     {noreply, State}.
 
 handle_cast(notify_node_up, State = #state{guid = GUID}) ->
-    Nodes = rabbit_nodes:all_running() -- [node()],
+    Nodes = rabbit_nodes:list_running() -- [node()],
     gen_server:abcast(Nodes, ?SERVER,
                       {node_up, node(), rabbit_db_cluster:node_type(), GUID}),
     %% register other active rabbits with this rabbit
@@ -466,7 +466,7 @@ handle_cast({announce_guid, Node, GUID}, State = #state{node_guids = GUIDs}) ->
 handle_cast({check_partial_partition, Node, Rep, NodeGUID, MyGUID, RepGUID},
             State = #state{guid       = MyGUID,
                            node_guids = GUIDs}) ->
-    case lists:member(Node, rabbit_nodes:all_running()) andalso
+    case lists:member(Node, rabbit_nodes:list_members()) andalso
         maps:find(Node, GUIDs) =:= {ok, NodeGUID} of
         true  -> spawn_link( %%[1]
                    fun () ->
@@ -618,7 +618,7 @@ handle_info({nodedown, Node, Info}, State = #state{guid       = MyGUID,
                              Node, node(), DownGUID, CheckGUID, MyGUID})
             end,
     case maps:find(Node, GUIDs) of
-        {ok, DownGUID} -> Alive = rabbit_nodes:all_running()
+        {ok, DownGUID} -> Alive = rabbit_nodes:list_members()
                               -- [node(), Node],
                           [case maps:find(N, GUIDs) of
                                {ok, CheckGUID} -> Check(N, CheckGUID, DownGUID);
@@ -817,7 +817,7 @@ handle_dead_rabbit(Node, State = #state{partitions = Partitions,
     %% going away. It's only safe to forget anything about partitions when
     %% there are no partitions.
     Down = Partitions -- alive_rabbit_nodes(),
-    NoLongerPartitioned = rabbit_nodes:all_running(),
+    NoLongerPartitioned = rabbit_nodes:list_running(),
     Partitions1 = case Partitions -- Down -- NoLongerPartitioned of
                       [] -> [];
                       _  -> Partitions
@@ -898,8 +898,7 @@ disconnect(Node) ->
 
 %%--------------------------------------------------------------------
 
-%% mnesia:system_info(db_nodes) (and hence
-%% rabbit_nodes:all_running()) does not return all nodes
+%% mnesia:system_info(db_nodes) does not return all nodes
 %% when partitioned, just those that we are sharing Mnesia state
 %% with. So we have a small set of replacement functions
 %% here. "rabbit" in a function's name implies we test if the rabbit
@@ -914,7 +913,7 @@ majority() ->
     majority([]).
 
 majority(NodesDown) ->
-    Nodes = rabbit_nodes:all(),
+    Nodes = rabbit_nodes:list_members(),
     AliveNodes = alive_nodes(Nodes) -- NodesDown,
     length(AliveNodes) / length(Nodes) > 0.5.
 
@@ -927,44 +926,44 @@ in_preferred_partition(PreferredNodes) ->
     in_preferred_partition(PreferredNodes, []).
 
 in_preferred_partition(PreferredNodes, NodesDown) ->
-    Nodes = rabbit_nodes:all(),
+    Nodes = rabbit_nodes:list_members(),
     RealPreferredNodes = [N || N <- PreferredNodes, lists:member(N, Nodes)],
     AliveNodes = alive_nodes(RealPreferredNodes) -- NodesDown,
     RealPreferredNodes =:= [] orelse AliveNodes =/= [].
 
 all_nodes_up() ->
-    Nodes = rabbit_nodes:all(),
+    Nodes = rabbit_nodes:list_members(),
     length(alive_nodes(Nodes)) =:= length(Nodes).
 
 -spec all_rabbit_nodes_up() -> boolean().
 
 all_rabbit_nodes_up() ->
-    Nodes = rabbit_nodes:all(),
+    Nodes = rabbit_nodes:list_members(),
     length(alive_rabbit_nodes(Nodes)) =:= length(Nodes).
 
-alive_nodes() -> alive_nodes(rabbit_nodes:all()).
+alive_nodes() -> rabbit_nodes:list_reachable().
 
 -spec alive_nodes([node()]) -> [node()].
 
-alive_nodes(Nodes) -> [N || N <- Nodes, lists:member(N, [node()|nodes()])].
+alive_nodes(Nodes) -> rabbit_nodes:filter_reachable(Nodes).
 
-alive_rabbit_nodes() -> alive_rabbit_nodes(rabbit_nodes:all()).
+alive_rabbit_nodes() -> rabbit_nodes:list_running().
 
 -spec alive_rabbit_nodes([node()]) -> [node()].
 
 alive_rabbit_nodes(Nodes) ->
-    [N || N <- alive_nodes(Nodes), rabbit:is_running(N)].
+    rabbit_nodes:filter_running(Nodes).
 
 %% This one is allowed to connect!
 
 -spec ping_all() -> 'ok'.
 
 ping_all() ->
-    [net_adm:ping(N) || N <- rabbit_nodes:all()],
+    [net_adm:ping(N) || N <- rabbit_nodes:list_members()],
     ok.
 
 possibly_partitioned_nodes() ->
-    alive_rabbit_nodes() -- rabbit_nodes:all_running().
+    rabbit_nodes:list_unreachable().
 
 startup_log([]) ->
     rabbit_log:info("Starting rabbit_node_monitor", []);