IGNITE-28751 Refactored TCP Discovery SPI joining node validation (#13217)

Author: petrov-mg

modules/core/src/main/java/org/apache/ignite/events/DiscoveryEvent.java

@@ -62,6 +62,7 @@
  * @see EventType#EVT_NODE_JOINED
  * @see EventType#EVT_NODE_LEFT
  * @see EventType#EVT_NODE_SEGMENTED
+ * @see EventType#EVT_NODE_VALIDATION_FAILED
  * @see EventType#EVTS_DISCOVERY_ALL
  * @see EventType#EVTS_DISCOVERY
  */

modules/core/src/main/java/org/apache/ignite/events/EventType.java

@@ -1102,6 +1102,7 @@ public interface EventType {
         EVT_NODE_JOINED,
         EVT_NODE_LEFT,
         EVT_NODE_FAILED,
+        EVT_NODE_VALIDATION_FAILED,
         EVT_NODE_SEGMENTED,
         EVT_CLIENT_NODE_DISCONNECTED,
         EVT_CLIENT_NODE_RECONNECTED
@@ -1119,6 +1120,7 @@ public interface EventType {
         EVT_NODE_LEFT,
         EVT_NODE_FAILED,
         EVT_NODE_SEGMENTED,
+        EVT_NODE_VALIDATION_FAILED,
         EVT_NODE_METRICS_UPDATED,
         EVT_CLIENT_NODE_DISCONNECTED,
         EVT_CLIENT_NODE_RECONNECTED

modules/core/src/main/java/org/apache/ignite/events/NodeValidationFailedEvent.java

@@ -31,13 +31,10 @@
  * @see EventType#EVT_NODE_VALIDATION_FAILED
  * @see GridComponent#validateNode
  */
-public class NodeValidationFailedEvent extends EventAdapter {
+public class NodeValidationFailedEvent extends DiscoveryEvent {
     /** */
     private static final long serialVersionUID = 0L;
 
-    /** The node that attempted to join cluster. */
-    private final ClusterNode evtNode;
-
     /** Validation result. */
     private final IgniteNodeValidationResult res;
 
@@ -49,17 +46,11 @@ public class NodeValidationFailedEvent extends EventAdapter {
      * @param res Joining node validation result.
      */
     public NodeValidationFailedEvent(ClusterNode node, ClusterNode evtNode, IgniteNodeValidationResult res) {
-        super(node, res.message(), EVT_NODE_VALIDATION_FAILED);
+        super(node, res.message(), EVT_NODE_VALIDATION_FAILED, evtNode);
 
-        this.evtNode = evtNode;
         this.res = res;
     }
 
-    /** @return Node that couldn't join the topology due to a validation failure. */
-    public ClusterNode eventNode() {
-        return evtNode;
-    }
-
     /** @return Joining node validation result. */
     public IgniteNodeValidationResult validationResult() {
         return res;

modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java

@@ -1895,6 +1895,16 @@ public IgniteInternalFuture<Long> topologyFuture(final long awaitVer) {
         return fut;
     }
 
+    /**
+     * Returns a collection of all remote nodes known to the underlying {@link DiscoverySpi} implementation.
+     *
+     * <p>Unlike {@link #remoteNodes()}, this method may include nodes that have successfully completed
+     * validation but have not yet completed their join routine.</p>
+     */
+    public Collection<ClusterNode> discoverySpiRemoteNodes() {
+        return getSpi().getRemoteNodes();
+    }
+
     /**
      * Gets discovery collection cache from SPI safely guarding against "floating" collections.
      *

modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeProcessor.java

@@ -18,15 +18,15 @@
 package org.apache.ignite.internal.processors.rollingupgrade;
 
 import java.util.Objects;
+import java.util.SortedSet;
+import java.util.TreeSet;
 import java.util.UUID;
 import java.util.concurrent.CountDownLatch;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.cluster.ClusterNode;
 import org.apache.ignite.events.DiscoveryEvent;
-import org.apache.ignite.events.Event;
 import org.apache.ignite.internal.GridKernalContext;
-import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
 import org.apache.ignite.internal.processors.GridProcessorAdapter;
 import org.apache.ignite.internal.processors.metastorage.DistributedMetaStorage;
 import org.apache.ignite.internal.processors.metastorage.DistributedMetastorageLifecycleListener;
@@ -39,14 +39,13 @@
 import org.apache.ignite.lang.IgniteProductVersion;
 import org.apache.ignite.plugin.security.SecurityPermission;
 import org.apache.ignite.spi.IgniteNodeValidationResult;
-import org.apache.ignite.spi.discovery.DiscoverySpi;
 import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
-import org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoveryNodesRing;
 import org.jetbrains.annotations.Nullable;
 
 import static org.apache.ignite.events.EventType.EVT_NODE_FAILED;
 import static org.apache.ignite.events.EventType.EVT_NODE_JOINED;
 import static org.apache.ignite.events.EventType.EVT_NODE_LEFT;
+import static org.apache.ignite.events.EventType.EVT_NODE_VALIDATION_FAILED;
 import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_BUILD_VER;
 import static org.apache.ignite.internal.processors.metastorage.DistributedMetaStorage.IGNITE_INTERNAL_KEY_PREFIX;
 
@@ -58,15 +57,9 @@ public class RollingUpgradeProcessor extends GridProcessorAdapter implements Dis
     /** Metastorage with the write access. */
     @Nullable private volatile DistributedMetaStorage metastorage;
 
-    /** TCP discovery nodes ring. */
-    private TcpDiscoveryNodesRing ring;
-
     /** Last joining node. */
     private ClusterNode lastJoiningNode;
 
-    /** Last joining node timestamp. */
-    private long lastJoiningNodeTimestamp;
-
     /** Lock for synchronization between tcp-disco-msg-worker thread and management operations. */
     private final Object lock = new Object();
 
@@ -85,26 +78,25 @@ public RollingUpgradeProcessor(GridKernalContext ctx) {
 
     /** {@inheritDoc} */
     @Override public void onKernalStart(boolean active) throws IgniteCheckedException {
-        DiscoverySpi spi = ctx.config().getDiscoverySpi();
-
-        if (spi instanceof TcpDiscoverySpi)
-            ring = ((TcpDiscoverySpi)spi).discoveryRing();
-
         startLatch.countDown();
     }
 
     /** {@inheritDoc} */
     @Override public void start() throws IgniteCheckedException {
-        ctx.event().addLocalEventListener(new GridLocalEventListener() {
-            @Override public void onEvent(Event evt) {
+        ctx.event().addLocalEventListener(
+            evt -> {
                 UUID nodeId = ((DiscoveryEvent)evt).eventNode().id();
 
                 synchronized (lock) {
                     if (lastJoiningNode != null && lastJoiningNode.id().equals(nodeId))
                         lastJoiningNode = null;
                 }
-            }
-        }, EVT_NODE_JOINED, EVT_NODE_FAILED, EVT_NODE_LEFT);
+            },
+            EVT_NODE_JOINED,
+            EVT_NODE_FAILED,
+            EVT_NODE_LEFT,
+            EVT_NODE_VALIDATION_FAILED
+        );
 
         ctx.internalSubscriptionProcessor().registerDistributedMetastorageListener(new DistributedMetastorageLifecycleListener() {
             @Override public void onReadyForWrite(DistributedMetaStorage metastorage) {
@@ -131,8 +123,6 @@ public RollingUpgradeProcessor(GridKernalContext ctx) {
     @Override public @Nullable IgniteNodeValidationResult validateNode(ClusterNode node) {
         synchronized (lock) {
             lastJoiningNode = node;
-
-            lastJoiningNodeTimestamp = U.currentTimeMillis();
         }
 
         ClusterNode locNode = ctx.discovery().localNode();
@@ -247,21 +237,14 @@ public void disable() throws IgniteCheckedException {
         if (rollUpVers == null)
             return;
 
-        IgnitePair<IgniteProductVersion> minMaxVerPair = ring.minMaxNodeVersions();
-
-        if (!minMaxVerPair.get1().equals(minMaxVerPair.get2()))
-            throw new IgniteCheckedException("Can't disable rolling upgrade with different versions in cluster: "
-                + minMaxVerPair.get1() + ", " + minMaxVerPair.get2());
+        IgnitePair<IgniteProductVersion> minMaxVerPair;
 
         synchronized (lock) {
-            if (lastJoiningNode != null) {
-                // Use 3 * joinTimeout as an upper time bound for joining nodes that may drop during validation
-                // without sending NODE_LEFT / NODE_FAILED events.
-                long timeout = ((TcpDiscoverySpi)ctx.config().getDiscoverySpi()).getJoinTimeout() * 3;
+            minMaxVerPair = resolveMinMaxNodeVersions();
 
-                if (ring.node(lastJoiningNode.id()) != null || (timeout > 0 && U.currentTimeMillis() - lastJoiningNodeTimestamp > timeout))
-                    lastJoiningNode = null;
-            }
+            if (!minMaxVerPair.get1().equals(minMaxVerPair.get2()))
+                throw new IgniteCheckedException("Can't disable rolling upgrade with different versions in cluster: "
+                    + minMaxVerPair.get1() + ", " + minMaxVerPair.get2());
 
             if (lastJoiningNode != null) {
                 IgniteProductVersion lastJoiningNodeVer = IgniteProductVersion.fromString(lastJoiningNode.attribute(ATTR_BUILD_VER));
@@ -280,6 +263,20 @@ public void disable() throws IgniteCheckedException {
             log.info("Rolling upgrade disabled. Current version of nodes in cluster: " + minMaxVerPair.get1());
     }
 
+    /** */
+    private IgnitePair<IgniteProductVersion> resolveMinMaxNodeVersions() {
+        assert Thread.holdsLock(lock);
+
+        SortedSet<IgniteProductVersion> clusterNodes = new TreeSet<>();
+
+        for (ClusterNode node : ctx.discovery().discoverySpiRemoteNodes())
+            clusterNodes.add(node.version());
+
+        clusterNodes.add(ctx.discovery().localNode().version());
+
+        return new IgnitePair<>(clusterNodes.first(), clusterNodes.last());
+    }
+
     /**
      * Returns a pair containing the current and target versions of the cluster.
      * <p>

modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java

@@ -251,7 +251,7 @@ class ClientImpl extends TcpDiscoveryImpl {
 
     /** {@inheritDoc} */
     @Override public void dumpRingStructure(IgniteLogger log) {
-        ClusterNode[] serverNodes = getRemoteNodes().stream()
+        ClusterNode[] serverNodes = remoteVisibleNodes().stream()
                 .filter(node -> !node.isClient())
                 .sorted(Comparator.comparingLong(ClusterNode::order))
                 .toArray(ClusterNode[]::new);
@@ -371,7 +371,7 @@ class ClientImpl extends TcpDiscoveryImpl {
 
     /** {@inheritDoc} */
     @Override public Collection<ClusterNode> getRemoteNodes() {
-        return U.arrayList(rmtNodes.values(), TcpDiscoveryNodesRing.VISIBLE_NODES);
+        return U.arrayList(rmtNodes.values());
     }
 
     /** {@inheritDoc} */
@@ -465,7 +465,7 @@ else if (state == DISCONNECTED) {
 
         spi.getSpiContext().deregisterPorts();
 
-        Collection<ClusterNode> rmts = getRemoteNodes();
+        Collection<ClusterNode> rmts = remoteVisibleNodes();
 
         // This is restart/disconnection and remote nodes are not empty.
         // We need to fire FAIL event for each.
@@ -1077,6 +1077,11 @@ private WorkersRegistry getWorkersRegistry() {
         return ignite instanceof IgniteEx ? ((IgniteEx)ignite).context().workersRegistry() : null;
     }
 
+    /** */
+    private Collection<ClusterNode> remoteVisibleNodes() {
+        return U.arrayList(rmtNodes.values(), TcpDiscoveryNodesRing.VISIBLE_NODES);
+    }
+
     /**
      * Metrics sender.
      */
@@ -2277,7 +2282,7 @@ private void processNodeAddFinishedMessage(TcpDiscoveryNodeAddFinishedMessage ms
                 return;
 
             if (log.isInfoEnabled()) {
-                for (ClusterNode node : getRemoteNodes()) {
+                for (ClusterNode node : remoteVisibleNodes()) {
                     if (node.id().equals(locNode.clientRouterNodeId())) {
                         if (log.isInfoEnabled())
                             log.info("Router node: " + node);