Skip to content

Commit

Permalink
Prevent version upgrade during remote migration (#13185) (#13347)
Browse files Browse the repository at this point in the history 
Signed-off-by: Lakshya Taragi <[email protected]>
(cherry picked from commit 2aad499)
  • Loading branch information
@ltaragi
ltaragi authored Apr 23, 2024
1 parent b4df35c commit cce8f63
Show file tree
Hide file tree
Showing 2 changed files with 76 additions and 3 deletions.
20 changes: 17 additions & 3 deletions server/src/main/java/org/opensearch/cluster/coordination/JoinTaskExecutor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@

package org.opensearch.cluster.coordination;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.opensearch.LegacyESVersion;
import org.opensearch.Version;
Expand Down Expand Up @@ -59,6 +60,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 All @@ -68,6 +70,7 @@
import static org.opensearch.cluster.decommission.DecommissionHelper.nodeCommissioned;
import static org.opensearch.gateway.GatewayService.STATE_NOT_RECOVERED_BLOCK;
import static org.opensearch.node.remotestore.RemoteStoreNodeService.CompatibilityMode;
import static org.opensearch.node.remotestore.RemoteStoreNodeService.CompatibilityMode.MIXED;
import static org.opensearch.node.remotestore.RemoteStoreNodeService.CompatibilityMode.STRICT;
import static org.opensearch.node.remotestore.RemoteStoreNodeService.REMOTE_STORE_COMPATIBILITY_MODE_SETTING;

Expand All @@ -80,7 +83,7 @@ public class JoinTaskExecutor implements ClusterStateTaskExecutor<JoinTaskExecut

private final AllocationService allocationService;

private final Logger logger;
private static Logger logger = LogManager.getLogger(JoinTaskExecutor.class);
private final RerouteService rerouteService;
private final TransportService transportService;

Expand Down Expand Up @@ -146,7 +149,7 @@ public JoinTaskExecutor(
RemoteStoreNodeService remoteStoreNodeService
) {
this.allocationService = allocationService;
this.logger = logger;
JoinTaskExecutor.logger = logger;
this.rerouteService = rerouteService;
this.transportService = transportService;
this.remoteStoreNodeService = remoteStoreNodeService;
Expand Down Expand Up @@ -582,7 +585,18 @@ private static void ensureRemoteStoreNodesCompatibility(DiscoveryNode joiningNod
}
}
} else {
if (remoteStoreCompatibilityMode == CompatibilityMode.MIXED) {
if (MIXED.equals(remoteStoreCompatibilityMode)) {
if (joiningNode.getVersion().after(currentNodes.getMaxNodeVersion())) {
String reason = String.format(
Locale.ROOT,
"remote migration : a node [%s] of higher version [%s] is not allowed to join a cluster with maximum version [%s]",
joiningNode,
joiningNode.getVersion(),
currentNodes.getMaxNodeVersion()
);
logger.warn(reason);
throw new IllegalStateException(reason);
}
if (joiningNode.isRemoteStoreNode()) {
Optional<DiscoveryNode> remoteDN = existingNodes.stream().filter(DiscoveryNode::isRemoteStoreNode).findFirst();
remoteDN.ifPresent(discoveryNode -> ensureRemoteStoreNodesCompatibility(joiningNode, discoveryNode));
Expand Down
59 changes: 59 additions & 0 deletions server/src/test/java/org/opensearch/cluster/coordination/JoinTaskExecutorTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,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.maxCompatibleVersion;
import static org.opensearch.test.VersionUtils.randomCompatibleVersion;
import static org.opensearch.test.VersionUtils.randomVersion;
Expand Down Expand Up @@ -967,6 +968,64 @@ 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() >= 2 : "test requires at least two open search versions";
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();

// joining node of a higher version than the current nodes
DiscoveryNode joiningNode1 = new DiscoveryNode(
randomAlphaOfLength(10),
randomAlphaOfLength(10),
buildNewFakeTransportAddress(),
remoteStoreNodeAttributes(SEGMENT_REPO, TRANSLOG_REPO),
Collections.singleton(DiscoveryNodeRole.CLUSTER_MANAGER_ROLE),
higherVersion
);
final IllegalStateException exception = expectThrows(
IllegalStateException.class,
() -> JoinTaskExecutor.ensureNodesCompatibility(joiningNode1, currentNodes, metadata)
);
String reason = String.format(
Locale.ROOT,
"remote migration : a node [%s] of higher version [%s] is not allowed to join a cluster with maximum version [%s]",
joiningNode1,
joiningNode1.getVersion(),
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 cce8f63

Please sign in to comment.