Merge pull request #7058 from rabbitmq/add-node-lists-functions-to-clarify-intent

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

Author: michaelklishin

deps/rabbit/src/rabbit_amqqueue.erl

@@ -371,7 +371,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),
@@ -1048,7 +1048,7 @@ check_queue_type(_Val, _Args) ->
 
 list() ->
     All = rabbit_db_queue:get_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),
@@ -1230,7 +1230,7 @@ is_in_virtual_host(Q, VHostName) ->
 -spec list(vhost:name()) -> [amqqueue:amqqueue()].
 list(VHostPath) ->
     All = rabbit_db_queue:get_all(VHostPath),
-    NodesRunning = rabbit_nodes:all_running(),
+    NodesRunning = rabbit_nodes:list_running(),
     lists:filter(fun (Q) ->
                          Pid = amqqueue:get_pid(Q),
                          St = amqqueue:get_state(Q),
@@ -1244,7 +1244,7 @@ list_down(VHostPath) ->
         false -> [];
         true  ->
             Alive = sets:from_list([amqqueue:get_name(Q) || Q <- list(VHostPath)]),
-            NodesRunning = rabbit_nodes:all_running(),
+            NodesRunning = rabbit_nodes:list_running(),
             rabbit_db_queue:filter_all_durable(
               fun (Q) ->
                       N = amqqueue:get_name(Q),
@@ -1348,7 +1348,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),
@@ -1736,10 +1736,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_autoheal.erl

@@ -145,7 +145,7 @@ enabled() ->
     end.
 
 leader() ->
-    [Leader | _] = lists:usort(rabbit_nodes:all()),
+    [Leader | _] = lists:usort(rabbit_nodes:list_members()),
     Leader.
 
 %% This is the winner receiving its last notification that a node has
@@ -411,7 +411,7 @@ partition_value(Partition) ->
 %% only know which nodes we have been partitioned from, not which
 %% nodes are partitioned from each other.
 check_other_nodes(LocalPartitions) ->
-    Nodes = rabbit_nodes:all(),
+    Nodes = rabbit_nodes:list_members(),
     {Results, Bad} = rabbit_node_monitor:status(Nodes -- [node()]),
     RemotePartitions = [{Node, proplists:get_value(partitions, Res)}
                         || {Node, Res} <- Results],

deps/rabbit/src/rabbit_channel.erl

@@ -369,7 +369,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

@@ -433,7 +433,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(_, []) ->
@@ -468,15 +468,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
@@ -549,7 +549,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_core_metrics_gc.erl

@@ -102,7 +102,7 @@ gc_exchanges() ->
     gc_process_and_entity(channel_exchange_metrics, GbSet).
 
 gc_nodes() ->
-    Nodes = rabbit_nodes:all(),
+    Nodes = rabbit_nodes:list_members(),
     GbSet = gb_sets:from_list(Nodes),
     gc_entity(node_node_metrics, GbSet).
 

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_ff_controller.erl

@@ -914,21 +914,21 @@ post_enable(#{states_per_node := _}, FeatureName, Nodes) ->
 
 -ifndef(TEST).
 all_nodes() ->
-    lists:usort([node() | mnesia:system_info(db_nodes)]).
+    lists:usort([node() | rabbit_nodes:list_members()]).
 
 running_nodes() ->
-    lists:usort([node() | mnesia:system_info(running_db_nodes)]).
+    lists:usort([node() | rabbit_nodes:list_running()]).
 -else.
 all_nodes() ->
     RemoteNodes = case rabbit_feature_flags:get_overriden_nodes() of
-                      undefined -> mnesia:system_info(db_nodes);
+                      undefined -> rabbit_nodes:list_members();
                       Nodes     -> Nodes
                   end,
     lists:usort([node() | RemoteNodes]).
 
 running_nodes() ->
     RemoteNodes = case rabbit_feature_flags:get_overriden_running_nodes() of
-                      undefined -> mnesia:system_info(running_db_nodes);
+                      undefined -> rabbit_nodes:list_running();
                       Nodes     -> Nodes
                   end,
     lists:usort([node() | RemoteNodes]).

deps/rabbit/src/rabbit_maintenance.erl

@@ -290,7 +290,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),