notify cluster of new nodes as soon as they join, and not when they start

2 files changed, 38 insertions, 14 deletions

diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl
index 679f1c8481..2e0864454c 100644
--- a/src/rabbit_mnesia.erl
+++ b/src/rabbit_mnesia.erl
@@ -44,6 +44,7 @@
 
          on_node_up/2,
          on_node_down/1,
+         on_node_join/2,
          on_node_leave/1
         ]).
 
@@ -209,12 +210,14 @@ join_cluster(DiscoveryNode, WantDiscNode) ->
     %% user.
     reset(false),
 
-    rabbit_misc:local_info_msg("Clustering with ~p~s~n", [ClusterNodes]),
+    rabbit_misc:local_info_msg("Clustering with ~p~n", [ClusterNodes]),
 
     %% Join the cluster
     ok = init_db_and_upgrade(DiscNodes, WantDiscNode, false),
     stop_mnesia(),
 
+    rabbit_node_monitor:notify_join_cluster(),
+
     ok.
 
 %% return node to its virgin state, where it is not member of any
@@ -729,6 +732,16 @@ on_node_down(Node) ->
     write_cluster_nodes_status({AllNodes, DiscNodes,
                                 ordsets:del_element(Node, RunningNodes)}).
 
+on_node_join(Node, IsDiscNode) ->
+    {AllNodes, DiscNodes, RunningNodes} = read_cluster_nodes_status(),
+    write_cluster_nodes_status({ordsets:add_element(Node, AllNodes),
+                                case IsDiscNode of
+                                    true  -> ordsets:add_element(Node,
+                                                                 DiscNodes);
+                                    false -> DiscNodes
+                                end,
+                                RunningNodes}).
+
 on_node_leave(Node) ->
     {AllNodes, DiscNodes, RunningNodes} = read_cluster_nodes_status(),
     write_cluster_nodes_status({ordsets:del_element(Node, AllNodes),
diff --git a/src/rabbit_node_monitor.erl b/src/rabbit_node_monitor.erl
index a25d0fe04d..342c744331 100644
--- a/src/rabbit_node_monitor.erl
+++ b/src/rabbit_node_monitor.erl
@@ -20,10 +20,10 @@
 
 -export([start_link/0]).
 
--export([init/1, handle_call/3, handle_cast/2, handle_info/2,
-         terminate/2, code_change/3]).
--export([rabbit_running_on/2, rabbit_left_cluster/1, notify_cluster/0,
-         notify_leave_cluster/1]).
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
+         code_change/3]).
+-export([rabbit_running_on/2, rabbit_left_cluster/1, rabbit_joined_cluster/2,
+         notify_cluster/0, notify_join_cluster/0, notify_leave_cluster/1]).
 
 -define(SERVER, ?MODULE).
 -define(RABBIT_UP_RPC_TIMEOUT, 2000).
@@ -55,28 +55,36 @@ rabbit_running_on(Node) ->
 rabbit_left_cluster(Node) ->
     gen_server:cast(rabbit_node_monitor, {rabbit_left_cluster, Node}).
 
-notify_cluster() ->
+rabbit_joined_cluster(Node, IsDiscNode) ->
+    gen_server:cast(rabbit_node_monitor,
+                    {rabbit_joined_cluster, Node, IsDiscNode}).
+
+cluster_multicall(Fun, Args) ->
     Node = node(),
     Nodes = rabbit_mnesia:running_clustered_nodes() -- [Node],
     %% notify other rabbits of this rabbit
-    case rpc:multicall(Nodes, rabbit_node_monitor, rabbit_running_on,
-                       [Node, rabbit_mnesia:is_disc_node()],
+    case rpc:multicall(Nodes, rabbit_node_monitor, Fun, Args,
                        ?RABBIT_UP_RPC_TIMEOUT) of
         {_, [] } -> ok;
         {_, Bad} -> rabbit_log:info("failed to contact nodes ~p~n", [Bad])
     end,
+    Nodes.
+
+notify_cluster() ->
+    Nodes = cluster_multicall(rabbit_running_on,
+                              [node(), rabbit_mnesia:is_disc_node()]),
     %% register other active rabbits with this rabbit
     [ rabbit_running_on(N) || N <- Nodes ],
     ok.
 
+notify_join_cluster() ->
+    cluster_multicall(rabbit_joined_cluster,
+                      [node(), rabbit_mnesia:is_disc_node()]),
+    ok.
+
 notify_leave_cluster(Node) ->
-    Nodes = rabbit_mnesia:running_clustered_nodes() -- [node()],
     rabbit_left_cluster(Node),
-    case rpc:multicall(Nodes, rabbit_node_monitor, rabbit_left_cluster, [Node],
-                       ?RABBIT_UP_RPC_TIMEOUT) of
-        {_, [] } -> ok;
-        {_, Bad} -> rabbit_log:info("failed to contact nodes ~p~n", [Bad])
-    end,
+    cluster_multicall(rabbit_left_cluster, [Node]),
     ok.
 
 %%--------------------------------------------------------------------
@@ -95,6 +103,9 @@ handle_cast({rabbit_running_on, Node, IsDiscNode}, Nodes) ->
                  ok = handle_live_rabbit(Node, IsDiscNode),
                  {noreply, ordsets:add_element(Node, Nodes)}
     end;
+handle_cast({rabbit_joined_cluster, Node, IsDiscNode}, Nodes) ->
+    ok = rabbit_mnesia:on_node_join(Node, IsDiscNode),
+    {noreply, Nodes};
 handle_cast({rabbit_left_cluster, Node}, Nodes) ->
     ok = rabbit_mnesia:on_node_leave(Node),
     {noreply, Nodes};