Pass the host name on as server_name if proxy mode is on (elastic#34559)

In remote cluster setup if we see a configured proxy we should set the seed nodes host name as the `server_name` to trigger SNI based routing even for seed nodes. Since remote cluster connections are plain TCP connections we have to set the host manually since the other side can't take it from the request URL like in the HTTP case. This also adds some more informative logging to remote cluster connection.

Author: s1monw

server/src/main/java/org/elasticsearch/transport/RemoteClusterAware.java

@@ -223,8 +223,7 @@ static DiscoveryNode buildSeedNode(String clusterName, String address, boolean p
  TransportAddress transportAddress = new TransportAddress(TransportAddress.META_ADDRESS, 0);
  String hostName = address.substring(0, indexOfPortSeparator(address));
  return new DiscoveryNode("", clusterName + "#" + address, UUIDs.randomBase64UUID(), hostName, address,
- transportAddress, Collections
- .emptyMap(), EnumSet.allOf(DiscoveryNode.Role.class),
+ transportAddress, Collections.singletonMap("server_name", hostName), EnumSet.allOf(DiscoveryNode.Role.class),
  Version.CURRENT.minimumCompatibilityVersion());
  } else {
  TransportAddress transportAddress = new TransportAddress(RemoteClusterAware.parseSeedAddress(address));

server/src/main/java/org/elasticsearch/transport/RemoteClusterConnection.java

@@ -490,6 +490,8 @@ private void collectRemoteNodes(Iterator<Supplier<DiscoveryNode>> seedNodes,
  if (seedNodes.hasNext()) {
  cancellableThreads.executeIO(() -> {
  final DiscoveryNode seedNode = maybeAddProxyAddress(proxyAddress, seedNodes.next().get());
+ logger.debug("[{}] opening connection to seed node: [{}] proxy address: [{}]", clusterAlias, seedNode,
+ proxyAddress);
  final TransportService.HandshakeResponse handshakeResponse;
  Transport.Connection connection = manager.openConnection(seedNode,
  ConnectionProfile.buildSingleChannelProfile(TransportRequestOptions.Type.REG, null, null));
@@ -523,7 +525,7 @@ private void collectRemoteNodes(Iterator<Supplier<DiscoveryNode>> seedNodes,
  ThreadContext threadContext = threadPool.getThreadContext();
  TransportService.ContextRestoreResponseHandler<ClusterStateResponse> responseHandler = new TransportService
  .ContextRestoreResponseHandler<>(threadContext.newRestorableContext(false),
- new SniffClusterStateResponseHandler(transportService, connection, listener, seedNodes,
+ new SniffClusterStateResponseHandler(connection, listener, seedNodes,
  cancellableThreads));
  try (ThreadContext.StoredContext ignore = threadContext.stashContext()) {
  // we stash any context here since this is an internal execution and should not leak any
@@ -543,13 +545,16 @@ private void collectRemoteNodes(Iterator<Supplier<DiscoveryNode>> seedNodes,
  listener.onFailure(new IllegalStateException("no seed node left"));
  }
  } catch (CancellableThreads.ExecutionCancelledException ex) {
+ logger.warn(() -> new ParameterizedMessage("fetching nodes from external cluster [{}] failed", clusterAlias), ex);
  listener.onFailure(ex); // we got canceled - fail the listener and step out
  } catch (ConnectTransportException | IOException | IllegalStateException ex) {
  // ISE if we fail the handshake with an version incompatible node
  if (seedNodes.hasNext()) {
- logger.debug(() -> new ParameterizedMessage("fetching nodes from external cluster {} failed", clusterAlias), ex);
+ logger.debug(() -> new ParameterizedMessage("fetching nodes from external cluster [{}] failed moving to next node",
+ clusterAlias), ex);
  collectRemoteNodes(seedNodes, transportService, manager, listener);
  } else {
+ logger.warn(() -> new ParameterizedMessage("fetching nodes from external cluster [{}] failed", clusterAlias), ex);
  listener.onFailure(ex);
  }
  }
@@ -581,8 +586,8 @@ private class SniffClusterStateResponseHandler implements TransportResponseHandl
  private final Iterator<Supplier<DiscoveryNode>> seedNodes;
  private final CancellableThreads cancellableThreads;
 
- SniffClusterStateResponseHandler(TransportService transportService, Transport.Connection connection,
- ActionListener<Void> listener, Iterator<Supplier<DiscoveryNode>> seedNodes,
+ SniffClusterStateResponseHandler(Transport.Connection connection, ActionListener<Void> listener,
+ Iterator<Supplier<DiscoveryNode>> seedNodes,
  CancellableThreads cancellableThreads) {
  this.connection = connection;
  this.listener = listener;

server/src/test/java/org/elasticsearch/transport/RemoteClusterConnectionTests.java

@@ -1483,6 +1483,7 @@ public void testProxyMode() throws Exception {
  service.acceptIncomingRequests();
  Supplier<DiscoveryNode> seedSupplier = () ->
  RemoteClusterAware.buildSeedNode("some-remote-cluster", "node_0:" + randomIntBetween(1, 10000), true);
+ assertEquals("node_0", seedSupplier.get().getAttributes().get("server_name"));
  try (RemoteClusterConnection connection = new RemoteClusterConnection(Settings.EMPTY, "test-cluster",
  Arrays.asList(seedSupplier), service, service.getConnectionManager(), Integer.MAX_VALUE, n -> true, proxyAddress)) {
  updateSeedNodes(connection, Arrays.asList(seedSupplier), proxyAddress);