Extract remote "sniffing" to connection strategy (elastic#47253)

Currently the connection strategy used by the remote cluster service is
implemented as a multi-step sniffing process in the
RemoteClusterConnection. We intend to introduce a new connection strategy
that will operate in a different manner. This commit extracts the
sniffing logic to a dedicated strategy class. Additionally, it implements
dedicated tests for this class.

Additionally, in previous commits we moved away from a world where the
remote cluster connection was mutable. Instead, when setting updates are
made, the connection is torn down and rebuilt. We still had methods and
tests hanging around for the mutable behavior. This commit removes those.

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

@@ -234,7 +234,7 @@ private synchronized void updateRemoteClusters(Map<String, Tuple<String, List<Tu
                         getNodePredicate(settings), proxyAddress, connectionProfile);
                     remoteClusters.put(clusterAlias, remote);
                 } else if (connectionProfileChanged(remote.getConnectionManager().getConnectionProfile(), connectionProfile)
-                        || seedsChanged(remote.getSeedNodes(), seedList)) {
+                        || seedsChanged(remote.getSeedNodes(), seedList) || proxyChanged(proxyAddress, remote.getProxyAddress())) {
                     // New ConnectionProfile. Must tear down existing connection
                     try {
                         IOUtils.close(remote);
@@ -249,7 +249,7 @@ private synchronized void updateRemoteClusters(Map<String, Tuple<String, List<Tu
 
                 // now update the seed nodes no matter if it's new or already existing
                 RemoteClusterConnection finalRemote = remote;
-                remote.updateSeedNodes(proxyAddress, seedList, ActionListener.wrap(
+                remote.ensureConnected(ActionListener.wrap(
                         response -> {
                             if (countDown.countDown()) {
                                 connectionListener.onResponse(response);
@@ -483,6 +483,14 @@ private boolean seedsChanged(final List<Tuple<String, Supplier<DiscoveryNode>>>
         return oldSeeds.equals(newSeeds) == false;
     }
 
+    private boolean proxyChanged(String oldProxy, String newProxy) {
+        if (oldProxy == null || oldProxy.isEmpty()) {
+            return (newProxy == null || newProxy.isEmpty()) == false;
+        }
+
+        return Objects.equals(oldProxy, newProxy) == false;
+    }
+
     /**
      * Collects all nodes of the given clusters and returns / passes a (clusterAlias, nodeId) to {@link DiscoveryNode}
      * function on success.

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

@@ -0,0 +1,164 @@
+/*
+ * 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.transport;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.message.ParameterizedMessage;
+import org.apache.lucene.store.AlreadyClosedException;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.support.ContextPreservingActionListener;
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.common.util.concurrent.AbstractRunnable;
+import org.elasticsearch.threadpool.ThreadPool;
+
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public abstract class RemoteConnectionStrategy implements TransportConnectionListener, Closeable {
+
+    protected static final Logger logger = LogManager.getLogger(RemoteConnectionStrategy.class);
+
+    private static final int MAX_LISTENERS = 100;
+    private final AtomicBoolean closed = new AtomicBoolean(false);
+    private final Object mutex = new Object();
+    private final ThreadPool threadPool;
+    protected final RemoteConnectionManager connectionManager;
+    private List<ActionListener<Void>> listeners = new ArrayList<>();
+
+    RemoteConnectionStrategy(ThreadPool threadPool, RemoteConnectionManager connectionManager) {
+        this.threadPool = threadPool;
+        this.connectionManager = connectionManager;
+        connectionManager.getConnectionManager().addListener(this);
+    }
+
+    /**
+     * Triggers a connect round unless there is one running already. If there is a connect round running, the listener will either
+     * be queued or rejected and failed.
+     */
+    void connect(ActionListener<Void> connectListener) {
+        boolean runConnect = false;
+        final ActionListener<Void> listener =
+            ContextPreservingActionListener.wrapPreservingContext(connectListener, threadPool.getThreadContext());
+        boolean closed;
+        synchronized (mutex) {
+            closed = this.closed.get();
+            if (closed) {
+                assert listeners.isEmpty();
+            } else {
+                if (listeners.size() >= MAX_LISTENERS) {
+                    assert listeners.size() == MAX_LISTENERS;
+                    listener.onFailure(new RejectedExecutionException("connect listener queue is full"));
+                    return;
+                } else {
+                    listeners.add(listener);
+                }
+                runConnect = listeners.size() == 1;
+            }
+        }
+        if (closed) {
+            connectListener.onFailure(new AlreadyClosedException("connect handler is already closed"));
+            return;
+        }
+        if (runConnect) {
+            ExecutorService executor = threadPool.executor(ThreadPool.Names.MANAGEMENT);
+            executor.submit(new AbstractRunnable() {
+                @Override
+                public void onFailure(Exception e) {
+                    ActionListener.onFailure(getAndClearListeners(), e);
+                }
+
+                @Override
+                protected void doRun() {
+                    connectImpl(new ActionListener<Void>() {
+                        @Override
+                        public void onResponse(Void aVoid) {
+                            ActionListener.onResponse(getAndClearListeners(), aVoid);
+                        }
+
+                        @Override
+                        public void onFailure(Exception e) {
+                            ActionListener.onFailure(getAndClearListeners(), e);
+                        }
+                    });
+                }
+            });
+        }
+    }
+
+    @Override
+    public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connection) {
+        if (shouldOpenMoreConnections()) {
+            // try to reconnect and fill up the slot of the disconnected node
+            connect(ActionListener.wrap(
+                ignore -> logger.trace("successfully connected after disconnect of {}", node),
+                e -> logger.trace(() -> new ParameterizedMessage("failed to connect after disconnect of {}", node), e)));
+        }
+    }
+
+    @Override
+    public void close() {
+        final List<ActionListener<Void>> toNotify;
+        synchronized (mutex) {
+            if (closed.compareAndSet(false, true)) {
+                connectionManager.getConnectionManager().removeListener(this);
+                toNotify = listeners;
+                listeners = Collections.emptyList();
+            } else {
+                toNotify = Collections.emptyList();
+            }
+        }
+        ActionListener.onFailure(toNotify, new AlreadyClosedException("connect handler is already closed"));
+    }
+
+    public boolean isClosed() {
+        return closed.get();
+    }
+
+    // for testing only
+    boolean assertNoRunningConnections() {
+        synchronized (mutex) {
+            assert listeners.isEmpty();
+        }
+        return true;
+    }
+
+    protected abstract boolean shouldOpenMoreConnections();
+
+    protected abstract void connectImpl(ActionListener<Void> listener);
+
+    private List<ActionListener<Void>> getAndClearListeners() {
+        final List<ActionListener<Void>> result;
+        synchronized (mutex) {
+            if (listeners.isEmpty()) {
+                result = Collections.emptyList();
+            } else {
+                result = listeners;
+                listeners = new ArrayList<>();
+            }
+        }
+        return result;
+    }
+}