Skip to content

Add remote to local pool change handling in TargetPoolAllocationDecider #18027

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
gbbafna merged 1 commit into from
Apr 24, 2025
Merged

Add remote to local pool change handling in TargetPoolAllocationDecider #18027

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 1 addition & 4 deletions ...nalClusterTest/java/org/opensearch/indices/replication/WarmIndexSegmentReplicationIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,10 +128,7 @@ private static String indexOrAlias() {

@Override
public Settings indexSettings() {
return Settings.builder()
.put(super.indexSettings())
.put(IndexModule.INDEX_STORE_LOCALITY_SETTING.getKey(), IndexModule.DataLocalityType.PARTIAL.name())
.build();
return Settings.builder().put(super.indexSettings()).put(IndexModule.IS_WARM_INDEX_SETTING.getKey(), true).build();
}

@Override
Expand Down
3 changes: 1 addition & 2 deletions server/src/main/java/org/opensearch/cluster/routing/OperationRouting.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -269,8 +269,7 @@ public GroupShardsIterator<ShardIterator> searchShards(
}

if (FeatureFlags.isEnabled(FeatureFlags.WRITABLE_WARM_INDEX_EXPERIMENTAL_FLAG)
&& IndexModule.DataLocalityType.PARTIAL.name()
.equals(indexMetadataForShard.getSettings().get(IndexModule.INDEX_STORE_LOCALITY_SETTING.getKey()))
&& indexMetadataForShard.getSettings().getAsBoolean(IndexModule.IS_WARM_INDEX_SETTING.getKey(), false)
&& (preference == null || preference.isEmpty())) {
preference = Preference.PRIMARY_FIRST.type();
}
Expand Down
23 changes: 21 additions & 2 deletions ...n/java/org/opensearch/cluster/routing/allocation/decider/TargetPoolAllocationDecider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,12 +91,13 @@ public Decision canForceAllocatePrimary(ShardRouting shardRouting, RoutingNode n
public Decision canRemain(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation) {
RoutingPool targetPool = RoutingPool.getShardPool(shardRouting, allocation);
RoutingPool currentNodePool = RoutingPool.getNodePool(allocation.routingNodes().node(shardRouting.currentNodeId()));

if (RoutingPool.REMOTE_CAPABLE.equals(targetPool) && targetPool != currentNodePool) {
logger.debug(
"Shard: [{}] has current pool: [{}], target pool: [{}]. Cannot remain on node: [{}]",
shardRouting.shortSummary(),
currentNodePool.name(),
RoutingPool.REMOTE_CAPABLE.name(),
targetPool.name(),
node.node()
);
return allocation.decision(
Expand All @@ -107,7 +108,25 @@ public Decision canRemain(ShardRouting shardRouting, RoutingNode node, RoutingAl
currentNodePool,
targetPool
);
}
} else if (RoutingPool.LOCAL_ONLY.equals(targetPool)
&& targetPool != currentNodePool
&& !node.node().getRoles().contains(DiscoveryNodeRole.DATA_ROLE)) {
logger.debug(
"Shard: [{}] has target pool: [{}]. Cannot remain on node: [{}] without the [{}] node role",
shardRouting,
targetPool,
node.node(),
DiscoveryNodeRole.DATA_ROLE.roleName()
);
return allocation.decision(
Decision.NO,
NAME,
"Routing pools are incompatible. Shard pool: [%s], node pool: [%s] without [%s] role",
targetPool,
currentNodePool,
DiscoveryNodeRole.DATA_ROLE.roleName()
);
}
return allocation.decision(
Decision.YES,
NAME,
Expand Down
2 changes: 1 addition & 1 deletion server/src/test/java/org/opensearch/cluster/routing/OperationRoutingTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1088,7 +1088,7 @@ public void testPartialIndexPrimaryDefault() throws Exception {
.settings(
Settings.builder()
.put(state.metadata().index(indexName).getSettings())
.put(IndexModule.INDEX_STORE_LOCALITY_SETTING.getKey(), IndexModule.DataLocalityType.PARTIAL)
.put(IndexModule.IS_WARM_INDEX_SETTING.getKey(), true)
.build()
)
.build();
Expand Down
43 changes: 31 additions & 12 deletions ...test/java/org/opensearch/cluster/routing/allocation/RemoteShardsBalancerBaseTestCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,10 +102,17 @@ public RoutingAllocation getRoutingAllocation(ClusterState clusterState, Routing
}

public ClusterState createInitialCluster(int localOnlyNodes, int remoteNodes, int localIndices, int remoteIndices) {
return createInitialCluster(localOnlyNodes, remoteNodes, false, localIndices, remoteIndices);
return createInitialCluster(localOnlyNodes, remoteNodes, false, localIndices, remoteIndices, false);
}

public ClusterState createInitialCluster(int localOnlyNodes, int remoteNodes, boolean remoteOnly, int localIndices, int remoteIndices) {
public ClusterState createInitialCluster(
int localOnlyNodes,
int remoteNodes,
boolean remoteRoleOnly,
int localIndices,
int remoteIndices,
boolean remoteIndexIsWarm
) {
Metadata.Builder mb = Metadata.builder();
for (int i = 0; i < localIndices; i++) {
mb.put(
Expand All @@ -117,15 +124,27 @@ public ClusterState createInitialCluster(int localOnlyNodes, int remoteNodes, bo
}

for (int i = 0; i < remoteIndices; i++) {
mb.put(
IndexMetadata.builder(getIndexName(i, true))
.settings(
settings(Version.CURRENT).put(UnassignedInfo.INDEX_DELAYED_NODE_LEFT_TIMEOUT_SETTING.getKey(), "0")
.put(IndexModule.INDEX_STORE_TYPE_SETTING.getKey(), IndexModule.Type.REMOTE_SNAPSHOT.getSettingsKey())
)
.numberOfShards(PRIMARIES)
.numberOfReplicas(REPLICAS)
);
if (remoteIndexIsWarm == false) {
mb.put(
IndexMetadata.builder(getIndexName(i, true))
.settings(
settings(Version.CURRENT).put(UnassignedInfo.INDEX_DELAYED_NODE_LEFT_TIMEOUT_SETTING.getKey(), "0")
.put(IndexModule.INDEX_STORE_TYPE_SETTING.getKey(), IndexModule.Type.REMOTE_SNAPSHOT.getSettingsKey())
)
.numberOfShards(PRIMARIES)
.numberOfReplicas(REPLICAS)
);
} else {
mb.put(
IndexMetadata.builder(getIndexName(i, true))
.settings(
settings(Version.CURRENT).put(UnassignedInfo.INDEX_DELAYED_NODE_LEFT_TIMEOUT_SETTING.getKey(), "0")
.put(IndexModule.IS_WARM_INDEX_SETTING.getKey(), true)
)
.numberOfShards(PRIMARIES)
.numberOfReplicas(REPLICAS)
);
}
}
Metadata metadata = mb.build();

Expand All @@ -143,7 +162,7 @@ public ClusterState createInitialCluster(int localOnlyNodes, int remoteNodes, bo
String name = getNodeId(i, false);
nb.add(newNode(name, name, MANAGER_DATA_ROLES));
}
if (remoteOnly) {
if (remoteRoleOnly) {
for (int i = 0; i < remoteNodes; i++) {
String name = getNodeId(i, true);
nb.add(newNode(name, name, WARM_ONLY_ROLE));
Expand Down
62 changes: 55 additions & 7 deletions ...src/test/java/org/opensearch/cluster/routing/allocation/ShardsTieringAllocationTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@
import static org.opensearch.cluster.routing.RoutingPool.REMOTE_CAPABLE;
import static org.opensearch.cluster.routing.RoutingPool.getIndexPool;
import static org.opensearch.common.util.FeatureFlags.WRITABLE_WARM_INDEX_EXPERIMENTAL_FLAG;
import static org.opensearch.index.IndexModule.INDEX_STORE_LOCALITY_SETTING;
import static org.opensearch.index.IndexModule.IS_WARM_INDEX_SETTING;

public class ShardsTieringAllocationTests extends TieringAllocationBaseTestCase {

Expand Down Expand Up @@ -81,8 +81,9 @@ public void testShardsWithTiering() {
clusterState = updateIndexMetadataForTiering(
clusterState,
localIndices,
remoteIndices,
IndexModule.TieringState.HOT_TO_WARM.name(),
IndexModule.DataLocalityType.PARTIAL.name()
true
);
// trigger shard relocation
clusterState = allocateShardsAndBalance(clusterState, service);
Expand All @@ -100,14 +101,61 @@ public void testShardsWithTiering() {
}
}

@LockFeatureFlag(WRITABLE_WARM_INDEX_EXPERIMENTAL_FLAG)
public void testShardsWithWarmToHotTiering() throws Exception {
int localOnlyNodes = 60;
int remoteCapableNodes = 13;
int localIndices = 0;
int remoteIndices = 10; // 5 primary, 1 replica

// Create a cluster with warm only roles (dedicated setup) and remote index is of type warm only.
ClusterState clusterState = createInitialCluster(localOnlyNodes, remoteCapableNodes, true, localIndices, remoteIndices, true);
AllocationService service = this.createRemoteCapableAllocationService();

// assign shards to respective nodes
clusterState = allocateShardsAndBalance(clusterState, service);
RoutingNodes routingNodes = clusterState.getRoutingNodes();
RoutingAllocation allocation = getRoutingAllocation(clusterState, routingNodes);
assertEquals(0, routingNodes.unassigned().size());

for (ShardRouting shard : clusterState.getRoutingTable().allShards()) {
assertTrue(shard.relocating() || shard.started());
RoutingPool shardPool = RoutingPool.getShardPool(shard, allocation);
RoutingNode node = routingNodes.node(shard.currentNodeId());
RoutingPool nodePool = RoutingPool.getNodePool(node);
assertEquals(REMOTE_CAPABLE, shardPool);
assertEquals(REMOTE_CAPABLE, nodePool);
}

// put indices in the hot to warm tiering state
clusterState = updateIndexMetadataForTiering(
clusterState,
localIndices,
remoteIndices,
IndexModule.TieringState.WARM_TO_HOT.name(),
false
);
// trigger shard relocation
clusterState = allocateShardsAndBalance(clusterState, service);
routingNodes = clusterState.getRoutingNodes();
allocation = getRoutingAllocation(clusterState, routingNodes);
assertEquals(0, routingNodes.unassigned().size());

for (ShardRouting shard : clusterState.getRoutingTable().allShards()) {
assertBusy(() -> { assertFalse(shard.unassigned()); });
RoutingNode node = routingNodes.node(shard.currentNodeId());
RoutingPool nodePool = RoutingPool.getNodePool(node);
RoutingPool shardPool = RoutingPool.getShardPool(shard, allocation);
assertEquals(LOCAL_ONLY, shardPool);
assertEquals(nodePool, shardPool);
}
}

@LockFeatureFlag(WRITABLE_WARM_INDEX_EXPERIMENTAL_FLAG)
public void testShardPoolForPartialIndices() {
String index = "test-index";
IndexMetadata indexMetadata = IndexMetadata.builder(index)
.settings(
settings(Version.CURRENT).put(INDEX_STORE_LOCALITY_SETTING.getKey(), IndexModule.DataLocalityType.PARTIAL.name())
.put(IndexModule.IS_WARM_INDEX_SETTING.getKey(), true)
)
.settings(settings(Version.CURRENT).put(IndexModule.IS_WARM_INDEX_SETTING.getKey(), true))
.numberOfShards(PRIMARIES)
.numberOfReplicas(REPLICAS)
.build();
Expand All @@ -119,7 +167,7 @@ public void testShardPoolForPartialIndices() {
public void testShardPoolForFullIndices() {
String index = "test-index";
IndexMetadata indexMetadata = IndexMetadata.builder(index)
.settings(settings(Version.CURRENT).put(INDEX_STORE_LOCALITY_SETTING.getKey(), IndexModule.DataLocalityType.FULL.name()))
.settings(settings(Version.CURRENT).put(IS_WARM_INDEX_SETTING.getKey(), false))
.numberOfShards(PRIMARIES)
.numberOfReplicas(REPLICAS)
.build();
Expand Down
21 changes: 17 additions & 4 deletions ...rc/test/java/org/opensearch/cluster/routing/allocation/TieringAllocationBaseTestCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,6 @@
import org.opensearch.common.settings.Settings;
import org.opensearch.index.IndexModule;

import static org.opensearch.index.IndexModule.INDEX_STORE_LOCALITY_SETTING;
import static org.opensearch.index.IndexModule.INDEX_TIERING_STATE;

@SuppressForbidden(reason = "feature flag overrides")
Expand All @@ -24,8 +23,9 @@ public abstract class TieringAllocationBaseTestCase extends RemoteShardsBalancer
public ClusterState updateIndexMetadataForTiering(
ClusterState clusterState,
int localIndices,
int remoteIndices,
String tieringState,
String dataLocality
boolean isWarmIndex
) {
Metadata.Builder mb = Metadata.builder(clusterState.metadata());
for (int i = 0; i < localIndices; i++) {
Expand All @@ -38,8 +38,21 @@ public ClusterState updateIndexMetadataForTiering(
.put(settings)
.put(settings)
.put(INDEX_TIERING_STATE.getKey(), tieringState)
.put(IndexModule.IS_WARM_INDEX_SETTING.getKey(), true)
.put(INDEX_STORE_LOCALITY_SETTING.getKey(), dataLocality)
.put(IndexModule.IS_WARM_INDEX_SETTING.getKey(), isWarmIndex)
)
);
}
for (int i = 0; i < remoteIndices; i++) {
IndexMetadata indexMetadata = clusterState.metadata().index(getIndexName(i, true));
Settings settings = indexMetadata.getSettings();
mb.put(
IndexMetadata.builder(indexMetadata)
.settings(
Settings.builder()
.put(settings)
.put(settings)
.put(INDEX_TIERING_STATE.getKey(), tieringState)
.put(IndexModule.IS_WARM_INDEX_SETTING.getKey(), isWarmIndex)
)
);
}
Expand Down
4 changes: 2 additions & 2 deletions ...a/org/opensearch/cluster/routing/allocation/decider/TargetPoolAllocationDeciderTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,7 +113,7 @@ public void testTargetPoolHybridAllocationDecisions() {
}

public void testTargetPoolDedicatedSearchNodeAllocationDecisions() {
ClusterState clusterState = createInitialCluster(3, 3, true, 2, 2);
ClusterState clusterState = createInitialCluster(3, 3, true, 2, 2, false);
AllocationService service = this.createRemoteCapableAllocationService();
clusterState = allocateShardsAndBalance(clusterState, service);

Expand Down Expand Up @@ -202,7 +202,7 @@ public void testTargetPoolDedicatedSearchNodeAllocationDecisions() {
}

public void testDebugMessage() {
ClusterState clusterState = createInitialCluster(3, 3, true, 2, 2);
ClusterState clusterState = createInitialCluster(3, 3, true, 2, 2, false);
AllocationService service = this.createRemoteCapableAllocationService();
clusterState = allocateShardsAndBalance(clusterState, service);

Expand Down
Loading