Merge pull request #1008 from rabbitmq/rabbitmq-server-988

DNS peer discovery backend

7 files changed, 366 insertions, 27 deletions

diff --git a/docs/rabbitmq.conf.example b/docs/rabbitmq.conf.example
index db65572c2a..33685191a3 100644
--- a/docs/rabbitmq.conf.example
+++ b/docs/rabbitmq.conf.example
@@ -307,14 +307,24 @@
 ## See http://www.rabbitmq.com/clustering.html#auto-config for
 ## further details.
 ##
+
+# autocluster.peer_discovery_backend     = rabbit_peer_discovery_classic_config
+#
 # autocluster.classic_config.nodes.node1 = rabbit1@hostname
 # autocluster.classic_config.nodes.node2 = rabbit2@hostname
 # autocluster.classic_config.nodes.node3 = rabbit3@hostname
 # autocluster.classic_config.nodes.node4 = rabbit4@hostname
 
+## DNS-based peer discovery. This backend will list A records
+## of the configured hostname and perform reverse lookups for
+## the addresses returned.
+
+# autocluster.peer_discovery_backend = rabbit_peer_discovery_dns
+# autocluster.dns.hostname           = rabbitmq.discovery.mycompany.local
+
 ## This node's type can be configured. If you are not sure
 ## what node type to use, always use 'disc'.
-# autocluster.classic_config.node_type = disc
+# autocluster.node_type = disc
 
 ## Interval (in milliseconds) at which we send keepalive messages
 ## to other cluster members. Note that this is not the same thing
diff --git a/priv/schema/rabbitmq.schema b/priv/schema/rabbitmq.schema
index 150a26b60d..f31ec5416c 100644
--- a/priv/schema/rabbitmq.schema
+++ b/priv/schema/rabbitmq.schema
@@ -757,6 +757,39 @@ end}.
 {mapping, "mirroring_sync_batch_size", "rabbit.mirroring_sync_batch_size",
     [{datatype, bytesize}, {validators, ["size_less_than_2G"]}]}.
 
+%% Peer discovery backend used by autoclustering.
+%%
+
+{mapping, "autocluster.peer_discovery_backend", "rabbit.autocluster.peer_discovery_backend", [
+    {datatype, atom}
+]}.
+
+%% Own node type used by autoclustering.
+%%
+
+{mapping, "autocluster.node_type", "rabbit.autocluster.node_type", [
+    {datatype, {enum, [disc, disk, ram]}}
+]}.
+
+{translation, "rabbit.autocluster.node_type",
+fun(Conf) ->
+    %% if peer discovery backend isn't configured, don't generate
+    %% node type
+    case cuttlefish:conf_get("autocluster.peer_discovery_backend", Conf, undefined) of
+        undefined -> cuttlefish:unset();
+        _Backend  ->
+            case cuttlefish:conf_get("autocluster.node_type", Conf) of
+                disc      -> disc;
+                %% always cast to `disc`
+                disk      -> disc;
+                ram       -> ram;
+                _Other    -> disc
+            end
+    end
+end}.
+
+%% Classic config-driven autocluster backend.
+%%
 %% Make clustering happen *automatically* at startup - only applied
 %% to nodes that have just been reset or started for the first time.
 %% See http://www.rabbitmq.com/clustering.html#auto-config for
@@ -767,11 +800,6 @@ end}.
 {mapping, "autocluster.classic_config.nodes.$node", "rabbit.cluster_nodes",
     [{datatype, atom}]}.
 
-{mapping, "autocluster.classic_config.node_type", "rabbit.cluster_nodes", [
-    {datatype, {enum, [disc, disk, ram]}},
-    {default,  disc}
-]}.
-
 {translation, "rabbit.cluster_nodes",
 fun(Conf) ->
     Nodes = [V || {_, V} <- cuttlefish_variable:filter_by_prefix("autocluster.classic_config.nodes", Conf)],
@@ -779,15 +807,29 @@ fun(Conf) ->
     case Nodes of
         []    -> cuttlefish:unset();
         Other ->
-            case cuttlefish:conf_get("autocluster.classic_config.node_type", Conf, disc) of
+            case cuttlefish:conf_get("autocluster.node_type", Conf, disc) of
                 disc -> {Other, disc};
-                % Always cast to `disc`
+                %% Always cast to `disc`
                 disk -> {Other, disc};
                 ram  -> {Other, ram}
             end
     end
 end}.
 
+%% DNS (A records and reverse lookups)-based peer discovery.
+%%
+
+{mapping, "autocluster.dns.hostname", "rabbit.autocluster.peer_discovery_dns.hostname",
+    [{datatype, string}]}.
+
+{translation, "rabbit.autocluster.peer_discovery_dns.hostname",
+fun(Conf) ->
+    case cuttlefish:conf_get("autocluster.dns.hostname", Conf, undefined) of
+        undefined -> cuttlefish:unset();
+        Value     -> list_to_binary(Value)
+    end
+end}.
+
 
 %% Interval (in milliseconds) at which we send keepalive messages
 %% to other cluster members. Note that this is not the same thing
@@ -799,6 +841,8 @@ end}.
 {mapping, "cluster_keepalive_interval", "rabbit.cluster_keepalive_interval",
     [{datatype, integer}]}.
 
+%% Queue master locator
+%%
 
 {mapping, "queue_master_locator", "rabbit.queue_master_locator",
     [{datatype, string}]}.
diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl
index 1ec9a46880..51deed8597 100644
--- a/src/rabbit_mnesia.erl
+++ b/src/rabbit_mnesia.erl
@@ -98,11 +98,10 @@ init() ->
     ensure_mnesia_dir(),
     case is_virgin_node() of
         true  ->
-            rabbit_log:info("Database directory at ~s is empty. "
+            rabbit_log:info("Node database directory at ~s is empty. "
                             "Assuming we need to join an existing cluster or initialise from scratch...~n",
                             [dir()]),
-            rabbit_log:info("Using ~p as peer discovery backend~n",
-                            [rabbit_peer_discovery:backend()]),
+             rabbit_peer_discovery:log_configured_backend(),
             init_from_config();
         false ->
             NodeType = node_type(),
@@ -141,7 +140,9 @@ init_from_config() ->
                 e(invalid_cluster_nodes_conf)
         end,
     case DiscoveredNodes of
-        [] -> init_db_and_upgrade([node()], disc, false, _Retry = true);
+        [] ->
+            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)]),
@@ -158,8 +159,9 @@ auto_cluster(TryNodes, NodeType) ->
             rabbit_node_monitor:notify_joined_cluster();
         none ->
             rabbit_log:warning(
-              "Could not find any node for auto-clustering from: ~p~n"
-              "Starting blank node...~n", [TryNodes]),
+              "Could not successfully contact any node of: ~s (as in Erlang distribution). "
+               "Starting as a blank standalone node...~n",
+                [string:join(lists:map(fun atom_to_list/1, TryNodes), ",")]),
             init_db_and_upgrade([node()], disc, false, _Retry = true)
     end.
 
diff --git a/src/rabbit_peer_discovery.erl b/src/rabbit_peer_discovery.erl
index 965be3946d..c0b554e3b3 100644
--- a/src/rabbit_peer_discovery.erl
+++ b/src/rabbit_peer_discovery.erl
@@ -16,21 +16,53 @@
 
 -module(rabbit_peer_discovery).
 
+%%
 %% API
--export([discover_cluster_nodes/0, backend/0,
-         normalize/1, format_discovered_nodes/1]).
+%%
+
+-export([discover_cluster_nodes/0, backend/0, node_type/0,
+         normalize/1, format_discovered_nodes/1, log_configured_backend/0]).
+-export([append_node_prefix/1, node_prefix/0]).
 
+-define(DEFAULT_BACKEND,   rabbit_peer_discovery_classic_config).
+%% what node type is used by default for this node when joining
+%% a new cluster as a virgin node
+-define(DEFAULT_NODE_TYPE, disc).
+%% default node prefix to attach to discovered hostnames
+-define(DEFAULT_PREFIX, "rabbit").
+-define(NODENAME_PART_SEPARATOR, "@").
 
 
 -spec backend() -> atom().
 
 backend() ->
-  case application:get_env(rabbit, peer_discovery_backend) of
-    {ok, Backend} when is_atom(Backend) -> Backend;
-    undefined                           -> rabbit_peer_discovery_classic_config
+  case application:get_env(rabbit, autocluster) of
+    {ok, Proplist} ->
+      proplists:get_value(peer_discovery_backend, Proplist, ?DEFAULT_BACKEND);
+    undefined      ->
+      ?DEFAULT_BACKEND
   end.
 
 
+
+-spec node_type() -> rabbit_types:node_type().
+
+node_type() ->
+  case application:get_env(rabbit, autocluster) of
+    {ok, Proplist} ->
+      proplists:get_value(node_type, Proplist, ?DEFAULT_NODE_TYPE);
+    undefined      ->
+      ?DEFAULT_NODE_TYPE
+  end.
+
+
+
+-spec log_configured_backend() -> ok.
+
+log_configured_backend() ->
+  rabbit_log:info("Configured peer discovery backend: ~s~n", [backend()]).
+
+
 -spec discover_cluster_nodes() -> {ok, Nodes :: list()} |
                                   {ok, {Nodes :: list(), NodeType :: rabbit_types:node_type()}} |
                                   {error, Reason :: string()}.
@@ -40,11 +72,17 @@ discover_cluster_nodes() ->
     normalize(Backend:list_nodes()).
 
 
--spec normalize({ok, Nodes :: list()} |
+-spec normalize(Nodes :: list() |
+                {Nodes :: list(), NodeType :: rabbit_types:node_type()} |
+                {ok, Nodes :: list()} |
                 {ok, {Nodes :: list(), NodeType :: rabbit_types:node_type()}} |
                 {error, Reason :: string()}) -> {ok, {Nodes :: list(), NodeType :: rabbit_types:node_type()}} |
                                                 {error, Reason :: string()}.
 
+normalize(Nodes) when is_list(Nodes) ->
+  {ok, {Nodes, disc}};
+normalize({Nodes, NodeType}) when is_list(Nodes) andalso is_atom(NodeType) ->
+  {ok, {Nodes, NodeType}};
 normalize({ok, Nodes}) when is_list(Nodes) ->
   {ok, {Nodes, disc}};
 normalize({ok, {Nodes, NodeType}}) when is_list(Nodes) andalso is_atom(NodeType) ->
@@ -57,3 +95,27 @@ normalize({error, Reason}) ->
 
 format_discovered_nodes(Nodes) ->
   string:join(lists:map(fun (Val) -> hd(io_lib:format("~s", [Val])) end, Nodes), ", ").
+
+
+
+-spec node_prefix() -> string().
+
+node_prefix() ->
+    case string:tokens(atom_to_list(node()), ?NODENAME_PART_SEPARATOR) of
+        [Prefix, _] -> Prefix;
+        [_]         -> ?DEFAULT_PREFIX
+    end.
+
+
+
+-spec append_node_prefix(Value :: binary() | list()) -> atom().
+
+append_node_prefix(Value) ->
+    Val = rabbit_data_coercion:to_list(Value),
+    Hostname = case string:tokens(Val, ?NODENAME_PART_SEPARATOR) of
+                   [_ExistingPrefix, Val] ->
+                       Val;
+                   [Val]                  ->
+                       Val
+               end,
+    string:join([node_prefix(), Hostname], ?NODENAME_PART_SEPARATOR).
diff --git a/src/rabbit_peer_discovery_dns.erl b/src/rabbit_peer_discovery_dns.erl
new file mode 100644
index 0000000000..554c914466
--- /dev/null
+++ b/src/rabbit_peer_discovery_dns.erl
@@ -0,0 +1,88 @@
+%% The contents of this file are subject to the Mozilla Public License
+%% Version 1.1 (the "License"); you may not use this file except in
+%% compliance with the License. You may obtain a copy of the License at
+%% http://www.mozilla.org/MPL/
+%%
+%% Software distributed under the License is distributed on an "AS IS"
+%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
+%% License for the specific language governing rights and limitations
+%% under the License.
+%%
+%% The Original Code is RabbitMQ.
+%%
+%% The Initial Developer of the Original Code is GoPivotal, Inc.
+%% Copyright (c) 2007-2016 Pivotal Software, Inc.  All rights reserved.
+%%
+
+-module(rabbit_peer_discovery_dns).
+-behaviour(rabbit_peer_discovery_backend).
+
+-include("rabbit.hrl").
+
+-export([list_nodes/0, register/0, unregister/0]).
+%% for tests
+-export([discover_nodes/2, discover_hostnames/2]).
+
+%%
+%% API
+%%
+
+-spec list_nodes() -> {ok, Nodes :: list()} | {error, Reason :: string()}.
+
+list_nodes() ->
+    case application:get_env(rabbit, autocluster) of
+      undefined         ->
+        {[], disc};
+      {ok, Autocluster} ->
+        case proplists:get_value(peer_discovery_dns, Autocluster) of
+            undefined ->
+              rabbit_log:warning("Peer discovery backend is set to ~s "
+                                 "but final config does not contain rabbit.autocluster.peer_discovery_dns. "
+                                 "Cannot discover any nodes because seed hostname is not configured!",
+                                 [?MODULE]),
+              {[], disc};
+            Proplist  ->
+              Hostname = rabbit_data_coercion:to_list(proplists:get_value(hostname, Proplist)),
+
+              {discover_nodes(Hostname, net_kernel:longnames()), rabbit_peer_discovery:node_type()}
+        end
+    end.
+
+-spec register() -> ok.
+
+register() ->
+    ok.
+
+-spec unregister() -> ok.
+
+unregister() ->
+    ok.
+
+
+%%
+%% Implementation
+%%
+
+discover_nodes(SeedHostname, LongNamesUsed) ->
+    [list_to_atom(rabbit_peer_discovery:append_node_prefix(H)) ||
+        H <- discover_hostnames(SeedHostname, LongNamesUsed)].
+
+discover_hostnames(SeedHostname, LongNamesUsed) ->
+    %% TODO: IPv6 support
+    IPs   = inet_res:lookup(SeedHostname, in, a),
+    rabbit_log:info("Addresses discovered via A records of ~s: ~s",
+      [SeedHostname, string:join([inet_parse:ntoa(IP) || IP <- IPs], ", ")]),
+    Hosts = [extract_host(inet_res:gethostbyaddr(A), LongNamesUsed, A) ||
+                A <- IPs],
+    lists:filter(fun(E) -> E =/= error end, Hosts).
+
+%% long node names are used
+extract_host({ok, {hostent, FQDN, _, _, _, _}}, true, _Address) ->
+  FQDN;
+%% short node names are used
+extract_host({ok, {hostent, FQDN, _, _, _, _}}, false, _Address) ->
+  lists:nth(1, string:tokens(FQDN, "."));
+extract_host({error, Error}, _, Address) ->
+  rabbit_log:error("Reverse DNS lookup for address ~s failed: ~p",
+                   [inet_parse:ntoa(Address), Error]),
+  error.
diff --git a/test/config_schema_SUITE_data/snippets.config b/test/config_schema_SUITE_data/snippets.config
index 8c4319fa53..a64b6e06c9 100644
--- a/test/config_schema_SUITE_data/snippets.config
+++ b/test/config_schema_SUITE_data/snippets.config
@@ -94,23 +94,29 @@ default_permissions.write = .*",
 {default_permissions, [<<".*">>, <<".*">>, <<".*">>]}]}],[]}
 ,
 {13,
-"autocluster.classic_config.nodes.peer1 = rabbit@hostname1
+"autocluster.peer_discovery_backend = rabbit_peer_discovery_classic_config
+autocluster.classic_config.nodes.peer1 = rabbit@hostname1
 autocluster.classic_config.nodes.peer2 = rabbit@hostname2
-autocluster.classic_config.node_type = disc",
+autocluster.node_type = disc",
 [{rabbit, [
+    {autocluster,   [{peer_discovery_backend, rabbit_peer_discovery_classic_config},
+                     {node_type, disc}]},
     {cluster_nodes, {[rabbit@hostname2,rabbit@hostname1], disc}}
 ]}],[]}
 ,
 {13.1,
-"autocluster.classic_config.nodes.peer1 = rabbit@hostname1
+"autocluster.peer_discovery_backend = rabbit_peer_discovery_classic_config
+autocluster.classic_config.nodes.peer1 = rabbit@hostname1
 autocluster.classic_config.nodes.peer2 = rabbit@hostname2
-autocluster.classic_config.node_type = disk",
+autocluster.node_type = disk",
 [{rabbit, [
+    {autocluster,   [{peer_discovery_backend, rabbit_peer_discovery_classic_config},
+                     {node_type, disc}]},
     {cluster_nodes, {[rabbit@hostname2,rabbit@hostname1], disc}}
 ]}],[]}
 ,
 {13.2,
-"autocluster.classic_config.node_type = ram",
+"autocluster.node_type = ram",
 [],[]}
 ,
 {14,
@@ -713,7 +719,7 @@ web_stomp.ssl.password   = changeme",
 [{rabbitmq_web_stomp,
     [{sockjs_opts, [{sockjs_url, "https://cdn.jsdelivr.net/sockjs/0.3.4/sockjs.min.js"}]}]}],
 [rabbitmq_web_stomp]},
-{69, 
+{69,
 "auth_backends.1 = http
 rabbitmq_auth_backend_http.user_path     = http://some-server/auth/user
 rabbitmq_auth_backend_http.vhost_path    = http://some-server/auth/vhost
@@ -741,5 +747,30 @@ tcp_listen_options.linger.timeout = 100",
 {74,
 "tcp_listen_options.linger.timeout = 100",
 [{rabbit, [{tcp_listen_options, [{linger, {false, 100}}]}]}],
-[]}
+[]},
+{75,
+"
+autocluster.peer_discovery_backend = rabbit_peer_discovery_dns
+autocluster.dns.hostname = 192.168.0.2.xip.io
+autocluster.node_type = disc",
+[{rabbit, [
+    {autocluster, [{peer_discovery_dns,     [{hostname,  <<"192.168.0.2.xip.io">>}]},
+                   {peer_discovery_backend, rabbit_peer_discovery_dns},
+                   {node_type,              disc}]}
+]}],[]}
+,
+{76,
+"autocluster.peer_discovery_backend = rabbit_peer_discovery_classic_config
+autocluster.node_type = disc",
+[{rabbit, [
+    {autocluster,        [{peer_discovery_backend, rabbit_peer_discovery_classic_config},
+                          {node_type, disc}]}
+]}],[]},
+{76.1,
+"autocluster.peer_discovery_backend = rabbit_peer_discovery_classic_config
+autocluster.node_type = ram",
+[{rabbit, [
+    {autocluster, [{peer_discovery_backend, rabbit_peer_discovery_classic_config},
+                   {node_type, ram}]}
+]}],[]}
 ].
diff --git a/test/peer_discovery_dns_SUITE.erl b/test/peer_discovery_dns_SUITE.erl
new file mode 100644
index 0000000000..da01aeebdd
--- /dev/null
+++ b/test/peer_discovery_dns_SUITE.erl
@@ -0,0 +1,102 @@
+%% The contents of this file are subject to the Mozilla Public License
+%% Version 1.1 (the "License"); you may not use this file except in
+%% compliance with the License. You may obtain a copy of the License at
+%% http://www.mozilla.org/MPL/
+%%
+%% Software distributed under the License is distributed on an "AS IS"
+%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
+%% License for the specific language governing rights and limitations
+%% under the License.
+%%
+%% The Original Code is RabbitMQ.
+%%
+%% The Initial Developer of the Original Code is GoPivotal, Inc.
+%% Copyright (c) 2011-2016 Pivotal Software, Inc.  All rights reserved.
+%%
+
+-module(peer_discovery_dns_SUITE).
+
+-include_lib("common_test/include/ct.hrl").
+-include_lib("amqp_client/include/amqp_client.hrl").
+-include_lib("eunit/include/eunit.hrl").
+
+-compile(export_all).
+
+all() ->
+    [
+     {group, non_parallel}
+    ].
+
+groups() ->
+    [
+      {non_parallel, [], [
+                          hostname_discovery_with_long_node_names,
+                          hostname_discovery_with_short_node_names,
+                          node_discovery_with_long_node_names,
+                          node_discovery_with_short_node_names
+                         ]}
+    ].
+
+suite() ->
+    [
+      %% If a test hangs, no need to wait for 30 minutes.
+      {timetrap, {minutes, 1}}
+    ].
+
+
+%% -------------------------------------------------------------------
+%% Testsuite setup/teardown.
+%% -------------------------------------------------------------------
+
+%% peer_discovery.tests.rabbitmq.net used in the tests below
+%% returns three A records two of which fail our resolution process:
+%%
+%% * One does not resolve to a [typically] non-reachable IP
+%% * One does not support reverse lookup queries
+
+-define(DISCOVERY_ENDPOINT, "peer_discovery.tests.rabbitmq.net").
+
+init_per_suite(Config) ->
+    rabbit_ct_helpers:log_environment(),
+    rabbit_ct_helpers:run_setup_steps(Config).
+
+end_per_suite(Config) ->
+    rabbit_ct_helpers:run_teardown_steps(Config).
+
+init_per_testcase(_Testcase, Config) ->
+    %% TODO: support IPv6-only environments
+    case inet_res:lookup(?DISCOVERY_ENDPOINT, in, a) of
+        []      ->
+            {skip, "pre-configured *.rabbitmq.net record does not resolve, skipping"};
+        [_ | _] ->
+            Config
+    end.
+
+end_per_testcase(_Testcase, Config) ->
+    case inet_res:lookup(?DISCOVERY_ENDPOINT, in, a) of
+        []      ->
+            {skip, "pre-configured *.rabbitmq.net record does not resolve, skipping"};
+        [_ | _] ->
+            Config
+    end.
+
+
+%% -------------------------------------------------------------------
+%% Test cases
+%% -------------------------------------------------------------------
+
+hostname_discovery_with_long_node_names(_) ->
+    Result = rabbit_peer_discovery_dns:discover_hostnames(?DISCOVERY_ENDPOINT, true),
+    ?assertEqual(["www.rabbitmq.com"], Result).
+
+hostname_discovery_with_short_node_names(_) ->
+    Result = rabbit_peer_discovery_dns:discover_hostnames(?DISCOVERY_ENDPOINT, false),
+    ?assertEqual(["www"], Result).
+
+node_discovery_with_long_node_names(_) ->
+    Result = rabbit_peer_discovery_dns:discover_nodes(?DISCOVERY_ENDPOINT, true),
+    ?assertEqual(['ct_rabbit@www.rabbitmq.com'], Result).
+
+node_discovery_with_short_node_names(_) ->
+    Result = rabbit_peer_discovery_dns:discover_nodes(?DISCOVERY_ENDPOINT, false),
+    ?assertEqual([ct_rabbit@www], Result).