Simplify Unicast Zen Ping (elastic#22277)

The `UnicastZenPing` shows it's age and is the result of many small changes. The current state of affairs is confusing and is hard to reason about. This PR cleans it up (while following the same original intentions). Highlights of the changes are:

1) Clear 3 round flow - no interleaving of scheduling.
2) The previous implementation did a best effort attempt to wait for ongoing pings to be sent and completed. The pings were guaranteed to complete because each used the total ping duration as a timeout. This did make it hard to reason about the total ping duration and the flow of the code. All of this is removed now and ping should just complete within the given duration or not be counted (note that it was very handy for testing, but I move the needed sync logic to the test).
3) Because of (2) the pinging scheduling changed a bit, to give a chance for the last round to complete. We now ping at the beginning, 1/3 and 2/3 of the duration.
4) To offset for (3) a bit, incoming ping requests are now added to on going ping collections.
5) UnicastZenPing never establishes full blown connections (but does reuse them if there). Relates to elastic#22120
6) Discovery host providers are only used once per pinging round. Closes elastic#21739
7) Usage of the ability to open a connection without connecting to a node ( elastic#22194 ) and shorter connection timeouts helps with connections piling up. Closes elastic#19370
8) Beefed up testing and sped them up.
9) removed light profile from production code

Author: bleskes

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

@@ -101,6 +101,21 @@ final class TransportClientNodesService extends AbstractComponent implements Clo
 
     private final TransportClient.HostFailureListener hostFailureListener;
 
+    // TODO: migrate this to use low level connections and single type channels
+    /** {@link ConnectionProfile} to use when to connecting to the listed nodes and doing a liveness check */
+    private static final ConnectionProfile LISTED_NODES_PROFILE;
+
+    static {
+        ConnectionProfile.Builder builder = new ConnectionProfile.Builder();
+        builder.addConnections(1,
+            TransportRequestOptions.Type.BULK,
+            TransportRequestOptions.Type.PING,
+            TransportRequestOptions.Type.RECOVERY,
+            TransportRequestOptions.Type.REG,
+            TransportRequestOptions.Type.STATE);
+        LISTED_NODES_PROFILE = builder.build();
+    }
+
     TransportClientNodesService(Settings settings, TransportService transportService,
                                        ThreadPool threadPool, TransportClient.HostFailureListener hostFailureListener) {
         super(settings);
@@ -389,8 +404,8 @@ protected void doSample() {
                 if (!transportService.nodeConnected(listedNode)) {
                     try {
                         // its a listed node, light connect to it...
-                        logger.trace("connecting to listed node (light) [{}]", listedNode);
-                        transportService.connectToNode(listedNode, ConnectionProfile.LIGHT_PROFILE);
+                        logger.trace("connecting to listed node [{}]", listedNode);
+                        transportService.connectToNode(listedNode, LISTED_NODES_PROFILE);
                     } catch (Exception e) {
                         logger.info(
                             (Supplier<?>)
@@ -470,7 +485,7 @@ public void run() {
                                     } else {
                                         // its a listed node, light connect to it...
                                         logger.trace("connecting to listed node (light) [{}]", listedNode);
-                                        transportService.connectToNode(listedNode, ConnectionProfile.LIGHT_PROFILE);
+                                        transportService.connectToNode(listedNode, LISTED_NODES_PROFILE);
                                     }
                                 } catch (Exception e) {
                                     logger.debug(

core/src/main/java/org/elasticsearch/discovery/zen/ElectMasterService.java

@@ -24,7 +24,6 @@
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.component.AbstractComponent;
-import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.settings.Settings;
@@ -174,7 +173,7 @@ public void logMinimumMasterNodesWarningIfNecessary(ClusterState oldState, Clust
      * Returns the given nodes sorted by likelihood of being elected as master, most likely first.
      * Non-master nodes are not removed but are rather put in the end
      */
-    public static List<DiscoveryNode> sortByMasterLikelihood(Iterable<DiscoveryNode> nodes) {
+    static List<DiscoveryNode> sortByMasterLikelihood(Iterable<DiscoveryNode> nodes) {
         ArrayList<DiscoveryNode> sortedNodes = CollectionUtils.iterableAsArrayList(nodes);
         CollectionUtil.introSort(sortedNodes, ElectMasterService::compareNodes);
         return sortedNodes;

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

@@ -67,11 +67,11 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Set;
-import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.function.BiFunction;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
@@ -1021,24 +1021,22 @@ public void handleException(TransportException exp) {
     }
 
     private ZenPing.PingCollection pingAndWait(TimeValue timeout) {
-        final ZenPing.PingCollection response = new ZenPing.PingCollection();
-        final CountDownLatch latch = new CountDownLatch(1);
+        final CompletableFuture<ZenPing.PingCollection> response = new CompletableFuture<>();
         try {
-            zenPing.ping(pings -> {
-                response.addPings(pings);
-                latch.countDown();
-            }, timeout);
+            zenPing.ping(response::complete, timeout);
         } catch (Exception ex) {
-            logger.warn("Ping execution failed", ex);
-            latch.countDown();
+            // logged later
+            response.completeExceptionally(ex);
         }
 
         try {
-            latch.await();
-            return response;
+            return response.get();
         } catch (InterruptedException e) {
             logger.trace("pingAndWait interrupted");
-            return response;
+            return new ZenPing.PingCollection();
+        } catch (ExecutionException e) {
+            logger.warn("Ping execution failed", e);
+            return new ZenPing.PingCollection();
         }
     }
 

core/src/main/java/org/elasticsearch/discovery/zen/ZenDiscovery.java

@@ -30,29 +30,19 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Consumer;
 
 import static org.elasticsearch.gateway.GatewayService.STATE_NOT_RECOVERED_BLOCK;
 
 public interface ZenPing extends Releasable {
 
     void start(PingContextProvider contextProvider);
 
-    void ping(PingListener listener, TimeValue timeout);
-
-    interface PingListener {
-
-        /**
-         * called when pinging is done.
-         *
-         * @param pings ping result *must
-         */
-        void onPing(Collection<PingResponse> pings);
-    }
+    void ping(Consumer<PingCollection> resultsConsumer, TimeValue timeout);
 
     class PingResponse implements Streamable {
 
@@ -191,13 +181,6 @@ public synchronized boolean addPing(PingResponse ping) {
             return false;
         }
 
-        /** adds multiple pings if newer than previous pings from the same node */
-        public synchronized void addPings(Iterable<PingResponse> pings) {
-            for (PingResponse ping : pings) {
-                addPing(ping);
-            }
-        }
-
         /** serialize current pings to a list. It is guaranteed that the list contains one ping response per node */
         public synchronized List<PingResponse> toList() {
             return new ArrayList<>(pings.values());

core/src/main/java/org/elasticsearch/discovery/zen/ZenPing.java

@@ -18,6 +18,7 @@
  */
 package org.elasticsearch.transport;
 
+import org.elasticsearch.common.inject.internal.Nullable;
 import org.elasticsearch.common.unit.TimeValue;
 
 import java.util.ArrayList;
@@ -35,16 +36,25 @@
 public final class ConnectionProfile {
 
     /**
-     * A pre-built light connection profile that shares a single connection across all
-     * types.
+     * Builds a connection profile that is dedicated to a single channel type. Use this
+     * when opening single use connections
      */
-    public static final ConnectionProfile LIGHT_PROFILE = new ConnectionProfile(
-        Collections.singletonList(new ConnectionTypeHandle(0, 1, EnumSet.of(
-            TransportRequestOptions.Type.BULK,
-            TransportRequestOptions.Type.PING,
-            TransportRequestOptions.Type.RECOVERY,
-            TransportRequestOptions.Type.REG,
-            TransportRequestOptions.Type.STATE))), 1, null, null);
+    public static ConnectionProfile buildSingleChannelProfile(TransportRequestOptions.Type channelType,
+                                                              @Nullable TimeValue connectTimeout,
+                                                               @Nullable TimeValue handshakeTimeout) {
+        Builder builder = new Builder();
+        builder.addConnections(1, channelType);
+        final EnumSet<TransportRequestOptions.Type> otherTypes = EnumSet.allOf(TransportRequestOptions.Type.class);
+        otherTypes.remove(channelType);
+        builder.addConnections(0, otherTypes.stream().toArray(TransportRequestOptions.Type[]::new));
+        if (connectTimeout != null) {
+            builder.setConnectTimeout(connectTimeout);
+        }
+        if (handshakeTimeout != null) {
+            builder.setHandshakeTimeout(handshakeTimeout);
+        }
+        return builder.build();
+    }
 
     private final List<ConnectionTypeHandle> handles;
     private final int numConnections;

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

@@ -63,8 +63,7 @@ public interface Transport extends LifecycleComponent {
     boolean nodeConnected(DiscoveryNode node);
 
     /**
-     * Connects to a node with the given connection profile. Use {@link ConnectionProfile#LIGHT_PROFILE} when just connecting for ping
-     * and then disconnecting. If the node is already connected this method has no effect
+     * Connects to a node with the given connection profile. If the node is already connected this method has no effect
      */
     void connectToNode(DiscoveryNode node, ConnectionProfile connectionProfile) throws ConnectTransportException;
 

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

@@ -62,7 +62,6 @@
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
@@ -328,32 +327,6 @@ public Transport.Connection openConnection(final DiscoveryNode node, ConnectionP
         }
     }
 
-    /**
-     * 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
-     * @return the connected node
-     * @throws ConnectTransportException if the connection failed
-     * @throws IllegalStateException if the handshake failed
-     */
-    public DiscoveryNode connectToNodeAndHandshake(
-        final DiscoveryNode node,
-        final long handshakeTimeout) throws IOException {
-        if (node.equals(localNode)) {
-            return localNode;
-        }
-        DiscoveryNode handshakeNode;
-        try (Transport.Connection connection = transport.openConnection(node, ConnectionProfile.LIGHT_PROFILE)) {
-            handshakeNode = handshake(connection, handshakeTimeout);
-        }
-        connectToNode(node, ConnectionProfile.LIGHT_PROFILE);
-        return handshakeNode;
-    }
-
     /**
      * Executes a high-level handshake using the given connection
      * and returns the discovery node of the node the connection