Introduce node handshake

This commit introduces a handshake when initiating a light
connection. During this handshake, node information, cluster name, and
version are received from the target node of the connection. This
information can be used to immediately validate that the target node is
a member of the same cluster, and used to set the version on the
stream. This will allow us to extend APIs that are used during initial
cluster recovery without a major version change.

Relates elastic#15971

core/src/main/java/org/elasticsearch/client/transport/TransportClientNodesService.java

@@ -360,9 +360,10 @@ protected void doSample() {
                     try {
                         // its a listed node, light connect to it...
                         logger.trace("connecting to listed node (light) [{}]", listedNode);
-                        transportService.connectToNodeLight(listedNode);
+                        transportService.connectToNodeLight(listedNode, pingTimeout, !ignoreClusterName);
                     } catch (Throwable e) {
                         logger.debug("failed to connect to node [{}], removed from nodes list", e, listedNode);
+                        newFilteredNodes.add(listedNode);
                         continue;
                     }
                 }
@@ -434,7 +435,7 @@ public void run() {
                                     } else {
                                         // its a listed node, light connect to it...
                                         logger.trace("connecting to listed node (light) [{}]", listedNode);
-                                        transportService.connectToNodeLight(listedNode);
+                                        transportService.connectToNodeLight(listedNode, pingTimeout, !ignoreClusterName);
                                     }
                                 } catch (Exception e) {
                                     logger.debug("failed to connect to node [{}], ignoring...", e, listedNode);

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

@@ -402,7 +402,7 @@ public void run() {
                             // connect to the node, see if we manage to do it, if not, bail
                             if (!nodeFoundByAddress) {
                                 logger.trace("[{}] connecting (light) to {}", sendPingsHandler.id(), finalNodeToSend);
-                                transportService.connectToNodeLight(finalNodeToSend);
+                                transportService.connectToNodeLight(finalNodeToSend, timeout.getMillis());
                             } else {
                                 logger.trace("[{}] connecting to {}", sendPingsHandler.id(), finalNodeToSend);
                                 transportService.connectToNode(finalNodeToSend);
@@ -473,12 +473,6 @@ public void handleResponse(UnicastPingResponse response) {
                             // that's us, ignore
                             continue;
                         }
-                        if (!pingResponse.clusterName().equals(clusterName)) {
-                            // not part of the cluster
-                            logger.debug("[{}] filtering out response from {}, not same cluster_name [{}]", id, pingResponse.node(),
-                                    pingResponse.clusterName().value());
-                            continue;
-                        }
                         SendPingsHandler sendPingsHandler = receivedResponses.get(response.id);
                         if (sendPingsHandler == null) {
                             if (!closed) {

core/src/main/java/org/elasticsearch/transport/TransportService.java

@@ -19,12 +19,16 @@
 
 package org.elasticsearch.transport;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.action.admin.cluster.node.liveness.TransportLivenessAction;
+import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.collect.MapBuilder;
 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.logging.ESLogger;
 import org.elasticsearch.common.logging.Loggers;
 import org.elasticsearch.common.metrics.MeanMetric;
@@ -50,6 +54,7 @@
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ScheduledFuture;
@@ -67,10 +72,12 @@
 public class TransportService extends AbstractLifecycleComponent<TransportService> {
 
     public static final String DIRECT_RESPONSE_PROFILE = ".direct";
+    private static final String HANDSHAKE_ACTION_NAME = "internal:transport/handshake";
 
     private final CountDownLatch blockIncomingRequestsLatch = new CountDownLatch(1);
     protected final Transport transport;
     protected final ThreadPool threadPool;
+    private final ClusterName clusterName;
     protected final TaskManager taskManager;
 
     volatile Map<String, RequestHandlerRegistry> requestHandlers = Collections.emptyMap();
@@ -110,15 +117,16 @@ protected boolean removeEldestEntry(Map.Entry eldest) {
     /** if set will call requests sent to this id to shortcut and executed locally */
     volatile DiscoveryNode localNode = null;
 
-    public TransportService(Transport transport, ThreadPool threadPool) {
-        this(EMPTY_SETTINGS, transport, threadPool);
+    public TransportService(Transport transport, ThreadPool threadPool, ClusterName clusterName) {
+        this(EMPTY_SETTINGS, transport, threadPool, clusterName);
     }
 
     @Inject
-    public TransportService(Settings settings, Transport transport, ThreadPool threadPool) {
+    public TransportService(Settings settings, Transport transport, ThreadPool threadPool, ClusterName clusterName) {
         super(settings);
         this.transport = transport;
         this.threadPool = threadPool;
+        this.clusterName = clusterName;
         setTracerLogInclude(TRACE_LOG_INCLUDE_SETTING.get(settings));
         setTracerLogExclude(TRACE_LOG_EXCLUDE_SETTING.get(settings));
         tracerLog = Loggers.getLogger(logger, ".tracer");
@@ -178,6 +186,12 @@ protected void doStart() {
                 logger.info("profile [{}]: {}", entry.getKey(), entry.getValue());
             }
         }
+        registerRequestHandler(
+            HANDSHAKE_ACTION_NAME,
+            () -> HandshakeRequest.INSTANCE,
+            ThreadPool.Names.SAME,
+            (request, channel) -> channel.sendResponse(
+                    new HandshakeResponse(localNode, clusterName, localNode != null ? localNode.getVersion() : Version.CURRENT)));
     }
 
     @Override
@@ -263,11 +277,120 @@ public void connectToNode(DiscoveryNode node) throws ConnectTransportException {
         transport.connectToNode(node);
     }
 
-    public void connectToNodeLight(DiscoveryNode node) throws ConnectTransportException {
+    /**
+     * Lightly connect to the specified node, and handshake cluster
+     * name and version
+     *
+     * @param node             the node to connect to
+     * @param handshakeTimeout handshake timeout
+     * @return the connected node with version set
+     * @throws ConnectTransportException if the connection or the
+     *                                   handshake failed
+     */
+    public DiscoveryNode connectToNodeLight(final DiscoveryNode node, final long handshakeTimeout) throws ConnectTransportException {
+        return connectToNodeLight(node, handshakeTimeout, true);
+    }
+
+    /**
+     * Lightly connect to the specified node, returning updated node
+     * information. The handshake will fail if the cluster name on the
+     * target node mismatches the local cluster name and
+     * {@code checkClusterName} is {@code true}.
+     *
+     * @param node             the node to connect to
+     * @param handshakeTimeout handshake timeout
+     * @param checkClusterName whether or not to ignore cluster name
+     *                         mismatches
+     * @return the connected node
+     * @throws ConnectTransportException if the connection or the
+     *                                   handshake failed
+     */
+    public DiscoveryNode connectToNodeLight(final DiscoveryNode node, final long handshakeTimeout, final boolean checkClusterName) {
         if (node.equals(localNode)) {
-            return;
+            return localNode;
         }
         transport.connectToNodeLight(node);
+        try {
+            return handshake(node, handshakeTimeout, checkClusterName);
+        } catch (ConnectTransportException e) {
+            transport.disconnectFromNode(node);
+            throw e;
+        }
+    }
+
+    private DiscoveryNode handshake(
+            final DiscoveryNode node,
+            final long handshakeTimeout,
+            final boolean checkClusterName) throws ConnectTransportException {
+        final HandshakeResponse response;
+        try {
+            response = this.submitRequest(
+                node,
+                HANDSHAKE_ACTION_NAME,
+                HandshakeRequest.INSTANCE,
+                TransportRequestOptions.builder().withTimeout(handshakeTimeout).build(),
+                new FutureTransportResponseHandler<HandshakeResponse>() {
+                    @Override
+                    public HandshakeResponse newInstance() {
+                        return new HandshakeResponse();
+                    }
+                }).txGet();
+        } catch (Exception e) {
+            throw new ConnectTransportException(node, "handshake failed", e);
+        }
+
+        if (checkClusterName && !Objects.equals(clusterName, response.clusterName)) {
+            throw new ConnectTransportException(node, "handshake failed, mismatched cluster name [" + response.clusterName + "]");
+        } else if (!isVersionCompatible(response.version)) {
+            throw new ConnectTransportException(node, "handshake failed, incompatible version [" + response.version + "]");
+        }
+
+        return response.discoveryNode;
+    }
+
+    private boolean isVersionCompatible(Version version) {
+        return version.minimumCompatibilityVersion().equals(
+                localNode != null ? localNode.getVersion().minimumCompatibilityVersion() : Version.CURRENT.minimumCompatibilityVersion());
+    }
+
+    public static class HandshakeRequest extends TransportRequest {
+
+        public static final HandshakeRequest INSTANCE = new HandshakeRequest();
+
+        private HandshakeRequest() {
+        }
+
+    }
+
+    public static class HandshakeResponse extends TransportResponse {
+        private DiscoveryNode discoveryNode;
+        private ClusterName clusterName;
+        private Version version;
+
+        public HandshakeResponse() {
+        }
+
+        public HandshakeResponse(DiscoveryNode discoveryNode, ClusterName clusterName, Version version) {
+            this.discoveryNode = discoveryNode;
+            this.version = version;
+            this.clusterName = clusterName;
+        }
+
+        @Override
+        public void readFrom(StreamInput in) throws IOException {
+            super.readFrom(in);
+            discoveryNode = in.readOptionalWriteable(DiscoveryNode::new);
+            clusterName = ClusterName.readClusterName(in);
+            version = Version.readVersion(in);
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            super.writeTo(out);
+            out.writeOptionalWriteable(discoveryNode);
+            clusterName.writeTo(out);
+            Version.writeVersion(version, out);
+        }
     }
 
     public void disconnectFromNode(DiscoveryNode node) {

core/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TaskManagerTestCase.java

@@ -184,9 +184,11 @@ protected boolean accumulateExceptions() {
 
     public static class TestNode implements Releasable {
         public TestNode(String name, ThreadPool threadPool, Settings settings) {
+            clusterService = createClusterService(threadPool);
+            ClusterName clusterName = clusterService.state().getClusterName();
             transportService = new TransportService(settings,
                     new LocalTransport(settings, threadPool, Version.CURRENT, new NamedWriteableRegistry(),
-                        new NoneCircuitBreakerService()), threadPool) {
+                        new NoneCircuitBreakerService()), threadPool, clusterName) {
                 @Override
                 protected TaskManager createTaskManager() {
                     if (MockTaskManager.USE_MOCK_TASK_MANAGER_SETTING.get(settings)) {
@@ -197,7 +199,6 @@ protected TaskManager createTaskManager() {
                 }
             };
             transportService.start();
-            clusterService = createClusterService(threadPool);
             clusterService.add(transportService.getTaskManager());
             discoveryNode = new DiscoveryNode(name, transportService.boundAddress().publishAddress(),
                     emptyMap(), emptySet(), Version.CURRENT);

core/src/test/java/org/elasticsearch/action/bulk/TransportBulkActionTookTests.java

@@ -86,7 +86,7 @@ public void tearDown() throws Exception {
 
     private TransportBulkAction createAction(boolean controlled, AtomicLong expected) {
         CapturingTransport capturingTransport = new CapturingTransport();
-        TransportService transportService = new TransportService(capturingTransport, threadPool);
+        TransportService transportService = new TransportService(capturingTransport, threadPool, clusterService.state().getClusterName());
         transportService.start();
         transportService.acceptIncomingRequests();
         IndexNameExpressionResolver resolver = new Resolver(Settings.EMPTY);

core/src/test/java/org/elasticsearch/action/support/broadcast/node/TransportBroadcastByNodeActionTests.java

@@ -190,7 +190,7 @@ public void setUp() throws Exception {
         super.setUp();
         transport = new CapturingTransport();
         clusterService = createClusterService(THREAD_POOL);
-        final TransportService transportService = new TransportService(transport, THREAD_POOL);
+        final TransportService transportService = new TransportService(transport, THREAD_POOL, clusterService.state().getClusterName());
         transportService.start();
         transportService.acceptIncomingRequests();
         setClusterState(clusterService, TEST_INDEX);

core/src/test/java/org/elasticsearch/action/support/master/TransportMasterNodeActionTests.java

@@ -85,7 +85,7 @@ public void setUp() throws Exception {
         super.setUp();
         transport = new CapturingTransport();
         clusterService = createClusterService(threadPool);
-        transportService = new TransportService(transport, threadPool);
+        transportService = new TransportService(transport, threadPool, clusterService.state().getClusterName());
         transportService.start();
         transportService.acceptIncomingRequests();
         localNode = new DiscoveryNode("local_node", DummyTransportAddress.INSTANCE, Collections.emptyMap(),

core/src/test/java/org/elasticsearch/action/support/nodes/TransportNodesActionTests.java

@@ -119,7 +119,7 @@ public void setUp() throws Exception {
         super.setUp();
         transport = new CapturingTransport();
         clusterService = createClusterService(THREAD_POOL);
-        final TransportService transportService = new TransportService(transport, THREAD_POOL);
+        final TransportService transportService = new TransportService(transport, THREAD_POOL, clusterService.state().getClusterName());
         transportService.start();
         transportService.acceptIncomingRequests();
         int numNodes = randomIntBetween(3, 10);

core/src/test/java/org/elasticsearch/action/support/replication/BroadcastReplicationTests.java

@@ -91,7 +91,7 @@ public void setUp() throws Exception {
         super.setUp();
         LocalTransport transport = new LocalTransport(Settings.EMPTY, threadPool, Version.CURRENT, new NamedWriteableRegistry(), circuitBreakerService);
         clusterService = createClusterService(threadPool);
-        transportService = new TransportService(transport, threadPool);
+        transportService = new TransportService(transport, threadPool, clusterService.state().getClusterName());
         transportService.start();
         transportService.acceptIncomingRequests();
         broadcastReplicationAction = new TestBroadcastReplicationAction(Settings.EMPTY, threadPool, clusterService, transportService, new ActionFilters(new HashSet<ActionFilter>()), new IndexNameExpressionResolver(Settings.EMPTY), null);

core/src/test/java/org/elasticsearch/action/support/replication/TransportReplicationActionTests.java

@@ -120,7 +120,7 @@ public void setUp() throws Exception {
         super.setUp();
         transport = new CapturingTransport();
         clusterService = createClusterService(threadPool);
-        transportService = new TransportService(transport, threadPool);
+        transportService = new TransportService(transport, threadPool, clusterService.state().getClusterName());
         transportService.start();
         transportService.acceptIncomingRequests();
         action = new Action(Settings.EMPTY, "testAction", transportService, clusterService, threadPool);

core/src/test/java/org/elasticsearch/action/support/single/instance/TransportInstanceSingleOperationActionTests.java

@@ -141,7 +141,7 @@ public void setUp() throws Exception {
         super.setUp();
         transport = new CapturingTransport();
         clusterService = createClusterService(THREAD_POOL);
-        transportService = new TransportService(transport, THREAD_POOL);
+        transportService = new TransportService(transport, THREAD_POOL, clusterService.state().getClusterName());
         transportService.start();
         transportService.acceptIncomingRequests();
         action = new TestTransportInstanceSingleOperationAction(

core/src/test/java/org/elasticsearch/client/transport/FailAndRetryMockTransport.java

@@ -20,7 +20,9 @@
 package org.elasticsearch.client.transport;
 
 import com.carrotsearch.randomizedtesting.generators.RandomInts;
+import org.elasticsearch.Version;
 import org.elasticsearch.action.admin.cluster.node.liveness.LivenessResponse;
+import org.elasticsearch.action.admin.cluster.node.liveness.TransportLivenessAction;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.component.Lifecycle;
@@ -34,6 +36,7 @@
 import org.elasticsearch.transport.TransportRequestOptions;
 import org.elasticsearch.transport.TransportResponse;
 import org.elasticsearch.transport.TransportResponseHandler;
+import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.transport.TransportServiceAdapter;
 
 import java.io.IOException;
@@ -47,6 +50,7 @@
 abstract class FailAndRetryMockTransport<Response extends TransportResponse> implements Transport {
 
     private final Random random;
+    private final ClusterName clusterName;
 
     private boolean connectMode = true;
 
@@ -57,8 +61,9 @@
     private final AtomicInteger successes = new AtomicInteger();
     private final Set<DiscoveryNode> triedNodes = new CopyOnWriteArraySet<>();
 
-    FailAndRetryMockTransport(Random random) {
+    FailAndRetryMockTransport(Random random, ClusterName clusterName) {
         this.random = new Random(random.nextLong());
+        this.clusterName = clusterName;
     }
 
     @Override
@@ -69,7 +74,11 @@ public void sendRequest(DiscoveryNode node, long requestId, String action, Trans
         //we make sure that nodes get added to the connected ones when calling addTransportAddress, by returning proper nodes info
         if (connectMode) {
             TransportResponseHandler transportResponseHandler = transportServiceAdapter.onResponseReceived(requestId);
-            transportResponseHandler.handleResponse(new LivenessResponse(ClusterName.DEFAULT, node));
+            if (action.equals(TransportLivenessAction.NAME)) {
+                transportResponseHandler.handleResponse(new LivenessResponse(clusterName, node));
+            } else {
+                transportResponseHandler.handleResponse(new TransportService.HandshakeResponse(node, clusterName, Version.CURRENT));
+            }
             return;
         }
 

core/src/test/java/org/elasticsearch/client/transport/TransportClientHeadersTests.java

@@ -123,8 +123,8 @@ public Settings additionalSettings() {
         CountDownLatch clusterStateLatch = new CountDownLatch(1);
 
         @Inject
-        public InternalTransportService(Settings settings, Transport transport, ThreadPool threadPool) {
-            super(settings, transport, threadPool);
+        public InternalTransportService(Settings settings, Transport transport, ThreadPool threadPool, ClusterName clusterName) {
+            super(settings, transport, threadPool, clusterName);
         }
 
         @Override @SuppressWarnings("unchecked")