Send cluster name and discovery node in handshake (#48916)

Tim-Brooks · web-flow · commit 0645ee88e2a3 · 2019-11-11T18:42:02.000-05:00
This commits sends the cluster name and discovery naode in the transport
level handshake response. This will allow us to stop sending the
transport service level handshake request in the 8.0-8.x release cycle.
It is necessary to start sending this in 7.x so that 8.0 is guaranteed
to be communicating with a version that sends the required information.
diff --git a/server/src/main/java/org/elasticsearch/transport/TcpTransport.java b/server/src/main/java/org/elasticsearch/transport/TcpTransport.java
@@ -27,6 +27,7 @@
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.ThreadedActionListener;
+import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.Booleans;
 import org.elasticsearch.common.Strings;
@@ -151,7 +152,7 @@ public TcpTransport(Settings settings, Version version, ThreadPool threadPool, P
         BigArrays bigArrays = new BigArrays(pageCacheRecycler, circuitBreakerService, CircuitBreaker.IN_FLIGHT_REQUESTS);
 
         this.outboundHandler = new OutboundHandler(nodeName, version, features, threadPool, bigArrays);
-        this.handshaker = new TransportHandshaker(version, threadPool,
+        this.handshaker = new TransportHandshaker(ClusterName.CLUSTER_NAME_SETTING.get(settings), version, threadPool,
             (node, channel, requestId, v) -> outboundHandler.sendRequest(node, channel, requestId,
                 TransportHandshaker.HANDSHAKE_ACTION_NAME, new TransportHandshaker.HandshakeRequest(version),
                 TransportRequestOptions.EMPTY, v, false, true),
@@ -167,6 +168,11 @@ public TcpTransport(Settings settings, Version version, ThreadPool threadPool, P
     protected void doStart() {
     }
 
+    @Override
+    public void setLocalNode(DiscoveryNode localNode) {
+        handshaker.setLocalNode(localNode);
+    }
+
     @Override
     public synchronized void setMessageListener(TransportMessageListener listener) {
         outboundHandler.setMessageListener(listener);
diff --git a/server/src/main/java/org/elasticsearch/transport/Transport.java b/server/src/main/java/org/elasticsearch/transport/Transport.java
@@ -52,6 +52,8 @@ public interface Transport extends LifecycleComponent {
 
     void setMessageListener(TransportMessageListener listener);
 
+    void setLocalNode(DiscoveryNode localNode);
+
     /**
      * The address the transport is bound on.
      */
diff --git a/server/src/main/java/org/elasticsearch/transport/TransportHandshaker.java b/server/src/main/java/org/elasticsearch/transport/TransportHandshaker.java
@@ -20,6 +20,7 @@
 
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
@@ -46,19 +47,26 @@ final class TransportHandshaker {
     private final ConcurrentMap<Long, HandshakeResponseHandler> pendingHandshakes = new ConcurrentHashMap<>();
     private final CounterMetric numHandshakes = new CounterMetric();
 
+    private final ClusterName clusterName;
     private final Version version;
     private final ThreadPool threadPool;
     private final HandshakeRequestSender handshakeRequestSender;
     private final HandshakeResponseSender handshakeResponseSender;
+    private volatile DiscoveryNode localNode;
 
-    TransportHandshaker(Version version, ThreadPool threadPool, HandshakeRequestSender handshakeRequestSender,
+    TransportHandshaker(ClusterName clusterName, Version version, ThreadPool threadPool, HandshakeRequestSender handshakeRequestSender,
                         HandshakeResponseSender handshakeResponseSender) {
+        this.clusterName = clusterName;
         this.version = version;
         this.threadPool = threadPool;
         this.handshakeRequestSender = handshakeRequestSender;
         this.handshakeResponseSender = handshakeResponseSender;
     }
 
+    void setLocalNode(DiscoveryNode localNode) {
+        this.localNode = localNode;
+    }
+
     void sendHandshake(long requestId, DiscoveryNode node, TcpChannel channel, TimeValue timeout, ActionListener<Version> listener) {
         numHandshakes.inc();
         final HandshakeResponseHandler handler = new HandshakeResponseHandler(requestId, version, listener);
@@ -89,14 +97,17 @@ void sendHandshake(long requestId, DiscoveryNode node, TcpChannel channel, TimeV
     }
 
     void handleHandshake(Version version, Set<String> features, TcpChannel channel, long requestId, StreamInput stream) throws IOException {
+        // The TransportService blocks incoming requests until this has been set.
+        assert localNode != null : "Local node must be set before handshake is handled";
+
         // Must read the handshake request to exhaust the stream
         HandshakeRequest handshakeRequest = new HandshakeRequest(stream);
         final int nextByte = stream.read();
         if (nextByte != -1) {
             throw new IllegalStateException("Handshake request not fully read for requestId [" + requestId + "], action ["
                 + TransportHandshaker.HANDSHAKE_ACTION_NAME + "], available [" + stream.available() + "]; resetting");
         }
-        HandshakeResponse response = new HandshakeResponse(this.version);
+        HandshakeResponse response = new HandshakeResponse(handshakeRequest.version, this.version, this.clusterName, this.localNode);
         handshakeResponseSender.sendResponse(version, features, channel, response, requestId);
     }
 
@@ -127,13 +138,13 @@ private HandshakeResponseHandler(long requestId, Version currentVersion, ActionL
 
         @Override
         public HandshakeResponse read(StreamInput in) throws IOException {
-            return new HandshakeResponse(in);
+            return new HandshakeResponse(this.currentVersion, in);
         }
 
         @Override
         public void handleResponse(HandshakeResponse response) {
             if (isDone.compareAndSet(false, true)) {
-                Version version = response.responseVersion;
+                Version version = response.version;
                 if (currentVersion.isCompatible(version) == false) {
                     listener.onFailure(new IllegalStateException("Received message from unsupported version: [" + version
                         + "] minimal compatible version is: [" + currentVersion.minimumCompatibilityVersion() + "]"));
@@ -201,25 +212,58 @@ public void writeTo(StreamOutput streamOutput) throws IOException {
 
     static final class HandshakeResponse extends TransportResponse {
 
-        private final Version responseVersion;
-
-        HandshakeResponse(Version responseVersion) {
-            this.responseVersion = responseVersion;
+        private final Version requestVersion;
+        private final Version version;
+        private final ClusterName clusterName;
+        private final DiscoveryNode discoveryNode;
+
+        HandshakeResponse(Version requestVersion, Version responseVersion, ClusterName clusterName, DiscoveryNode discoveryNode) {
+            this.requestVersion = requestVersion;
+            this.version = responseVersion;
+            this.clusterName = clusterName;
+            this.discoveryNode = discoveryNode;
         }
 
-        private HandshakeResponse(StreamInput in) throws IOException {
+        private HandshakeResponse(Version requestVersion, StreamInput in) throws IOException {
             super(in);
-            responseVersion = Version.readVersion(in);
+            this.requestVersion = requestVersion;
+            version = Version.readVersion(in);
+            // During the handshake process, nodes set their stream version to the minimum compatibility
+            // version they support. When deserializing the response, we use the version the other node
+            // told us that it actually is in the handshake response (`version`).
+            if (requestVersion.onOrAfter(Version.V_7_6_0) && version.onOrAfter(Version.V_7_6_0)) {
+                clusterName = new ClusterName(in);
+                discoveryNode = new DiscoveryNode(in);
+            } else {
+                clusterName = null;
+                discoveryNode = null;
+            }
         }
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
-            assert responseVersion != null;
-            Version.writeVersion(responseVersion, out);
+            assert version != null;
+            Version.writeVersion(version, out);
+            // During the handshake process, nodes set their stream version to the minimum compatibility
+            // version they support. When deciding what response to send, we use the version the other node
+            // told us that it actually is in the handshake request (`requestVersion`). If it did not tell
+            // us a `requestVersion`, it is at least a pre-7.6 node.
+            if (requestVersion != null && requestVersion.onOrAfter(Version.V_7_6_0) && version.onOrAfter(Version.V_7_6_0)) {
+                clusterName.writeTo(out);
+                discoveryNode.writeTo(out);
+            }
+        }
+
+        Version getVersion() {
+            return version;
+        }
+
+        ClusterName getClusterName() {
+            return clusterName;
         }
 
-        Version getResponseVersion() {
-            return responseVersion;
+        DiscoveryNode getDiscoveryNode() {
+            return discoveryNode;
         }
     }
 
diff --git a/server/src/main/java/org/elasticsearch/transport/TransportService.java b/server/src/main/java/org/elasticsearch/transport/TransportService.java
@@ -235,6 +235,7 @@ protected void doStart() {
             }
         }
         localNode = localNodeFactory.apply(transport.boundAddress());
+        transport.setLocalNode(localNode);
 
         if (connectToRemoteCluster) {
             // here we start to connect to the remote clusters
diff --git a/server/src/test/java/org/elasticsearch/client/transport/FailAndRetryMockTransport.java b/server/src/test/java/org/elasticsearch/client/transport/FailAndRetryMockTransport.java
@@ -229,4 +229,9 @@ public RequestHandlerRegistry getRequestHandler(String action) {
     public void setMessageListener(TransportMessageListener listener) {
         this.listener = listener;
     }
+
+    @Override
+    public void setLocalNode(DiscoveryNode localNode) {
+
+    }
 }
diff --git a/server/src/test/java/org/elasticsearch/cluster/NodeConnectionsServiceTests.java b/server/src/test/java/org/elasticsearch/cluster/NodeConnectionsServiceTests.java
@@ -415,6 +415,10 @@ public RequestHandlerRegistry getRequestHandler(String action) {
         public void setMessageListener(TransportMessageListener listener) {
         }
 
+        @Override
+        public void setLocalNode(DiscoveryNode localNode) {
+        }
+
         @Override
         public BoundTransportAddress boundAddress() {
             return null;
diff --git a/server/src/test/java/org/elasticsearch/transport/InboundHandlerTests.java b/server/src/test/java/org/elasticsearch/transport/InboundHandlerTests.java
@@ -21,6 +21,7 @@
 
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.Version;
+import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
@@ -58,9 +59,8 @@ public void setUp() throws Exception {
         channel = new FakeTcpChannel(randomBoolean(), buildNewFakeTransportAddress().address(), buildNewFakeTransportAddress().address());
         NamedWriteableRegistry namedWriteableRegistry = new NamedWriteableRegistry(Collections.emptyList());
         InboundMessage.Reader reader = new InboundMessage.Reader(version, namedWriteableRegistry, threadPool.getThreadContext());
-        TransportHandshaker handshaker = new TransportHandshaker(version, threadPool, (n, c, r, v) -> {
-        }, (v, f, c, r, r_id) -> {
-        });
+        TransportHandshaker handshaker = new TransportHandshaker(new ClusterName("cluster-name"), version, threadPool, (n, c, r, v) -> {
+        }, (v, f, c, r, r_id) -> {});
         TransportKeepAlive keepAlive = new TransportKeepAlive(threadPool, TcpChannel::sendMessage);
         OutboundHandler outboundHandler =
             new OutboundHandler("node", version, new String[0], threadPool, BigArrays.NON_RECYCLING_INSTANCE);
diff --git a/server/src/test/java/org/elasticsearch/transport/TransportHandshakerTests.java b/server/src/test/java/org/elasticsearch/transport/TransportHandshakerTests.java
@@ -20,16 +20,20 @@
 
 import org.elasticsearch.Version;
 import org.elasticsearch.action.support.PlainActionFuture;
+import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.common.SuppressForbidden;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.transport.TransportAddress;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.threadpool.TestThreadPool;
 import org.mockito.ArgumentCaptor;
 
 import java.io.IOException;
+import java.net.InetAddress;
 import java.util.Collections;
 import java.util.concurrent.TimeUnit;
 
@@ -42,23 +46,29 @@
 public class TransportHandshakerTests extends ESTestCase {
 
     private TransportHandshaker handshaker;
-    private DiscoveryNode node;
+    private DiscoveryNode remoteNode;
     private TcpChannel channel;
     private TestThreadPool threadPool;
     private TransportHandshaker.HandshakeRequestSender requestSender;
     private TransportHandshaker.HandshakeResponseSender responseSender;
+    private ClusterName clusterName;
+    private DiscoveryNode localNode;
 
     @Override
+    @SuppressForbidden(reason = "Allow accessing localhost")
     public void setUp() throws Exception {
         super.setUp();
-        String nodeId = "node-id";
+        String nodeId = "remote-node-id";
         channel = mock(TcpChannel.class);
         requestSender = mock(TransportHandshaker.HandshakeRequestSender.class);
         responseSender = mock(TransportHandshaker.HandshakeResponseSender.class);
-        node = new DiscoveryNode(nodeId, nodeId, nodeId, "host", "host_address", buildNewFakeTransportAddress(), Collections.emptyMap(),
-            Collections.emptySet(), Version.CURRENT);
+        remoteNode = new DiscoveryNode(nodeId, nodeId, nodeId, "host", "host_address", buildNewFakeTransportAddress(),
+            Collections.emptyMap(), Collections.emptySet(), Version.CURRENT);
         threadPool = new TestThreadPool("thread-poll");
-        handshaker = new TransportHandshaker(Version.CURRENT, threadPool, requestSender, responseSender);
+        clusterName = new ClusterName("cluster");
+        localNode = new DiscoveryNode("local-node-id", new TransportAddress(InetAddress.getLocalHost(), 0), Version.CURRENT);
+        handshaker = new TransportHandshaker(clusterName, Version.CURRENT, threadPool, requestSender, responseSender);
+        handshaker.setLocalNode(localNode);
     }
 
     @Override
@@ -70,9 +80,9 @@ public void tearDown() throws Exception {
     public void testHandshakeRequestAndResponse() throws IOException {
         PlainActionFuture<Version> versionFuture = PlainActionFuture.newFuture();
         long reqId = randomLongBetween(1, 10);
-        handshaker.sendHandshake(reqId, node, channel, new TimeValue(30, TimeUnit.SECONDS), versionFuture);
+        handshaker.sendHandshake(reqId, remoteNode, channel, new TimeValue(30, TimeUnit.SECONDS), versionFuture);
 
-        verify(requestSender).sendRequest(node, channel, reqId, Version.CURRENT.minimumCompatibilityVersion());
+        verify(requestSender).sendRequest(remoteNode, channel, reqId, Version.CURRENT.minimumCompatibilityVersion());
 
         assertFalse(versionFuture.isDone());
 
@@ -88,18 +98,39 @@ public void testHandshakeRequestAndResponse() throws IOException {
         verify(responseSender).sendResponse(eq(Version.CURRENT), eq(Collections.emptySet()), eq(mockChannel), responseCaptor.capture(),
             eq(reqId));
 
+
         TransportResponseHandler<TransportHandshaker.HandshakeResponse> handler = handshaker.removeHandlerForHandshake(reqId);
         handler.handleResponse((TransportHandshaker.HandshakeResponse) responseCaptor.getValue());
 
         assertTrue(versionFuture.isDone());
         assertEquals(Version.CURRENT, versionFuture.actionGet());
+        TransportHandshaker.HandshakeResponse response = (TransportHandshaker.HandshakeResponse) responseCaptor.getValue();
+        assertEquals(Version.CURRENT, response.getVersion());
+        assertEquals(clusterName, response.getClusterName());
+        assertEquals(localNode, response.getDiscoveryNode());
+    }
+
+    public void testHandshakeRequestAndResponsePreV7_6() throws IOException {
+        PlainActionFuture<Version> versionFuture = PlainActionFuture.newFuture();
+        long reqId = randomLongBetween(1, 10);
+        handshaker.sendHandshake(reqId, remoteNode, channel, new TimeValue(30, TimeUnit.SECONDS), versionFuture);
+
+        TransportResponseHandler<TransportHandshaker.HandshakeResponse> handler = handshaker.removeHandlerForHandshake(reqId);
+        try (BytesStreamOutput out = new BytesStreamOutput()) {
+            new TransportHandshaker.HandshakeResponse(Version.V_7_5_0, Version.V_7_5_0, clusterName, localNode).writeTo(out);
+            TransportHandshaker.HandshakeResponse response = handler.read(out.bytes().streamInput());
+            assertEquals(Version.V_7_5_0, response.getVersion());
+            // When writing or reading a 6.6 stream, these are not serialized
+            assertNull(response.getDiscoveryNode());
+            assertNull(response.getClusterName());
+        }
     }
 
     public void testHandshakeRequestFutureVersionsCompatibility() throws IOException {
         long reqId = randomLongBetween(1, 10);
-        handshaker.sendHandshake(reqId, node, channel, new TimeValue(30, TimeUnit.SECONDS), PlainActionFuture.newFuture());
+        handshaker.sendHandshake(reqId, remoteNode, channel, new TimeValue(30, TimeUnit.SECONDS), PlainActionFuture.newFuture());
 
-        verify(requestSender).sendRequest(node, channel, reqId, Version.CURRENT.minimumCompatibilityVersion());
+        verify(requestSender).sendRequest(remoteNode, channel, reqId, Version.CURRENT.minimumCompatibilityVersion());
 
         TcpChannel mockChannel = mock(TcpChannel.class);
         TransportHandshaker.HandshakeRequest handshakeRequest = new TransportHandshaker.HandshakeRequest(Version.CURRENT);
@@ -131,15 +162,15 @@ public void testHandshakeRequestFutureVersionsCompatibility() throws IOException
 
         TransportHandshaker.HandshakeResponse response = (TransportHandshaker.HandshakeResponse) responseCaptor.getValue();
 
-        assertEquals(Version.CURRENT, response.getResponseVersion());
+        assertEquals(Version.CURRENT, response.getVersion());
     }
 
     public void testHandshakeError() throws IOException {
         PlainActionFuture<Version> versionFuture = PlainActionFuture.newFuture();
         long reqId = randomLongBetween(1, 10);
-        handshaker.sendHandshake(reqId, node, channel, new TimeValue(30, TimeUnit.SECONDS), versionFuture);
+        handshaker.sendHandshake(reqId, remoteNode, channel, new TimeValue(30, TimeUnit.SECONDS), versionFuture);
 
-        verify(requestSender).sendRequest(node, channel, reqId, Version.CURRENT.minimumCompatibilityVersion());
+        verify(requestSender).sendRequest(remoteNode, channel, reqId, Version.CURRENT.minimumCompatibilityVersion());
 
         assertFalse(versionFuture.isDone());
 
@@ -155,9 +186,9 @@ public void testSendRequestThrowsException() throws IOException {
         PlainActionFuture<Version> versionFuture = PlainActionFuture.newFuture();
         long reqId = randomLongBetween(1, 10);
         Version compatibilityVersion = Version.CURRENT.minimumCompatibilityVersion();
-        doThrow(new IOException("boom")).when(requestSender).sendRequest(node, channel, reqId, compatibilityVersion);
+        doThrow(new IOException("boom")).when(requestSender).sendRequest(remoteNode, channel, reqId, compatibilityVersion);
 
-        handshaker.sendHandshake(reqId, node, channel, new TimeValue(30, TimeUnit.SECONDS), versionFuture);
+        handshaker.sendHandshake(reqId, remoteNode, channel, new TimeValue(30, TimeUnit.SECONDS), versionFuture);
 
         assertTrue(versionFuture.isDone());
         ConnectTransportException cte = expectThrows(ConnectTransportException.class, versionFuture::actionGet);
@@ -168,9 +199,9 @@ public void testSendRequestThrowsException() throws IOException {
     public void testHandshakeTimeout() throws IOException {
         PlainActionFuture<Version> versionFuture = PlainActionFuture.newFuture();
         long reqId = randomLongBetween(1, 10);
-        handshaker.sendHandshake(reqId, node, channel, new TimeValue(100, TimeUnit.MILLISECONDS), versionFuture);
+        handshaker.sendHandshake(reqId, remoteNode, channel, new TimeValue(100, TimeUnit.MILLISECONDS), versionFuture);
 
-        verify(requestSender).sendRequest(node, channel, reqId, Version.CURRENT.minimumCompatibilityVersion());
+        verify(requestSender).sendRequest(remoteNode, channel, reqId, Version.CURRENT.minimumCompatibilityVersion());
 
         ConnectTransportException cte = expectThrows(ConnectTransportException.class, versionFuture::actionGet);
         assertThat(cte.getMessage(), containsString("handshake_timeout"));
diff --git a/test/framework/src/main/java/org/elasticsearch/test/transport/MockTransport.java b/test/framework/src/main/java/org/elasticsearch/test/transport/MockTransport.java
diff --git a/test/framework/src/main/java/org/elasticsearch/test/transport/StubbableTransport.java b/test/framework/src/main/java/org/elasticsearch/test/transport/StubbableTransport.java

Original file line number	Diff line number	Diff line change
`@@ -235,6 +235,7 @@ protected void doStart() {`
`235`	`235`	`}`
`236`	`236`	`}`
`237`	`237`	`localNode = localNodeFactory.apply(transport.boundAddress());`
	`238`	`+ transport.setLocalNode(localNode);`
`238`	`239`
`239`	`240`	`if (connectToRemoteCluster) {`
`240`	`241`	`// here we start to connect to the remote clusters`
Original file line number	Diff line number	Diff line change
`@@ -229,4 +229,9 @@ public RequestHandlerRegistry getRequestHandler(String action) {`
`229`	`229`	`public void setMessageListener(TransportMessageListener listener) {`
`230`	`230`	`this.listener = listener;`
`231`	`231`	`}`
	`232`	`+`
	`233`	`+ @Override`
	`234`	`+ public void setLocalNode(DiscoveryNode localNode) {`
	`235`	`+`
	`236`	`+ }`
`232`	`237`	`}`