Merge pull request #13027 from rjernst/cast_away_multicast

Move multicast discovery to a plugin

core/src/main/java/org/elasticsearch/common/util/ExtensionPoint.java

@@ -187,7 +187,7 @@ public abstract class ExtensionPoint {
         protected final void bindExtensions(Binder binder) {
             Multibinder<T> allocationMultibinder = Multibinder.newSetBinder(binder, extensionClass);
             for (Class<? extends T> clazz : extensions) {
-                allocationMultibinder.addBinding().to(clazz);
+                allocationMultibinder.addBinding().to(clazz).asEagerSingleton();
             }
         }
     }

core/src/main/java/org/elasticsearch/discovery/DiscoveryModule.java

@@ -19,18 +19,20 @@
 
 package org.elasticsearch.discovery;
 
-import com.google.common.collect.Lists;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.inject.AbstractModule;
 import org.elasticsearch.common.inject.multibindings.Multibinder;
-import org.elasticsearch.common.logging.Loggers;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.ExtensionPoint;
 import org.elasticsearch.discovery.local.LocalDiscovery;
 import org.elasticsearch.discovery.zen.ZenDiscovery;
 import org.elasticsearch.discovery.zen.elect.ElectMasterService;
+import org.elasticsearch.discovery.zen.ping.ZenPing;
 import org.elasticsearch.discovery.zen.ping.ZenPingService;
 import org.elasticsearch.discovery.zen.ping.unicast.UnicastHostsProvider;
+import org.elasticsearch.discovery.zen.ping.unicast.UnicastZenPing;
 
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -44,7 +46,8 @@ public class DiscoveryModule extends AbstractModule {
     public static final String ZEN_MASTER_SERVICE_TYPE_KEY = "discovery.zen.masterservice.type";
 
     private final Settings settings;
-    private final List<Class<? extends UnicastHostsProvider>> unicastHostProviders = Lists.newArrayList();
+    private final List<Class<? extends UnicastHostsProvider>> unicastHostProviders = new ArrayList<>();
+    private final ExtensionPoint.ClassSet<ZenPing> zenPings = new ExtensionPoint.ClassSet<>("zen_ping", ZenPing.class);
     private final Map<String, Class<? extends Discovery>> discoveryTypes = new HashMap<>();
     private final Map<String, Class<? extends ElectMasterService>> masterServiceType = new HashMap<>();
 
@@ -53,6 +56,8 @@ public class DiscoveryModule extends AbstractModule {
         addDiscoveryType("local", LocalDiscovery.class);
         addDiscoveryType("zen", ZenDiscovery.class);
         addElectMasterService("zen", ElectMasterService.class);
+        // always add the unicast hosts, or things get angry!
+        addZenPing(UnicastZenPing.class);
     }
 
     /**
@@ -82,6 +87,10 @@ public class DiscoveryModule extends AbstractModule {
         this.masterServiceType.put(type, masterService);
     }
 
+    public void addZenPing(Class<? extends ZenPing> clazz) {
+        zenPings.registerExtension(clazz);
+    }
+
     @Override
     protected void configure() {
         String defaultType = DiscoveryNode.localNode(settings) ? "local" : "zen";
@@ -107,6 +116,7 @@ public class DiscoveryModule extends AbstractModule {
             for (Class<? extends UnicastHostsProvider> unicastHostProvider : unicastHostProviders) {
                 unicastHostsProviderMultibinder.addBinding().to(unicastHostProvider);
             }
+            zenPings.bind(binder());
         }
         bind(Discovery.class).to(discoveryClass).asEagerSingleton();
         bind(DiscoveryService.class).asEagerSingleton();

core/src/main/java/org/elasticsearch/discovery/zen/ping/ZenPingService.java

@@ -19,50 +19,31 @@
 
 package org.elasticsearch.discovery.zen.ping;
 
-import com.google.common.collect.ImmutableList;
-import org.elasticsearch.Version;
-import org.elasticsearch.cluster.ClusterName;
-import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.component.AbstractLifecycleComponent;
 import org.elasticsearch.common.inject.Inject;
-import org.elasticsearch.common.network.NetworkService;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
-import org.elasticsearch.discovery.zen.elect.ElectMasterService;
-import org.elasticsearch.discovery.zen.ping.multicast.MulticastZenPing;
-import org.elasticsearch.discovery.zen.ping.unicast.UnicastHostsProvider;
-import org.elasticsearch.discovery.zen.ping.unicast.UnicastZenPing;
-import org.elasticsearch.threadpool.ThreadPool;
-import org.elasticsearch.transport.TransportService;
 
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
-/**
- *
- */
 public class ZenPingService extends AbstractLifecycleComponent<ZenPing> implements ZenPing {
 
-    private volatile ImmutableList<? extends ZenPing> zenPings = ImmutableList.of();
+    private List<ZenPing> zenPings = Collections.emptyList();
 
     @Inject
-    public ZenPingService(Settings settings, ThreadPool threadPool, TransportService transportService, ClusterName clusterName, NetworkService networkService,
+    public ZenPingService(Settings settings, Set<ZenPing> zenPings) {
-                          Version version, ElectMasterService electMasterService, @Nullable Set<UnicastHostsProvider> unicastHostsProviders) {
         super(settings);
-        ImmutableList.Builder<ZenPing> zenPingsBuilder = ImmutableList.builder();
+        this.zenPings = Collections.unmodifiableList(new ArrayList<>(zenPings));
-        if (this.settings.getAsBoolean("discovery.zen.ping.multicast.enabled", false)) {
-            zenPingsBuilder.add(new MulticastZenPing(settings, threadPool, transportService, clusterName, networkService, version));
-        }
-        // always add the unicast hosts, or things get angry!
-        zenPingsBuilder.add(new UnicastZenPing(settings, threadPool, transportService, clusterName, version, electMasterService, unicastHostsProviders));
-
-        this.zenPings = zenPingsBuilder.build();
     }
 
-    public ImmutableList<? extends ZenPing> zenPings() {
+    public List<ZenPing> zenPings() {
         return this.zenPings;
     }
 
@@ -118,7 +99,7 @@ public class ZenPingService extends AbstractLifecycleComponent<ZenPing> implemen
 
     @Override
     public void ping(PingListener listener, TimeValue timeout) {
-        ImmutableList<? extends ZenPing> zenPings = this.zenPings;
+        List<? extends ZenPing> zenPings = this.zenPings;
         CompoundPingListener compoundPingListener = new CompoundPingListener(listener, zenPings);
         for (ZenPing zenPing : zenPings) {
             try {
@@ -138,7 +119,7 @@ public class ZenPingService extends AbstractLifecycleComponent<ZenPing> implemen
 
         private PingCollection responses = new PingCollection();
 
-        private CompoundPingListener(PingListener listener, ImmutableList<? extends ZenPing> zenPings) {
+        private CompoundPingListener(PingListener listener, List<? extends ZenPing> zenPings) {
             this.listener = listener;
             this.counter = new AtomicInteger(zenPings.size());
         }

core/src/main/java/org/elasticsearch/discovery/zen/ping/unicast/UnicastZenPing.java

@@ -29,6 +29,7 @@ import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.component.AbstractLifecycleComponent;
+import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.settings.Settings;
@@ -99,6 +100,7 @@ public class UnicastZenPing extends AbstractLifecycleComponent<ZenPing> implemen
 
     private volatile boolean closed = false;
 
+    @Inject
     public UnicastZenPing(Settings settings, ThreadPool threadPool, TransportService transportService, ClusterName clusterName,
                           Version version, ElectMasterService electMasterService, @Nullable Set<UnicastHostsProvider> unicastHostsProviders) {
         super(settings);

core/src/main/resources/org/elasticsearch/plugins/plugin-install.help

@@ -47,6 +47,7 @@ OFFICIAL PLUGINS
     - cloud-azure
     - cloud-gce
     - delete-by-query
+    - discovery-multicast
     - lang-javascript
     - lang-python
     - mapper-murmur3

core/src/test/java/org/elasticsearch/discovery/DiscoveryWithServiceDisruptionsIT.java

@@ -105,23 +105,10 @@ public class DiscoveryWithServiceDisruptionsIT extends ESIntegTestCase {
     }
 
     private List<String> startCluster(int numberOfNodes, int minimumMasterNode) throws ExecutionException, InterruptedException {
-        configureCluster(numberOfNodes, minimumMasterNode);
+        return startCluster(numberOfNodes, minimumMasterNode, null);
-        List<String> nodes = internalCluster().startNodesAsync(numberOfNodes).get();
-        ensureStableCluster(numberOfNodes);
-
-        // TODO: this is a temporary solution so that nodes will not base their reaction to a partition based on previous successful results
-        for (ZenPingService pingService : internalCluster().getInstances(ZenPingService.class)) {
-            for (ZenPing zenPing : pingService.zenPings()) {
-                if (zenPing instanceof UnicastZenPing) {
-                    ((UnicastZenPing) zenPing).clearTemporalResponses();
-                }
-            }
-        }
-        return nodes;
     }
 
-
+    private List<String> startCluster(int numberOfNodes, int minimumMasterNode, @Nullable int[] unicastHostsOrdinals) throws ExecutionException, InterruptedException {
-    private List<String> startUnicastCluster(int numberOfNodes, @Nullable int[] unicastHostsOrdinals, int minimumMasterNode) throws ExecutionException, InterruptedException {
         configureUnicastCluster(numberOfNodes, unicastHostsOrdinals, minimumMasterNode);
         List<String> nodes = internalCluster().startNodesAsync(numberOfNodes).get();
         ensureStableCluster(numberOfNodes);
@@ -150,33 +137,6 @@ public class DiscoveryWithServiceDisruptionsIT extends ESIntegTestCase {
             .put("plugin.types", MockTransportService.TestPlugin.class.getName())
             .build();
 
-    private void configureCluster(int numberOfNodes, int minimumMasterNode) throws ExecutionException, InterruptedException {
-        if (randomBoolean() && canUseMuticast()) {
-            configureMulticastCluster(numberOfNodes, minimumMasterNode);
-        } else {
-            configureUnicastCluster(numberOfNodes, null, minimumMasterNode);
-        }
-
-    }
-
-    private void configureMulticastCluster(int numberOfNodes, int minimumMasterNode) throws ExecutionException, InterruptedException {
-        if (minimumMasterNode < 0) {
-            minimumMasterNode = numberOfNodes / 2 + 1;
-        }
-        logger.info("---> configured multicast");
-        // TODO: Rarely use default settings form some of these
-        Settings settings = Settings.builder()
-                .put(DEFAULT_SETTINGS)
-                .put("discovery.zen.ping.multicast.enabled", true)
-                .put(ElectMasterService.DISCOVERY_ZEN_MINIMUM_MASTER_NODES, minimumMasterNode)
-                .put()
-                .build();
-
-        if (discoveryConfig == null) {
-            discoveryConfig = new ClusterDiscoveryConfiguration(numberOfNodes, settings);
-        }
-    }
-
     private void configureUnicastCluster(int numberOfNodes, @Nullable int[] unicastHostsOrdinals, int minimumMasterNode) throws ExecutionException, InterruptedException {
         if (minimumMasterNode < 0) {
             minimumMasterNode = numberOfNodes / 2 + 1;
@@ -564,9 +524,7 @@ public class DiscoveryWithServiceDisruptionsIT extends ESIntegTestCase {
      */
     @Test
     public void testMasterNodeGCs() throws Exception {
-        // TODO: on mac OS multicast threads are shared between nodes and we therefore we can't simulate GC and stop pinging for just one node
+        List<String> nodes = startCluster(3, -1);
-        // find a way to block thread creation in the generic thread pool to avoid this.
-        List<String> nodes = startUnicastCluster(3, null, -1);
 
         String oldMasterNode = internalCluster().getMasterName();
         // a very long GC, but it's OK as we remove the disruption when it has had an effect
@@ -608,10 +566,8 @@ public class DiscoveryWithServiceDisruptionsIT extends ESIntegTestCase {
      */
     @Test
     public void testStaleMasterNotHijackingMajority() throws Exception {
-        // TODO: on mac OS multicast threads are shared between nodes and we therefore we can't simulate GC and stop pinging for just one node
-        // find a way to block thread creation in the generic thread pool to avoid this.
         // 3 node cluster with unicast discovery and minimum_master_nodes set to 2:
-        final List<String> nodes = startUnicastCluster(3, null, 2);
+        final List<String> nodes = startCluster(3, 2);
 
         // Save the current master node as old master node, because that node will get frozen
         final String oldMasterNode = internalCluster().getMasterName();
@@ -778,7 +734,7 @@ public class DiscoveryWithServiceDisruptionsIT extends ESIntegTestCase {
      */
     @Test
     public void unicastSinglePingResponseContainsMaster() throws Exception {
-        List<String> nodes = startUnicastCluster(4, new int[]{0}, -1);
+        List<String> nodes = startCluster(4, -1, new int[] {0});
         // Figure out what is the elected master node
         final String masterNode = internalCluster().getMasterName();
         logger.info("---> legit elected master node=" + masterNode);
@@ -815,7 +771,7 @@ public class DiscoveryWithServiceDisruptionsIT extends ESIntegTestCase {
     @Test
     @TestLogging("discovery.zen:TRACE,cluster.service:TRACE")
     public void isolatedUnicastNodes() throws Exception {
-        List<String> nodes = startUnicastCluster(4, new int[]{0}, -1);
+        List<String> nodes = startCluster(4, -1, new int[]{0});
         // Figure out what is the elected master node
         final String unicastTarget = nodes.get(0);
 
@@ -898,7 +854,7 @@ public class DiscoveryWithServiceDisruptionsIT extends ESIntegTestCase {
 
     @Test
     public void testClusterFormingWithASlowNode() throws Exception {
-        configureCluster(3, 2);
+        configureUnicastCluster(3, null, 2);
 
         SlowClusterStateProcessing disruption = new SlowClusterStateProcessing(getRandom(), 0, 0, 1000, 2000);
 
@@ -959,7 +915,7 @@ public class DiscoveryWithServiceDisruptionsIT extends ESIntegTestCase {
     @Test
     public void testIndexImportedFromDataOnlyNodesIfMasterLostDataFolder() throws Exception {
         // test for https://github.com/elastic/elasticsearch/issues/8823
-        configureCluster(2, 1);
+        configureUnicastCluster(2, null, 1);
         String masterNode = internalCluster().startMasterOnlyNode(Settings.EMPTY);
         internalCluster().startDataOnlyNode(Settings.EMPTY);
 
@@ -982,7 +938,7 @@ public class DiscoveryWithServiceDisruptionsIT extends ESIntegTestCase {
     @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/11665")
     @Test
     public void testIndicesDeleted() throws Exception {
-        configureCluster(3, 2);
+        configureUnicastCluster(3, null, 2);
         Future<List<String>> masterNodes= internalCluster().startMasterOnlyNodesAsync(2);
         Future<String> dataNode = internalCluster().startDataOnlyNodeAsync();
         dataNode.get();

core/src/test/java/org/elasticsearch/test/ESIntegTestCase.java

@@ -1999,24 +1999,4 @@ public abstract class ESIntegTestCase extends ESTestCase {
     @Inherited
     public @interface SuppressNetworkMode {}
 
-    /**
-     * Annotation used to set if working multicast is required to run the test.
-     * By default, tests annotated with @Multicast won't be executed.
-     * Set -Dtests.multicast=true when running test to launch multicast tests
-     */
-    @Retention(RetentionPolicy.RUNTIME)
-    @Inherited
-    @TestGroup(enabled = false, sysProperty = "tests.multicast")
-    public @interface Multicast {
-    }
-
-
-    /**
-     * Returns true if tests can use multicast. Default is false.
-     * To disable an entire test use {@link org.elasticsearch.test.ESIntegTestCase.Multicast} instead
-     */
-    protected boolean canUseMuticast() {
-        return Boolean.parseBoolean(System.getProperty("tests.multicast", "false"));
-    }
-
 }

core/src/test/java/org/elasticsearch/threadpool/SimpleThreadPoolIT.java

@@ -23,7 +23,6 @@ import com.google.common.collect.Sets;
 import org.elasticsearch.action.admin.cluster.node.info.NodeInfo;
 import org.elasticsearch.action.admin.cluster.node.info.NodesInfoResponse;
 import org.elasticsearch.action.index.IndexRequestBuilder;
-import org.elasticsearch.common.network.MulticastChannel;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
@@ -112,8 +111,7 @@ public class SimpleThreadPoolIT extends ESIntegTestCase {
         for (String threadName : threadNames) {
             // ignore some shared threads we know that are created within the same VM, like the shared discovery one
             // or the ones that are occasionally come up from ESSingleNodeTestCase
-            if (threadName.contains("[" + MulticastChannel.SHARED_CHANNEL_NAME + "]")
+            if (threadName.contains("[" + ESSingleNodeTestCase.nodeName() + "]")
-                    || threadName.contains("[" + ESSingleNodeTestCase.nodeName() + "]")
                     || threadName.contains("Keep-Alive-Timer")) {
                 continue;
             }

dev-tools/build_release.py

@@ -420,7 +420,7 @@ def smoke_test_release(release, files, expected_hash, plugins):
 
     background = '-d'
     print('  Starting elasticsearch deamon from [%s]' % os.path.join(tmp_dir, 'elasticsearch-%s' % release))
-    run('%s; %s -Des.node.name=smoke_tester -Des.cluster.name=prepare_release -Des.discovery.zen.ping.multicast.enabled=false -Des.script.inline=on -Des.script.indexed=on %s'
+    run('%s; %s -Des.node.name=smoke_tester -Des.cluster.name=prepare_release -Des.script.inline=on -Des.script.indexed=on %s'
          % (java_exe(), es_run_path, background))
     conn = HTTPConnection('127.0.0.1', 9200, 20);
     wait_for_node_startup()

dev-tools/create_bwc_index.py

@@ -141,7 +141,6 @@ def start_node(version, release_dir, data_dir, repo_dir, tcp_port=DEFAULT_TRANSP
     '-Des.path.logs=logs',
     '-Des.cluster.name=%s' % cluster_name,
     '-Des.network.host=localhost',
-    '-Des.discovery.zen.ping.multicast.enabled=false',
     '-Des.transport.tcp.port=%s' % tcp_port,
     '-Des.http.port=%s' % http_port,
     '-Des.path.repo=%s' % repo_dir

dev-tools/src/main/resources/ant/integration-tests.xml

@@ -168,7 +168,6 @@
           <arg value="-Des.discovery.zen.ping.unicast.hosts=@{es.unicast.hosts}"/>
           <arg value="-Des.path.repo=@{home}/repo"/>
           <arg value="-Des.path.shared_data=@{home}/../"/>
-          <arg value="-Des.discovery.zen.ping.multicast.enabled=false"/>
           <arg value="-Des.script.inline=on"/>
           <arg value="-Des.script.indexed=on"/>
           <arg value="-Des.repositories.url.allowed_urls=http://snapshot.test*"/>

dev-tools/upgrade-tests.py

@@ -106,7 +106,6 @@ def start_node(version, data_dir, node_dir, unicast_host_list, tcp_port, http_po
   return subprocess.Popen([es_run_path,
     '-Des.path.data=%s' % data_dir, '-Des.cluster.name=upgrade_test',  
     '-Des.discovery.zen.ping.unicast.hosts=%s' % unicast_host_list,
-    '-Des.discovery.zen.ping.multicast.enabled=false',
     '-Des.transport.tcp.port=%s' % tcp_port,
     '-Des.http.port=%s' % http_port,
     foreground], stdout=subprocess.PIPE, stderr=subprocess.PIPE)

distribution/src/main/resources/config/elasticsearch.yml

@@ -82,10 +82,6 @@
 #
 # discovery.zen.minimum_master_nodes: 3
 #
-# To use multicast for discovery, enable it:
-#
-# discovery.zen.ping.multicast.enabled: true
-#
 # For more information, see the documentation at:
 # <http://www.elastic.co/guide/en/elasticsearch/reference/current/modules-discovery.html>
 #

docs/plugins/discovery-multicast.asciidoc

@@ -0,0 +1,55 @@
+[[mapper-murmur3]]
+=== Multicast Discovery Plugin
+
+The Multicast Discovery plugin provides the ability to form a cluster using
+TCP/IP multicast messages.
+
+[[discovery-multicast-install]]
+[float]
+==== Installation
+
+This plugin can be installed using the plugin manager:
+
+[source,sh]
+----------------------------------------------------------------
+sudo bin/plugin install discovery-multicast
+----------------------------------------------------------------
+
+The plugin must be installed on every node in the cluster, and each node must
+be restarted after installation.
+
+[[discovery-multicast-remove]]
+[float]
+==== Removal
+
+The plugin can be removed with the following command:
+
+[source,sh]
+----------------------------------------------------------------
+sudo bin/plugin remove discovery-multicast
+----------------------------------------------------------------
+
+The node must be stopped before removing the plugin.
+
+[[discovery-multicast-usage]]
+==== Configuring multicast discovery
+
+Multicast ping discovery of other nodes is done by sending one or more
+multicast requests which existing nodes will receive and
+respond to. It provides the following settings with the
+`discovery.zen.ping.multicast` prefix:
+
+[cols="<,<",options="header",]
+|=======================================================================
+|Setting |Description
+|`group` |The group address to use. Defaults to `224.2.2.4`.
+
+|`port` |The port to use. Defaults to `54328`.
+
+|`ttl` |The ttl of the multicast message. Defaults to `3`.
+
+|`address` |The address to bind to, defaults to `null` which means it
+will bind `network.bind_host`
+
+|`enabled` |Whether multicast ping discovery is enabled. Defaults to `false`.
+|=======================================================================

docs/plugins/discovery.asciidoc

@@ -26,6 +26,10 @@ support for using Azure as a repository for
 
 The Google Compute Engine Cloud plugin uses the GCE API for unicast discovery.
 
+<<discovery-multicast,Multicast>>::
+
+The multicast plugin sends multicast messages to discover other nodes in the cluster.
+
 [float]
 ==== Community contributed discovery plugins
 
@@ -41,5 +45,7 @@ include::cloud-azure.asciidoc[]
 
 include::cloud-gce.asciidoc[]
 
+include::discovery-multicast.asciidoc[]
+
 
 

docs/reference/glossary.asciidoc

@@ -86,9 +86,8 @@
   server for testing purposes, but usually you should have one node per
   server.
   +
-  At startup, a node will use unicast (or multicast, if specified) to
+  At startup, a node will use unicast to discover an existing cluster with
-  discover an existing cluster with the same cluster name and will try
+  the same cluster name and will try to join that cluster.
-  to join that cluster.
 
  [[glossary-primary-shard]] primary shard ::
 

docs/reference/modules/discovery/zen.asciidoc

@@ -2,8 +2,8 @@
 === Zen Discovery
 
 The zen discovery is the built in discovery module for elasticsearch and
-the default. It provides both unicast and multicast discovery as well
+the default. It provides unicast discovery, but can be extended to
-being easily extended to support cloud environments.
+support cloud environments and other forms of discovery.
 
 The zen discovery is integrated with other modules, for example, all
 communication between nodes is done using the
@@ -16,39 +16,13 @@ It is separated into several sub modules, which are explained below:
 ==== Ping
 
 This is the process where a node uses the discovery mechanisms to find
-other nodes. There is support for both multicast and unicast based
+other nodes.
-discovery (these mechanisms can be used in conjunction as well).
-
-[float]
-[[multicast]]
-===== Multicast
-
-Multicast ping discovery of other nodes is done by sending one or more
-multicast requests which existing nodes will receive and
-respond to. It provides the following settings with the
-`discovery.zen.ping.multicast` prefix:
-
-[cols="<,<",options="header",]
-|=======================================================================
-|Setting |Description
-|`group` |The group address to use. Defaults to `224.2.2.4`.
-
-|`port` |The port to use. Defaults to `54328`.
-
-|`ttl` |The ttl of the multicast message. Defaults to `3`.
-
-|`address` |The address to bind to, defaults to `null` which means it
-will bind `network.bind_host`
-
-|`enabled` |Whether multicast ping discovery is enabled. Defaults to `false`.
-|=======================================================================
 
 [float]
 [[unicast]]
 ===== Unicast
 
-The unicast discovery allows for discovery when multicast is
+The unicast discovery requires a list of hosts to use that will act
-not enabled. It basically requires a list of hosts to use that will act
 as gossip routers. It provides the following settings with the
 `discovery.zen.ping.unicast` prefix:
 
@@ -128,45 +102,6 @@ The following settings control the fault detection process using the
 considered failed. Defaults to `3`.
 |=======================================================================
 
-[float]
-==== External Multicast
-
-The multicast discovery also supports external multicast requests to
-discover nodes. The external client can send a request to the multicast
-IP/group and port, in the form of:
-
-[source,js]
---------------------------------------------------
-{
-    "request" : {
-        "cluster_name": "test_cluster"
-    }
-}
---------------------------------------------------
-
-And the response will be similar to node info response (with node level
-information only, including transport/http addresses, and node
-attributes):
-
-[source,js]
---------------------------------------------------
-{
-    "response" : {
-        "cluster_name" : "test_cluster",
-        "transport_address" : "...",
-        "http_address" : "...",
-        "attributes" : {
-            "..."
-        }
-    }
-}
---------------------------------------------------
-
-Note, it can still be enabled, with disabled internal multicast
-discovery, but still have external discovery working by keeping
-`discovery.zen.ping.multicast.enabled` set to `true` (the default), but,
-setting `discovery.zen.ping.multicast.ping.enabled` to `false`.
-
 [float]
 ==== Cluster state updates
 

docs/reference/modules/tribe.asciidoc

@@ -25,7 +25,7 @@ tribe:
 
 The example above configures connections to two clusters, name `t1` and `t2`
 respectively.  The tribe node will create a <<modules-node,node client>> to
-connect each cluster using <<multicast,multicast discovery>> by default. Any
+connect each cluster using <<unicast,unicast discovery>> by default. Any
 other settings for the connection can be configured under `tribe.{name}`, just
 like the `cluster.name` in the example.
 

plugins/cloud-gce/src/main/java/org/elasticsearch/discovery/gce/GceDiscovery.java

@@ -45,8 +45,5 @@ public class GceDiscovery extends ZenDiscovery {
                         ElectMasterService electMasterService) {
         super(settings, clusterName, threadPool, transportService, clusterService, nodeSettingsService,
                 pingService, electMasterService, discoverySettings);
-
-        // TODO Add again force disable multicast
-        // See related issue in AWS plugin https://github.com/elastic/elasticsearch-cloud-aws/issues/179
     }
 }

plugins/discovery-multicast/LICENSE.txt

@@ -0,0 +1,202 @@
+
+                                 Apache License
+                           Version 2.0, January 2004
+                        http://www.apache.org/licenses/
+
+   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+   1. Definitions.
+
+      "License" shall mean the terms and conditions for use, reproduction,
+      and distribution as defined by Sections 1 through 9 of this document.
+
+      "Licensor" shall mean the copyright owner or entity authorized by
+      the copyright owner that is granting the License.
+
+      "Legal Entity" shall mean the union of the acting entity and all
+      other entities that control, are controlled by, or are under common
+      control with that entity. For the purposes of this definition,
+      "control" means (i) the power, direct or indirect, to cause the
+      direction or management of such entity, whether by contract or
+      otherwise, or (ii) ownership of fifty percent (50%) or more of the
+      outstanding shares, or (iii) beneficial ownership of such entity.
+
+      "You" (or "Your") shall mean an individual or Legal Entity
+      exercising permissions granted by this License.
+
+      "Source" form shall mean the preferred form for making modifications,
+      including but not limited to software source code, documentation
+      source, and configuration files.
+
+      "Object" form shall mean any form resulting from mechanical
+      transformation or translation of a Source form, including but
+      not limited to compiled object code, generated documentation,
+      and conversions to other media types.
+
+      "Work" shall mean the work of authorship, whether in Source or
+      Object form, made available under the License, as indicated by a
+      copyright notice that is included in or attached to the work
+      (an example is provided in the Appendix below).
+
+      "Derivative Works" shall mean any work, whether in Source or Object
+      form, that is based on (or derived from) the Work and for which the
+      editorial revisions, annotations, elaborations, or other modifications
+      represent, as a whole, an original work of authorship. For the purposes
+      of this License, Derivative Works shall not include works that remain
+      separable from, or merely link (or bind by name) to the interfaces of,
+      the Work and Derivative Works thereof.
+
+      "Contribution" shall mean any work of authorship, including
+      the original version of the Work and any modifications or additions
+      to that Work or Derivative Works thereof, that is intentionally
+      submitted to Licensor for inclusion in the Work by the copyright owner
+      or by an individual or Legal Entity authorized to submit on behalf of
+      the copyright owner. For the purposes of this definition, "submitted"
+      means any form of electronic, verbal, or written communication sent
+      to the Licensor or its representatives, including but not limited to
+      communication on electronic mailing lists, source code control systems,
+      and issue tracking systems that are managed by, or on behalf of, the
+      Licensor for the purpose of discussing and improving the Work, but
+      excluding communication that is conspicuously marked or otherwise
+      designated in writing by the copyright owner as "Not a Contribution."
+
+      "Contributor" shall mean Licensor and any individual or Legal Entity
+      on behalf of whom a Contribution has been received by Licensor and
+      subsequently incorporated within the Work.
+
+   2. Grant of Copyright License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      copyright license to reproduce, prepare Derivative Works of,
+      publicly display, publicly perform, sublicense, and distribute the
+      Work and such Derivative Works in Source or Object form.
+
+   3. Grant of Patent License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      (except as stated in this section) patent license to make, have made,
+      use, offer to sell, sell, import, and otherwise transfer the Work,
+      where such license applies only to those patent claims licensable
+      by such Contributor that are necessarily infringed by their
+      Contribution(s) alone or by combination of their Contribution(s)
+      with the Work to which such Contribution(s) was submitted. If You
+      institute patent litigation against any entity (including a
+      cross-claim or counterclaim in a lawsuit) alleging that the Work
+      or a Contribution incorporated within the Work constitutes direct
+      or contributory patent infringement, then any patent licenses
+      granted to You under this License for that Work shall terminate
+      as of the date such litigation is filed.
+
+   4. Redistribution. You may reproduce and distribute copies of the
+      Work or Derivative Works thereof in any medium, with or without
+      modifications, and in Source or Object form, provided that You
+      meet the following conditions:
+
+      (a) You must give any other recipients of the Work or
+          Derivative Works a copy of this License; and
+
+      (b) You must cause any modified files to carry prominent notices
+          stating that You changed the files; and
+
+      (c) You must retain, in the Source form of any Derivative Works
+          that You distribute, all copyright, patent, trademark, and
+          attribution notices from the Source form of the Work,
+          excluding those notices that do not pertain to any part of
+          the Derivative Works; and
+
+      (d) If the Work includes a "NOTICE" text file as part of its
+          distribution, then any Derivative Works that You distribute must
+          include a readable copy of the attribution notices contained
+          within such NOTICE file, excluding those notices that do not
+          pertain to any part of the Derivative Works, in at least one
+          of the following places: within a NOTICE text file distributed
+          as part of the Derivative Works; within the Source form or
+          documentation, if provided along with the Derivative Works; or,
+          within a display generated by the Derivative Works, if and
+          wherever such third-party notices normally appear. The contents
+          of the NOTICE file are for informational purposes only and
+          do not modify the License. You may add Your own attribution
+          notices within Derivative Works that You distribute, alongside
+          or as an addendum to the NOTICE text from the Work, provided
+          that such additional attribution notices cannot be construed
+          as modifying the License.
+
+      You may add Your own copyright statement to Your modifications and
+      may provide additional or different license terms and conditions
+      for use, reproduction, or distribution of Your modifications, or
+      for any such Derivative Works as a whole, provided Your use,
+      reproduction, and distribution of the Work otherwise complies with
+      the conditions stated in this License.
+
+   5. Submission of Contributions. Unless You explicitly state otherwise,
+      any Contribution intentionally submitted for inclusion in the Work
+      by You to the Licensor shall be under the terms and conditions of
+      this License, without any additional terms or conditions.
+      Notwithstanding the above, nothing herein shall supersede or modify
+      the terms of any separate license agreement you may have executed
+      with Licensor regarding such Contributions.
+
+   6. Trademarks. This License does not grant permission to use the trade
+      names, trademarks, service marks, or product names of the Licensor,
+      except as required for reasonable and customary use in describing the
+      origin of the Work and reproducing the content of the NOTICE file.
+
+   7. Disclaimer of Warranty. Unless required by applicable law or
+      agreed to in writing, Licensor provides the Work (and each
+      Contributor provides its Contributions) on an "AS IS" BASIS,
+      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+      implied, including, without limitation, any warranties or conditions
+      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+      PARTICULAR PURPOSE. You are solely responsible for determining the
+      appropriateness of using or redistributing the Work and assume any
+      risks associated with Your exercise of permissions under this License.
+
+   8. Limitation of Liability. In no event and under no legal theory,
+      whether in tort (including negligence), contract, or otherwise,
+      unless required by applicable law (such as deliberate and grossly
+      negligent acts) or agreed to in writing, shall any Contributor be
+      liable to You for damages, including any direct, indirect, special,
+      incidental, or consequential damages of any character arising as a
+      result of this License or out of the use or inability to use the
+      Work (including but not limited to damages for loss of goodwill,
+      work stoppage, computer failure or malfunction, or any and all
+      other commercial damages or losses), even if such Contributor
+      has been advised of the possibility of such damages.
+
+   9. Accepting Warranty or Additional Liability. While redistributing
+      the Work or Derivative Works thereof, You may choose to offer,
+      and charge a fee for, acceptance of support, warranty, indemnity,
+      or other liability obligations and/or rights consistent with this
+      License. However, in accepting such obligations, You may act only
+      on Your own behalf and on Your sole responsibility, not on behalf
+      of any other Contributor, and only if You agree to indemnify,
+      defend, and hold each Contributor harmless for any liability
+      incurred by, or claims asserted against, such Contributor by reason
+      of your accepting any such warranty or additional liability.
+
+   END OF TERMS AND CONDITIONS
+
+   APPENDIX: How to apply the Apache License to your work.
+
+      To apply the Apache License to your work, attach the following
+      boilerplate notice, with the fields enclosed by brackets "[]"
+      replaced with your own identifying information. (Don't include
+      the brackets!)  The text should be enclosed in the appropriate
+      comment syntax for the file format. We also recommend that a
+      file or class name and description of purpose be included on the
+      same "printed page" as the copyright notice for easier
+      identification within third-party archives.
+
+   Copyright [yyyy] [name of copyright owner]
+
+   Licensed under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.

plugins/discovery-multicast/NOTICE.txt

@@ -0,0 +1,8 @@
+Elasticsearch
+Copyright 2009-2015 Elasticsearch
+
+This product includes software developed by The Apache Software
+Foundation (http://www.apache.org/).
+
+The LICENSE and NOTICE files for all dependencies may be found in the licenses/
+directory.

plugins/discovery-multicast/pom.xml

@@ -0,0 +1,33 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>org.elasticsearch.plugin</groupId>
+        <artifactId>plugins</artifactId>
+        <version>2.1.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>discovery-multicast</artifactId>
+    <name>Plugin: Discovery: Multicast</name>
+    <description>The Multicast Discovery plugin allows discovery other nodes using multicast requests</description>
+
+    <properties>
+        <elasticsearch.plugin.classname>org.elasticsearch.plugin.discovery.multicast.MulticastDiscoveryPlugin</elasticsearch.plugin.classname>
+        <tests.jvms>1</tests.jvms>
+        <tests.rest.suite>discovery_multicast</tests.rest.suite>
+        <tests.rest.load_packaged>false</tests.rest.load_packaged>
+    </properties>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-assembly-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>

plugins/discovery-multicast/rest-api-spec/test/discovery_multicast/10_basic.yaml

@@ -0,0 +1,14 @@
+# Integration tests for multicast discovery
+#
+"Multicast discovery loaded":
+    - do:
+        cluster.state: {}
+
+    # Get master node id
+    - set: { master_node: master }
+
+    - do:
+        nodes.info: {}
+
+    - match:  { nodes.$master.plugins.0.name: discovery-multicast  }
+    - match:  { nodes.$master.plugins.0.jvm: true  }

plugins/discovery-multicast/src/main/java/org/elasticsearch/plugin/discovery/multicast/MulticastChannel.java

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.elasticsearch.common.network;
+package org.elasticsearch.plugin.discovery.multicast;
 
 import com.google.common.collect.Maps;
 

plugins/discovery-multicast/src/main/java/org/elasticsearch/plugin/discovery/multicast/MulticastDiscoveryPlugin.java

@@ -0,0 +1,53 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.plugin.discovery.multicast;
+
+import org.elasticsearch.common.inject.Module;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.discovery.DiscoveryModule;
+import org.elasticsearch.plugin.discovery.multicast.MulticastZenPing;
+import org.elasticsearch.plugins.Plugin;
+
+import java.util.Collection;
+
+public class MulticastDiscoveryPlugin extends Plugin {
+
+    private final Settings settings;
+
+    public MulticastDiscoveryPlugin(Settings settings) {
+        this.settings = settings;
+    }
+
+    @Override
+    public String name() {
+        return "discovery-multicast";
+    }
+
+    @Override
+    public String description() {
+        return "Multicast Discovery Plugin";
+    }
+    
+    public void onModule(DiscoveryModule module) {
+        if (settings.getAsBoolean("discovery.zen.ping.multicast.enabled", false)) {
+            module.addZenPing(MulticastZenPing.class);
+        }
+    }
+}

plugins/discovery-multicast/src/main/java/org/elasticsearch/plugin/discovery/multicast/MulticastZenPing.java

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.elasticsearch.discovery.zen.ping.multicast;
+package org.elasticsearch.plugin.discovery.multicast;
 
 import org.apache.lucene.util.Constants;
 import org.elasticsearch.ExceptionsHelper;
@@ -28,10 +28,10 @@ import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.component.AbstractLifecycleComponent;
+import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
-import org.elasticsearch.common.network.MulticastChannel;
 import org.elasticsearch.common.network.NetworkService;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
@@ -92,6 +92,7 @@ public class MulticastZenPing extends AbstractLifecycleComponent<ZenPing> implem
         this(EMPTY_SETTINGS, threadPool, transportService, clusterName, new NetworkService(EMPTY_SETTINGS), version);
     }
 
+    @Inject
     public MulticastZenPing(Settings settings, ThreadPool threadPool, TransportService transportService, ClusterName clusterName, NetworkService networkService, Version version) {
         super(settings);
         this.threadPool = threadPool;

plugins/discovery-multicast/src/test/java/org/elasticsearch/plugin/discovery/multicast/MulticastDiscoveryRestIT.java

@@ -0,0 +1,41 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.plugin.discovery.multicast;
+
+import com.carrotsearch.randomizedtesting.annotations.Name;
+import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
+import org.elasticsearch.test.rest.ESRestTestCase;
+import org.elasticsearch.test.rest.RestTestCandidate;
+import org.elasticsearch.test.rest.parser.RestTestParseException;
+
+import java.io.IOException;
+
+public class MulticastDiscoveryRestIT extends ESRestTestCase {
+
+    public MulticastDiscoveryRestIT(@Name("yaml") RestTestCandidate testCandidate) {
+        super(testCandidate);
+    }
+
+    @ParametersFactory
+    public static Iterable<Object[]> parameters() throws IOException, RestTestParseException {
+        return ESRestTestCase.createParameters(0, 1);
+    }
+}
+

plugins/discovery-multicast/src/test/java/org/elasticsearch/plugin/discovery/multicast/MulticastZenPingTests.java

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.elasticsearch.discovery.zen.ping.multicast;
+package org.elasticsearch.plugin.discovery.multicast;
 
 import org.elasticsearch.Version;
 import org.elasticsearch.cluster.ClusterName;
@@ -33,21 +33,19 @@ import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.discovery.zen.ping.PingContextProvider;
 import org.elasticsearch.discovery.zen.ping.ZenPing;
 import org.elasticsearch.node.service.NodeService;
-import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.transport.local.LocalTransport;
+import org.hamcrest.Matchers;
+import org.junit.Assert;
 import org.junit.Test;
 
 import java.net.DatagramPacket;
 import java.net.InetAddress;
 import java.net.MulticastSocket;
 
-import static org.hamcrest.Matchers.equalTo;
+public class MulticastZenPingTests extends ESTestCase {
-
-@ESIntegTestCase.Multicast
-public class MulticastZenPingIT extends ESTestCase {
 
     private Settings buildRandomMulticast(Settings settings) {
         Settings.Builder builder = Settings.builder().put(settings);
@@ -64,6 +62,7 @@ public class MulticastZenPingIT extends ESTestCase {
     public void testSimplePings() throws InterruptedException {
         Settings settings = Settings.EMPTY;
         settings = buildRandomMulticast(settings);
+        Thread.sleep(30000);
 
         ThreadPool threadPool = new ThreadPool("testSimplePings");
         final ClusterName clusterName = new ClusterName("test");
@@ -114,15 +113,15 @@ public class MulticastZenPingIT extends ESTestCase {
         try {
             logger.info("ping from A");
             ZenPing.PingResponse[] pingResponses = zenPingA.pingAndWait(TimeValue.timeValueSeconds(1));
-            assertThat(pingResponses.length, equalTo(1));
+            Assert.assertThat(pingResponses.length, Matchers.equalTo(1));
-            assertThat(pingResponses[0].node().id(), equalTo("B"));
+            Assert.assertThat(pingResponses[0].node().id(), Matchers.equalTo("B"));
-            assertTrue(pingResponses[0].hasJoinedOnce());
+            Assert.assertTrue(pingResponses[0].hasJoinedOnce());
 
             logger.info("ping from B");
             pingResponses = zenPingB.pingAndWait(TimeValue.timeValueSeconds(1));
-            assertThat(pingResponses.length, equalTo(1));
+            Assert.assertThat(pingResponses.length, Matchers.equalTo(1));
-            assertThat(pingResponses[0].node().id(), equalTo("A"));
+            Assert.assertThat(pingResponses[0].node().id(), Matchers.equalTo("A"));
-            assertFalse(pingResponses[0].hasJoinedOnce());
+            Assert.assertFalse(pingResponses[0].hasJoinedOnce());
 
         } finally {
             zenPingA.close();

plugins/pom.xml

@@ -434,6 +434,7 @@
         <module>cloud-azure</module>
         <module>cloud-aws</module>
         <module>delete-by-query</module>
+        <module>discovery-multicast</module>
         <module>lang-python</module>
         <module>lang-javascript</module>
         <module>mapper-murmur3</module>

qa/smoke-test-plugins/pom.xml

@@ -312,6 +312,14 @@
                    <overWrite>true</overWrite>
                  </artifactItem>
 
+                 <artifactItem>
+                   <groupId>org.elasticsearch.plugin</groupId>
+                   <artifactId>discovery-multicast</artifactId>
+                   <version>${elasticsearch.version}</version>
+                   <type>zip</type>
+                   <overWrite>true</overWrite>
+                 </artifactItem>
+
                  <artifactItem>
                    <groupId>org.elasticsearch.plugin</groupId>
                    <artifactId>lang-python</artifactId>