Merge pull request #1175 from rabbitmq/rabbitmq-server-1143

Register with peer discovery backend on start and stop

Author: michaelklishin

src/rabbit.erl

@@ -22,7 +22,7 @@
          stop_and_halt/0, await_startup/0, status/0, is_running/0, alarms/0,
          is_running/1, environment/0, rotate_logs/0, force_event_refresh/1,
          start_fhc/0]).
--export([start/2, stop/1]).
+-export([start/2, stop/1, prep_stop/1]).
 -export([start_apps/1, stop_apps/1]).
 -export([log_locations/0, config_files/0, decrypt_config/2]). %% for testing and mgmt-agent
 
@@ -327,7 +327,9 @@ broker_start() ->
     ToBeLoaded = Plugins ++ ?APPS,
     start_apps(ToBeLoaded),
     maybe_sd_notify(),
-    ok = log_broker_started(rabbit_plugins:strictly_plugins(rabbit_plugins:active())).
+    ok = log_broker_started(rabbit_plugins:strictly_plugins(rabbit_plugins:active())),
+    rabbit_peer_discovery:maybe_register(),
+    ok.
 
 %% Try to send systemd ready notification if it makes sense in the
 %% current environment. standard_error is used intentionally in all
@@ -471,6 +473,8 @@ stop() ->
     end,
     rabbit_log:info("RabbitMQ is asked to stop...~n", []),
     Apps = ?APPS ++ rabbit_plugins:active(),
+    %% this will also perform unregistration with the peer discovery backend
+    %% as needed
     stop_apps(app_utils:app_dependency_order(Apps, true)),
     rabbit_log:info("Successfully stopped RabbitMQ and its dependencies~n", []).
 
@@ -759,6 +763,10 @@ start(normal, []) ->
             Error
     end.
 
+prep_stop(State) ->
+  rabbit_peer_discovery:maybe_unregister(),
+  State.
+
 stop(_State) ->
     ok = rabbit_alarm:stop(),
     ok = case rabbit_mnesia:is_clustered() of

src/rabbit_mnesia.erl

@@ -139,14 +139,15 @@ init_from_config() ->
             {ok, _} ->
                 e(invalid_cluster_nodes_conf)
         end,
-    case DiscoveredNodes of
+    Peers = nodes_excl_me(DiscoveredNodes),
+    case Peers of
         [] ->
             rabbit_log:info("Discovered no peer nodes to cluster with"),
             init_db_and_upgrade([node()], disc, false, _Retry = true);
         _  ->
             rabbit_log:info("Discovered peer nodes: ~s~n",
-                [rabbit_peer_discovery:format_discovered_nodes(DiscoveredNodes)]),
-            join_discovered_peers(DiscoveredNodes, NodeType)
+                [rabbit_peer_discovery:format_discovered_nodes(Peers)]),
+            join_discovered_peers(Peers, NodeType)
     end.
 
 %% Attempts to join discovered,

src/rabbit_peer_discovery.erl

@@ -21,7 +21,8 @@
 %%
 
 -export([discover_cluster_nodes/0, backend/0, node_type/0,
-         normalize/1, format_discovered_nodes/1, log_configured_backend/0]).
+         normalize/1, format_discovered_nodes/1, log_configured_backend/0,
+         register/0, unregister/0, maybe_register/0, maybe_unregister/0]).
 -export([append_node_prefix/1, node_prefix/0]).
 
 -define(DEFAULT_BACKEND,   rabbit_peer_discovery_classic_config).
@@ -72,6 +73,64 @@ discover_cluster_nodes() ->
     normalize(Backend:list_nodes()).
 
 
+-spec maybe_register() -> ok.
+
+maybe_register() ->
+  Backend = backend(),
+  case Backend:supports_registration() of
+    true  ->
+      register();
+    false ->
+      rabbit_log:info("Peer discovery backend ~s does not support registration, skipping registration.", [Backend]),
+      ok
+  end.
+
+
+-spec maybe_unregister() -> ok.
+
+maybe_unregister() ->
+  Backend = backend(),
+  case Backend:supports_registration() of
+    true  ->
+      unregister();
+    false ->
+      rabbit_log:info("Peer discovery backend ~s does not support registration, skipping unregistration.", [Backend]),
+      ok
+  end.
+
+
+-spec register() -> ok.
+
+register() ->
+  Backend = backend(),
+  rabbit_log:info("Will register with peer discovery backend ~s", [Backend]),
+  case Backend:register() of
+    ok             -> ok;
+    {error, Error} ->
+      rabbit_log:error("Failed to register with peer discovery backend ~s: ~p",
+        [Backend, Error]),
+      ok
+  end.
+
+
+-spec unregister() -> ok.
+
+unregister() ->
+  Backend = backend(),
+  rabbit_log:info("Will unregister with peer discovery backend ~s", [Backend]),
+  case Backend:unregister() of
+    ok             -> ok;
+    {error, Error} ->
+      rabbit_log:error("Failed to unregister with peer discovery backend ~s: ~p",
+        [Backend, Error]),
+      ok
+  end.
+
+
+%%
+%% Implementation
+%%
+
 -spec normalize(Nodes :: list() |
                 {Nodes :: list(), NodeType :: rabbit_types:node_type()} |
                 {ok, Nodes :: list()} |
@@ -90,7 +149,6 @@ normalize({ok, {Nodes, NodeType}}) when is_list(Nodes) andalso is_atom(NodeType)
 normalize({error, Reason}) ->
   {error, Reason}.
 
-
 -spec format_discovered_nodes(Nodes :: list()) -> string().
 
 format_discovered_nodes(Nodes) ->

src/rabbit_peer_discovery_classic_config.erl

@@ -19,7 +19,7 @@
 
 -include("rabbit.hrl").
 
--export([list_nodes/0, register/0, unregister/0]).
+-export([list_nodes/0, supports_registration/0, register/0, unregister/0]).
 
 %%
 %% API
@@ -34,6 +34,11 @@ list_nodes() ->
         undefined                  -> {[], disc}
     end.
 
+-spec supports_registration() -> boolean().
+
+supports_registration() ->
+    false.
+
 -spec register() -> ok.
 
 register() ->

src/rabbit_peer_discovery_dns.erl

@@ -19,7 +19,7 @@
 
 -include("rabbit.hrl").
 
--export([list_nodes/0, register/0, unregister/0]).
+-export([list_nodes/0, supports_registration/0, register/0, unregister/0]).
 %% for tests
 -export([discover_nodes/2, discover_hostnames/2]).
 
@@ -48,6 +48,13 @@ list_nodes() ->
         end
     end.
 
+
+-spec supports_registration() -> boolean().
+
+supports_registration() ->
+    false.
+
+
 -spec register() -> ok.
 
 register() ->