Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,8 @@

package org.opensearch.cluster.routing.allocation.decider;

import org.opensearch.action.admin.indices.settings.put.UpdateSettingsRequest;
import org.opensearch.action.support.clustermanager.AcknowledgedResponse;
import org.opensearch.cluster.ClusterState;
import org.opensearch.cluster.routing.IndexShardRoutingTable;
import org.opensearch.cluster.routing.ShardRouting;
Expand Down Expand Up @@ -99,6 +101,75 @@ public void testIndexPrimaryShardLimit() throws Exception {
});
}

public void testUpdatingIndexPrimaryShardLimit() throws Exception {
// Create first index with primary shard limit
Settings firstIndexSettings = Settings.builder()
.put(remoteStoreIndexSettings(0, 4)) // 4 shards, 0 replicas
.put(INDEX_TOTAL_PRIMARY_SHARDS_PER_NODE_SETTING.getKey(), 1)
.build();

// Create first index
createIndex("test1", firstIndexSettings);

// Update the index settings to set INDEX_TOTAL_PRIMARY_SHARDS_PER_NODE_SETTING
UpdateSettingsRequest updateSettingsRequest = new UpdateSettingsRequest("test1");
Settings updatedSettings = Settings.builder().put(INDEX_TOTAL_PRIMARY_SHARDS_PER_NODE_SETTING.getKey(), 1).build();
updateSettingsRequest.settings(updatedSettings);

AcknowledgedResponse response = client().admin().indices().updateSettings(updateSettingsRequest).actionGet();

assertTrue(response.isAcknowledged());

// Create second index
createIndex("test2", remoteStoreIndexSettings(0, 4));

assertBusy(() -> {
ClusterState state = client().admin().cluster().prepareState().get().getState();

// Check total number of shards (8 total: 4 from each index)
assertEquals("Total shards should be 8", 8, state.getRoutingTable().allShards().size());

// Count assigned and unassigned shards for test1
int test1AssignedShards = 0;
int test1UnassignedShards = 0;
Map<String, Integer> nodePrimaryCount = new HashMap<>();

// Check test1 shard distribution
for (IndexShardRoutingTable shardRouting : state.routingTable().index("test1")) {
for (ShardRouting shard : shardRouting) {
if (shard.assignedToNode()) {
test1AssignedShards++;
// Count primaries per node for test1
String nodeId = shard.currentNodeId();
nodePrimaryCount.merge(nodeId, 1, Integer::sum);
} else {
test1UnassignedShards++;
}
}
}

// Check test2 shard assignment
int test2UnassignedShards = 0;
for (IndexShardRoutingTable shardRouting : state.routingTable().index("test2")) {
for (ShardRouting shard : shardRouting) {
if (!shard.assignedToNode()) {
test2UnassignedShards++;
}
}
}

// Assertions
assertEquals("test1 should have 3 assigned shards", 3, test1AssignedShards);
assertEquals("test1 should have 1 unassigned shard", 1, test1UnassignedShards);
assertEquals("test2 should have no unassigned shards", 0, test2UnassignedShards);

// Verify no node has more than one primary shard of test1
for (Integer count : nodePrimaryCount.values()) {
assertTrue("No node should have more than 1 primary shard of test1", count <= 1);
}
});
}

public void testClusterPrimaryShardLimitss() throws Exception {
// Update cluster setting to limit primary shards per node
updateClusterSetting(CLUSTER_TOTAL_PRIMARY_SHARDS_PER_NODE_SETTING.getKey(), 1);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1847,7 +1847,8 @@ public static void validateRefreshIntervalSettings(Settings requestSettings, Clu
}

/**
* Validates {@code index.routing.allocation.total_primary_shards_per_node} is only set for remote store enabled cluster
* Validates the {@code index.routing.allocation.total_primary_shards_per_node} setting during index creation.
* Ensures this setting is only specified for remote store enabled clusters.
*/
// TODO : Update this check for SegRep to DocRep migration on need basis
public static void validateIndexTotalPrimaryShardsPerNodeSetting(Settings indexSettings) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@
import org.opensearch.cluster.ack.ClusterStateUpdateResponse;
import org.opensearch.cluster.block.ClusterBlock;
import org.opensearch.cluster.block.ClusterBlocks;
import org.opensearch.cluster.node.DiscoveryNode;
import org.opensearch.cluster.routing.RoutingTable;
import org.opensearch.cluster.routing.allocation.AllocationService;
import org.opensearch.cluster.routing.allocation.AwarenessReplicaBalance;
Expand Down Expand Up @@ -78,12 +79,12 @@
import static org.opensearch.action.support.ContextPreservingActionListener.wrapPreservingContext;
import static org.opensearch.cluster.metadata.IndexMetadata.SETTING_NUMBER_OF_SEARCH_REPLICAS;
import static org.opensearch.cluster.metadata.IndexMetadata.SETTING_REMOTE_STORE_ENABLED;
import static org.opensearch.cluster.metadata.MetadataCreateIndexService.validateIndexTotalPrimaryShardsPerNodeSetting;
import static org.opensearch.cluster.metadata.MetadataCreateIndexService.validateOverlap;
import static org.opensearch.cluster.metadata.MetadataCreateIndexService.validateRefreshIntervalSettings;
import static org.opensearch.cluster.metadata.MetadataCreateIndexService.validateTranslogDurabilitySettings;
import static org.opensearch.cluster.metadata.MetadataCreateIndexService.validateTranslogFlushIntervalSettingsForCompositeIndex;
import static org.opensearch.cluster.metadata.MetadataIndexTemplateService.findComponentTemplate;
import static org.opensearch.cluster.routing.allocation.decider.ShardsLimitAllocationDecider.INDEX_TOTAL_PRIMARY_SHARDS_PER_NODE_SETTING;
import static org.opensearch.common.settings.AbstractScopedSettings.ARCHIVED_SETTINGS_PREFIX;
import static org.opensearch.index.IndexSettings.same;

Expand Down Expand Up @@ -140,7 +141,7 @@ public void updateSettings(

validateRefreshIntervalSettings(normalizedSettings, clusterService.getClusterSettings());
validateTranslogDurabilitySettings(normalizedSettings, clusterService.getClusterSettings(), clusterService.getSettings());
validateIndexTotalPrimaryShardsPerNodeSetting(normalizedSettings);
validateIndexTotalPrimaryShardsPerNodeSetting(normalizedSettings, clusterService);
final int defaultReplicaCount = clusterService.getClusterSettings().get(Metadata.DEFAULT_REPLICA_COUNT_SETTING);

Settings.Builder settingsForClosedIndices = Settings.builder();
Expand Down Expand Up @@ -549,4 +550,31 @@ private void validateSearchReplicaCountSettings(Settings requestSettings, Index[
}
}
}

/**
* Validates the 'index.routing.allocation.total_primary_shards_per_node' setting during index settings update.
* Ensures this setting can only be modified for existing indices in remote store enabled clusters.
*/
public static void validateIndexTotalPrimaryShardsPerNodeSetting(Settings indexSettings, ClusterService clusterService) {
// Get the setting value
int indexPrimaryShardsPerNode = INDEX_TOTAL_PRIMARY_SHARDS_PER_NODE_SETTING.get(indexSettings);

// If default value (-1), no validation needed
if (indexPrimaryShardsPerNode == -1) {
return;
}

// Check if remote store is enabled
boolean isRemoteStoreEnabled = clusterService.state()
.nodes()
.getNodes()
.values()
.stream()
.allMatch(DiscoveryNode::isRemoteStoreNode);
if (!isRemoteStoreEnabled) {
throw new IllegalArgumentException(
"Setting [" + INDEX_TOTAL_PRIMARY_SHARDS_PER_NODE_SETTING.getKey() + "] can only be used with remote store enabled clusters"
);
}
}
}