Make remote setting updates support diff strategies (elastic#47891)

Currently the entire remote cluster settings infrastructure is designed
around the sniff strategy. As we introduce an additional conneciton
strategy this infrastructure needs to be modified to support it. This
commit modifies the code so that the strategy implementations will tell
the service if the connection needs to be torn down and rebuilt.

As part of this commit, we will wait 10 seconds for new clusters to
connect when they are added through the "update" settings
infrastructure.

Author: Tim-Brooks

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

@@ -25,7 +25,6 @@
 import org.elasticsearch.action.support.ContextPreservingActionListener;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
-import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
@@ -36,11 +35,7 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.Collections;
-import java.util.List;
 import java.util.function.Function;
-import java.util.function.Predicate;
-import java.util.function.Supplier;
-import java.util.stream.Collectors;
 
 /**
  * Represents a connection to a single remote cluster. In contrast to a local cluster a remote cluster is not joined such that the
@@ -61,46 +56,29 @@ final class RemoteClusterConnection implements Closeable {
     private final RemoteConnectionManager remoteConnectionManager;
     private final RemoteConnectionStrategy connectionStrategy;
     private final String clusterAlias;
-    private final int maxNumRemoteConnections;
     private final ThreadPool threadPool;
-    private final List<Tuple<String, Supplier<DiscoveryNode>>> seedNodes;
-    private final String proxyAddress;
     private volatile boolean skipUnavailable;
     private final TimeValue initialConnectionTimeout;
 
     /**
      * Creates a new {@link RemoteClusterConnection}
      * @param settings the nodes settings object
      * @param clusterAlias the configured alias of the cluster to connect to
-     * @param seedNodes a list of seed nodes to discover eligible nodes from
      * @param transportService the local nodes transport service
-     * @param maxNumRemoteConnections the maximum number of connections to the remote cluster
-     * @param nodePredicate a predicate to filter eligible remote nodes to connect to
-     * @param proxyAddress the proxy address
-     * @param connectionProfile the connection profile to use
      */
-    RemoteClusterConnection(Settings settings, String clusterAlias, List<Tuple<String, Supplier<DiscoveryNode>>> seedNodes,
-                            TransportService transportService, int maxNumRemoteConnections, Predicate<DiscoveryNode> nodePredicate,
-                            String proxyAddress, ConnectionProfile connectionProfile) {
-        this(settings, clusterAlias, seedNodes, transportService, maxNumRemoteConnections, nodePredicate, proxyAddress,
-            createConnectionManager(connectionProfile, transportService));
+    RemoteClusterConnection(Settings settings, String clusterAlias, TransportService transportService) {
+        this(settings, clusterAlias, transportService,
+            createConnectionManager(buildConnectionProfileFromSettings(settings, clusterAlias), transportService));
     }
 
-    // Public for tests to pass a StubbableConnectionManager
-    RemoteClusterConnection(Settings settings, String clusterAlias, List<Tuple<String, Supplier<DiscoveryNode>>> seedNodes,
-                            TransportService transportService, int maxNumRemoteConnections, Predicate<DiscoveryNode> nodePredicate,
-                            String proxyAddress, ConnectionManager connectionManager) {
+    RemoteClusterConnection(Settings settings, String clusterAlias, TransportService transportService,
+                            ConnectionManager connectionManager) {
         this.transportService = transportService;
-        this.maxNumRemoteConnections = maxNumRemoteConnections;
         this.clusterAlias = clusterAlias;
         this.remoteConnectionManager = new RemoteConnectionManager(clusterAlias, connectionManager);
-        this.connectionStrategy = new SniffConnectionStrategy(clusterAlias, transportService, remoteConnectionManager,
-            proxyAddress, maxNumRemoteConnections, nodePredicate,
-            Collections.unmodifiableList(seedNodes));
+        this.connectionStrategy = new SniffConnectionStrategy(clusterAlias, transportService, remoteConnectionManager, settings);
         // we register the transport service here as a listener to make sure we notify handlers on disconnect etc.
         connectionManager.addListener(transportService);
-        this.seedNodes = Collections.unmodifiableList(seedNodes);
-        this.proxyAddress = proxyAddress;
         this.skipUnavailable = RemoteClusterService.REMOTE_CLUSTER_SKIP_UNAVAILABLE
             .getConcreteSettingForNamespace(clusterAlias).get(settings);
         this.threadPool = transportService.threadPool;
@@ -125,11 +103,11 @@ boolean isSkipUnavailable() {
      * Ensures that this cluster is connected. If the cluster is connected this operation
      * will invoke the listener immediately.
      */
-    void ensureConnected(ActionListener<Void> voidActionListener) {
+    void ensureConnected(ActionListener<Void> listener) {
         if (remoteConnectionManager.size() == 0) {
-            connectionStrategy.connect(voidActionListener);
+            connectionStrategy.connect(listener);
         } else {
-            voidActionListener.onResponse(null);
+            listener.onResponse(null);
         }
     }
 
@@ -215,10 +193,6 @@ public boolean isClosed() {
         return connectionStrategy.isClosed();
     }
 
-    List<Tuple<String, Supplier<DiscoveryNode>>> getSeedNodes() {
-        return seedNodes;
-    }
-
     // for testing only
     boolean assertNoRunningConnections() {
         return connectionStrategy.assertNoRunningConnections();
@@ -232,13 +206,24 @@ boolean isNodeConnected(final DiscoveryNode node) {
      * Get the information about remote nodes to be rendered on {@code _remote/info} requests.
      */
     public RemoteConnectionInfo getConnectionInfo() {
-        return new RemoteConnectionInfo(
+        if (connectionStrategy instanceof SniffConnectionStrategy) {
+            SniffConnectionStrategy sniffStrategy = (SniffConnectionStrategy) this.connectionStrategy;
+            return new RemoteConnectionInfo(
+                clusterAlias,
+                sniffStrategy.getSeedNodes(),
+                sniffStrategy.getMaxConnections(),
+                getNumNodesConnected(),
+                initialConnectionTimeout,
+                skipUnavailable);
+        } else {
+            return new RemoteConnectionInfo(
                 clusterAlias,
-                seedNodes.stream().map(Tuple::v1).collect(Collectors.toList()),
-                maxNumRemoteConnections,
+                Collections.emptyList(),
+                0,
                 getNumNodesConnected(),
                 initialConnectionTimeout,
                 skipUnavailable);
+        }
     }
 
     int getNumNodesConnected() {
@@ -253,7 +238,21 @@ ConnectionManager getConnectionManager() {
         return remoteConnectionManager.getConnectionManager();
     }
 
-    public String getProxyAddress() {
-        return proxyAddress;
+    public boolean shouldRebuildConnection(Settings newSettings) {
+        return connectionStrategy.shouldRebuildConnection(newSettings);
+    }
+
+    static ConnectionProfile buildConnectionProfileFromSettings(Settings settings, String clusterName) {
+        return new ConnectionProfile.Builder()
+            .setConnectTimeout(TransportSettings.CONNECT_TIMEOUT.get(settings))
+            .setHandshakeTimeout(TransportSettings.CONNECT_TIMEOUT.get(settings))
+            .addConnections(6, TransportRequestOptions.Type.REG, TransportRequestOptions.Type.PING) // TODO make this configurable?
+            // we don't want this to be used for anything else but search
+            .addConnections(0, TransportRequestOptions.Type.BULK,
+                TransportRequestOptions.Type.STATE,
+                TransportRequestOptions.Type.RECOVERY)
+            .setCompressionEnabled(RemoteClusterService.REMOTE_CLUSTER_COMPRESS.getConcreteSettingForNamespace(clusterName).get(settings))
+            .setPingInterval(RemoteClusterService.REMOTE_CLUSTER_PING_SCHEDULE.getConcreteSettingForNamespace(clusterName).get(settings))
+            .build();
     }
 }