Reject different version node join in mixed mode

Signed-off-by: Lakshya Taragi <lakshya.taragi@gmail.com>
opensearch-project · Apr 16, 2024 · 6723944 · 6723944
1 parent 695fbde
commit 6723944
Show file tree

Hide file tree

Showing 2 changed files with 83 additions and 6 deletions.
diff --git a/server/src/main/java/org/opensearch/cluster/coordination/JoinTaskExecutor.java b/server/src/main/java/org/opensearch/cluster/coordination/JoinTaskExecutor.java
@@ -57,6 +57,7 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
@@ -520,12 +521,24 @@ private static void ensureRemoteStoreNodesCompatibility(DiscoveryNode joiningNod
                     );
                 }
             }
-        } else {
-            if (remoteStoreCompatibilityMode == CompatibilityMode.MIXED) {
-                if (joiningNode.isRemoteStoreNode()) {
-                    Optional<DiscoveryNode> remoteDN = existingNodes.stream().filter(DiscoveryNode::isRemoteStoreNode).findFirst();
-                    remoteDN.ifPresent(discoveryNode -> ensureRemoteStoreNodesCompatibility(joiningNode, discoveryNode));
-                }
+        } else if (remoteStoreCompatibilityMode == CompatibilityMode.MIXED) {
+            Version joiningNodeVersion = joiningNode.getVersion();
+            if (joiningNodeVersion.after(currentNodes.getMaxNodeVersion()) || joiningNodeVersion.before(currentNodes.getMinNodeVersion())) {
+                boolean isAfter = joiningNodeVersion.after(currentNodes.getMaxNodeVersion());
+                String reason = String.format(
+                    Locale.ROOT,
+                    "mixed mode: a %s version [%s] node [%s] is not allowed to join cluster with %s version [%s]",
+                    isAfter ? "higher" : "lower",
+                    joiningNode.getVersion(),
+                    joiningNode,
+                    isAfter ? "maximum" : "minimum",
+                    isAfter ? currentNodes.getMaxNodeVersion() : currentNodes.getMinNodeVersion()
+                );
+                throw new IllegalStateException(reason);
+            }
+            if (joiningNode.isRemoteStoreNode()) {
+                Optional<DiscoveryNode> remoteDN = existingNodes.stream().filter(DiscoveryNode::isRemoteStoreNode).findFirst();
+                remoteDN.ifPresent(discoveryNode -> ensureRemoteStoreNodesCompatibility(joiningNode, discoveryNode));
             }
         }
     }

diff --git a/server/src/test/java/org/opensearch/cluster/coordination/JoinTaskExecutorTests.java b/server/src/test/java/org/opensearch/cluster/coordination/JoinTaskExecutorTests.java
@@ -76,6 +76,7 @@
 import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_TRANSLOG_REPOSITORY_NAME_ATTRIBUTE_KEY;
 import static org.opensearch.node.remotestore.RemoteStoreNodeService.MIGRATION_DIRECTION_SETTING;
 import static org.opensearch.node.remotestore.RemoteStoreNodeService.REMOTE_STORE_COMPATIBILITY_MODE_SETTING;
+import static org.opensearch.test.VersionUtils.allOpenSearchVersions;
 import static org.opensearch.test.VersionUtils.allVersions;
 import static org.opensearch.test.VersionUtils.maxCompatibleVersion;
 import static org.opensearch.test.VersionUtils.randomCompatibleVersion;
@@ -885,6 +886,69 @@ public void testUpdatesClusterStateWithMultiNodeClusterAndSameRepository() throw
         validateRepositoryMetadata(result.resultingState, clusterManagerNode, 2);
     }
 
+    public void testNodeJoinInMixedMode() {
+        Settings nodeSettings = Settings.builder().put(REMOTE_STORE_MIGRATION_EXPERIMENTAL, "true").build();
+        FeatureFlags.initializeFeatureFlags(nodeSettings);
+
+        List<Version> versions = allOpenSearchVersions();
+        assert versions.size() >= 3 : "test requires at least three open search versions";
+        Version lowerVersion = versions.get(versions.size() - 3);
+        Version baseVersion = versions.get(versions.size() - 2);
+        Version higherVersion = versions.get(versions.size() - 1);
+
+        DiscoveryNode currentNode1 = new DiscoveryNode(UUIDs.base64UUID(), buildNewFakeTransportAddress(), baseVersion);
+        DiscoveryNode currentNode2 = new DiscoveryNode(UUIDs.base64UUID(), buildNewFakeTransportAddress(), baseVersion);
+        DiscoveryNodes currentNodes = DiscoveryNodes.builder()
+            .add(currentNode1)
+            .localNodeId(currentNode1.getId())
+            .add(currentNode2)
+            .localNodeId(currentNode2.getId())
+            .build();
+
+        Settings mixedModeCompatibilitySettings = Settings.builder()
+            .put(REMOTE_STORE_COMPATIBILITY_MODE_SETTING.getKey(), RemoteStoreNodeService.CompatibilityMode.MIXED)
+            .build();
+
+        Metadata metadata = Metadata.builder().persistentSettings(mixedModeCompatibilitySettings).build();
+
+        boolean joiningNodeIsHigher = randomBoolean();
+
+        // joining node of a different version (higher or lower) than the current nodes
+        DiscoveryNode joiningNode1 = new DiscoveryNode(
+            randomAlphaOfLength(10),
+            randomAlphaOfLength(10),
+            buildNewFakeTransportAddress(),
+            remoteStoreNodeAttributes(SEGMENT_REPO, TRANSLOG_REPO),
+            Collections.singleton(DiscoveryNodeRole.CLUSTER_MANAGER_ROLE),
+            joiningNodeIsHigher ? higherVersion : lowerVersion
+        );
+        final IllegalStateException exception = expectThrows(
+            IllegalStateException.class,
+            () -> JoinTaskExecutor.ensureNodesCompatibility(joiningNode1, currentNodes, metadata)
+        );
+        String reason = String.format(
+            Locale.ROOT,
+            "mixed mode: a %s version [%s] node [%s] is not allowed to join cluster with %s version [%s]",
+            joiningNodeIsHigher ? "higher" : "lower",
+            joiningNode1.getVersion(),
+            joiningNode1,
+            joiningNodeIsHigher ? "maximum" : "minimum",
+            currentNodes.getMaxNodeVersion()
+        );
+        assertEquals(reason, exception.getMessage());
+
+        // joining node of the same version as the current nodes
+        DiscoveryNode joiningNode2 = new DiscoveryNode(
+            randomAlphaOfLength(10),
+            randomAlphaOfLength(10),
+            buildNewFakeTransportAddress(),
+            remoteStoreNodeAttributes(SEGMENT_REPO, TRANSLOG_REPO),
+            Collections.singleton(DiscoveryNodeRole.CLUSTER_MANAGER_ROLE),
+            baseVersion
+        );
+        JoinTaskExecutor.ensureNodesCompatibility(joiningNode2, currentNodes, metadata);
+    }
+
     private void validateRepositoryMetadata(ClusterState updatedState, DiscoveryNode existingNode, int expectedRepositories)
         throws Exception {