Skip to content

Commit

Permalink
Reject different version node join in mixed mode
Browse files Browse the repository at this point in the history
Signed-off-by: Lakshya Taragi <[email protected]>
  • Loading branch information
ltaragi committed Apr 16, 2024
1 parent 695fbde commit 6723944
Show file tree
Hide file tree
Showing 2 changed files with 83 additions and 6 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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));
}
}
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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 {

Expand Down

0 comments on commit 6723944

Please sign in to comment.