Skip to content

Commit

Permalink
Avoid passing in the cluster state parameter to methods related to in…
Browse files Browse the repository at this point in the history 
…dex creation (#112466)

Avoids the ClusterState type as a method parameter in favor of the more specific
Metadata, RoutingTable, and ClusterBlocks ones.
  • Loading branch information
@albertzaharovits
albertzaharovits authored Sep 4, 2024
1 parent d946d0c commit 0aff606
Show file tree
Hide file tree
Showing 13 changed files with 235 additions and 137 deletions.
3 changes: 2 additions & 1 deletion server/src/main/java/org/elasticsearch/action/admin/indices/rollover/LazyRolloverAction.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import org.elasticsearch.cluster.metadata.DataStream;
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
import org.elasticsearch.cluster.metadata.Metadata;
import org.elasticsearch.cluster.metadata.MetadataCreateIndexService;
import org.elasticsearch.cluster.metadata.MetadataDataStreamsService;
import org.elasticsearch.cluster.routing.allocation.AllocationService;
import org.elasticsearch.cluster.routing.allocation.allocator.AllocationActionMultiListener;
Expand Down Expand Up @@ -134,7 +135,7 @@ protected void masterOperation(
);
final String trialSourceIndexName = trialRolloverNames.sourceName();
final String trialRolloverIndexName = trialRolloverNames.rolloverName();
MetadataRolloverService.validateIndexName(clusterState, trialRolloverIndexName);
MetadataCreateIndexService.validateIndexName(trialRolloverIndexName, clusterState.metadata(), clusterState.routingTable());

assert metadata.dataStreams().containsKey(rolloverRequest.getRolloverTarget()) : "Auto-rollover applies only to data streams";

Expand Down
10 changes: 4 additions & 6 deletions ...rc/main/java/org/elasticsearch/action/admin/indices/rollover/MetadataRolloverService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -179,10 +179,6 @@ public RolloverResult rolloverClusterState(
};
}

public static void validateIndexName(ClusterState state, String index) {
MetadataCreateIndexService.validateIndexName(index, state);
}

/**
* Returns the names that rollover would use, but does not perform the actual rollover
*/
Expand Down Expand Up @@ -252,7 +248,8 @@ private RolloverResult rolloverAlias(
final Boolean isHidden = IndexMetadata.INDEX_HIDDEN_SETTING.exists(createIndexRequest.settings())
? IndexMetadata.INDEX_HIDDEN_SETTING.get(createIndexRequest.settings())
: null;
MetadataCreateIndexService.validateIndexName(rolloverIndexName, currentState); // fails if the index already exists
MetadataCreateIndexService.validateIndexName(rolloverIndexName, metadata, currentState.routingTable()); // fails if the index
// already exists
checkNoDuplicatedAliasInIndexTemplate(metadata, rolloverIndexName, aliasName, isHidden);
if (onlyValidate) {
return new RolloverResult(rolloverIndexName, sourceIndexName, currentState);
Expand Down Expand Up @@ -328,7 +325,8 @@ private RolloverResult rolloverDataStream(
final Tuple<String, Long> nextIndexAndGeneration = dataStream.nextWriteIndexAndGeneration(metadata, dataStreamIndices);
final String newWriteIndexName = nextIndexAndGeneration.v1();
final long newGeneration = nextIndexAndGeneration.v2();
MetadataCreateIndexService.validateIndexName(newWriteIndexName, currentState); // fails if the index already exists
MetadataCreateIndexService.validateIndexName(newWriteIndexName, metadata, currentState.routingTable()); // fails if the index
// already exists
if (onlyValidate) {
return new RolloverResult(newWriteIndexName, isLazyCreation ? NON_EXISTENT_SOURCE : originalWriteIndex.getName(), currentState);
}
Expand Down
3 changes: 2 additions & 1 deletion ...rc/main/java/org/elasticsearch/action/admin/indices/rollover/TransportRolloverAction.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
import org.elasticsearch.cluster.metadata.IndexMetadataStats;
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
import org.elasticsearch.cluster.metadata.Metadata;
import org.elasticsearch.cluster.metadata.MetadataCreateIndexService;
import org.elasticsearch.cluster.metadata.MetadataDataStreamsService;
import org.elasticsearch.cluster.routing.allocation.AllocationService;
import org.elasticsearch.cluster.routing.allocation.allocator.AllocationActionMultiListener;
Expand Down Expand Up @@ -179,7 +180,7 @@ protected void masterOperation(
);
final String trialSourceIndexName = trialRolloverNames.sourceName();
final String trialRolloverIndexName = trialRolloverNames.rolloverName();
MetadataRolloverService.validateIndexName(clusterState, trialRolloverIndexName);
MetadataCreateIndexService.validateIndexName(trialRolloverIndexName, metadata, clusterState.routingTable());

boolean isDataStream = metadata.dataStreams().containsKey(rolloverRequest.getRolloverTarget());
if (rolloverRequest.isLazy()) {
Expand Down
89 changes: 60 additions & 29 deletions server/src/main/java/org/elasticsearch/cluster/metadata/MetadataCreateIndexService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -167,21 +167,21 @@ public MetadataCreateIndexService(
/**
* Validate the name for an index against some static rules and a cluster state.
*/
public static void validateIndexName(String index, ClusterState state) {
public static void validateIndexName(String index, Metadata metadata, RoutingTable routingTable) {
validateIndexOrAliasName(index, InvalidIndexNameException::new);
if (index.toLowerCase(Locale.ROOT).equals(index) == false) {
throw new InvalidIndexNameException(index, "must be lowercase");
}

// NOTE: dot-prefixed index names are validated after template application, not here

if (state.routingTable().hasIndex(index)) {
throw new ResourceAlreadyExistsException(state.routingTable().index(index).getIndex());
if (routingTable.hasIndex(index)) {
throw new ResourceAlreadyExistsException(routingTable.index(index).getIndex());
}
if (state.metadata().hasIndex(index)) {
throw new ResourceAlreadyExistsException(state.metadata().index(index).getIndex());
if (metadata.hasIndex(index)) {
throw new ResourceAlreadyExistsException(metadata.index(index).getIndex());
}
if (state.metadata().hasAlias(index)) {
if (metadata.hasAlias(index)) {
throw new InvalidIndexNameException(index, "already exists as alias");
}
}
Expand Down Expand Up @@ -344,7 +344,7 @@ public ClusterState applyCreateIndexRequest(
normalizeRequestSetting(request);
logger.trace("executing IndexCreationTask for [{}] against cluster state version [{}]", request, currentState.version());

validate(request, currentState);
validate(request, currentState.metadata(), currentState.routingTable());

final Index recoverFromIndex = request.recoverFrom();
final IndexMetadata sourceMetadata = recoverFromIndex == null ? null : currentState.metadata().getIndexSafe(recoverFromIndex);
Expand Down Expand Up @@ -1069,7 +1069,9 @@ static Settings aggregateIndexSettings(
if (sourceMetadata != null) {
assert request.resizeType() != null;
prepareResizeIndexSettings(
currentState,
currentState.metadata(),
currentState.blocks(),
currentState.routingTable(),
indexSettingsBuilder,
request.recoverFrom(),
request.index(),
Expand All @@ -1084,7 +1086,7 @@ static Settings aggregateIndexSettings(
* We can not validate settings until we have applied templates, otherwise we do not know the actual settings
* that will be used to create this index.
*/
shardLimitValidator.validateShardLimit(indexSettings, currentState);
shardLimitValidator.validateShardLimit(indexSettings, currentState.nodes(), currentState.metadata());
validateSoftDeleteSettings(indexSettings);
validateTranslogRetentionSettings(indexSettings);
validateStoreTypeSetting(indexSettings);
Expand Down Expand Up @@ -1363,8 +1365,8 @@ private static void validateActiveShardCount(ActiveShardCount waitForActiveShard
}
}

private void validate(CreateIndexClusterStateUpdateRequest request, ClusterState state) {
validateIndexName(request.index(), state);
private void validate(CreateIndexClusterStateUpdateRequest request, Metadata metadata, RoutingTable routingTable) {
validateIndexName(request.index(), metadata, routingTable);
validateIndexSettings(request.index(), request.settings(), forbidPrivateIndexSettings);
}

Expand Down Expand Up @@ -1428,8 +1430,15 @@ private static List<String> validateIndexCustomPath(Settings settings, @Nullable
*
* @return the list of nodes at least one instance of the source index shards are allocated
*/
static List<String> validateShrinkIndex(ClusterState state, String sourceIndex, String targetIndexName, Settings targetIndexSettings) {
IndexMetadata sourceMetadata = validateResize(state, sourceIndex, targetIndexName, targetIndexSettings);
static List<String> validateShrinkIndex(
Metadata metadata,
ClusterBlocks clusterBlocks,
RoutingTable routingTable,
String sourceIndex,
String targetIndexName,
Settings targetIndexSettings
) {
IndexMetadata sourceMetadata = validateResize(metadata, clusterBlocks, sourceIndex, targetIndexName, targetIndexSettings);
if (sourceMetadata.isSearchableSnapshot()) {
throw new IllegalArgumentException("can't shrink searchable snapshot index [" + sourceIndex + ']');
}
Expand All @@ -1441,7 +1450,7 @@ static List<String> validateShrinkIndex(ClusterState state, String sourceIndex,
}

// now check that index is all on one node
final IndexRoutingTable table = state.routingTable().index(sourceIndex);
final IndexRoutingTable table = routingTable.index(sourceIndex);
Map<String, AtomicInteger> nodesToNumRouting = new HashMap<>();
int numShards = sourceMetadata.getNumberOfShards();
for (ShardRouting routing : table.shardsWithState(ShardRoutingState.STARTED)) {
Expand All @@ -1461,16 +1470,28 @@ static List<String> validateShrinkIndex(ClusterState state, String sourceIndex,
return nodesToAllocateOn;
}

static void validateSplitIndex(ClusterState state, String sourceIndex, String targetIndexName, Settings targetIndexSettings) {
IndexMetadata sourceMetadata = validateResize(state, sourceIndex, targetIndexName, targetIndexSettings);
static void validateSplitIndex(
Metadata metadata,
ClusterBlocks clusterBlocks,
String sourceIndex,
String targetIndexName,
Settings targetIndexSettings
) {
IndexMetadata sourceMetadata = validateResize(metadata, clusterBlocks, sourceIndex, targetIndexName, targetIndexSettings);
if (sourceMetadata.isSearchableSnapshot()) {
throw new IllegalArgumentException("can't split searchable snapshot index [" + sourceIndex + ']');
}
IndexMetadata.selectSplitShard(0, sourceMetadata, INDEX_NUMBER_OF_SHARDS_SETTING.get(targetIndexSettings));
}

static void validateCloneIndex(ClusterState state, String sourceIndex, String targetIndexName, Settings targetIndexSettings) {
IndexMetadata sourceMetadata = validateResize(state, sourceIndex, targetIndexName, targetIndexSettings);
static void validateCloneIndex(
Metadata metadata,
ClusterBlocks clusterBlocks,
String sourceIndex,
String targetIndexName,
Settings targetIndexSettings
) {
IndexMetadata sourceMetadata = validateResize(metadata, clusterBlocks, sourceIndex, targetIndexName, targetIndexSettings);
if (sourceMetadata.isSearchableSnapshot()) {
for (Setting<?> nonCloneableSetting : Arrays.asList(INDEX_STORE_TYPE_SETTING, INDEX_RECOVERY_TYPE_SETTING)) {
if (nonCloneableSetting.exists(targetIndexSettings) == false) {
Expand All @@ -1487,16 +1508,22 @@ static void validateCloneIndex(ClusterState state, String sourceIndex, String ta
IndexMetadata.selectCloneShard(0, sourceMetadata, INDEX_NUMBER_OF_SHARDS_SETTING.get(targetIndexSettings));
}

static IndexMetadata validateResize(ClusterState state, String sourceIndex, String targetIndexName, Settings targetIndexSettings) {
if (state.metadata().hasIndex(targetIndexName)) {
throw new ResourceAlreadyExistsException(state.metadata().index(targetIndexName).getIndex());
static IndexMetadata validateResize(
Metadata metadata,
ClusterBlocks clusterBlocks,
String sourceIndex,
String targetIndexName,
Settings targetIndexSettings
) {
if (metadata.hasIndex(targetIndexName)) {
throw new ResourceAlreadyExistsException(metadata.index(targetIndexName).getIndex());
}
final IndexMetadata sourceMetadata = state.metadata().index(sourceIndex);
final IndexMetadata sourceMetadata = metadata.index(sourceIndex);
if (sourceMetadata == null) {
throw new IndexNotFoundException(sourceIndex);
}

IndexAbstraction source = state.metadata().getIndicesLookup().get(sourceIndex);
IndexAbstraction source = metadata.getIndicesLookup().get(sourceIndex);
assert source != null;
if (source.getParentDataStream() != null && source.getParentDataStream().getWriteIndex().equals(sourceMetadata.getIndex())) {
throw new IllegalArgumentException(
Expand All @@ -1509,7 +1536,7 @@ static IndexMetadata validateResize(ClusterState state, String sourceIndex, Stri
);
}
// ensure index is read-only
if (state.blocks().indexBlocked(ClusterBlockLevel.WRITE, sourceIndex) == false) {
if (clusterBlocks.indexBlocked(ClusterBlockLevel.WRITE, sourceIndex) == false) {
throw new IllegalStateException("index " + sourceIndex + " must be read-only to resize index. use \"index.blocks.write=true\"");
}

Expand All @@ -1522,28 +1549,32 @@ static IndexMetadata validateResize(ClusterState state, String sourceIndex, Stri
}

static void prepareResizeIndexSettings(
final ClusterState currentState,
final Metadata metadata,
final ClusterBlocks clusterBlocks,
final RoutingTable routingTable,
final Settings.Builder indexSettingsBuilder,
final Index resizeSourceIndex,
final String resizeIntoName,
final ResizeType type,
final boolean copySettings,
final IndexScopedSettings indexScopedSettings
) {
final IndexMetadata sourceMetadata = currentState.metadata().index(resizeSourceIndex.getName());
final IndexMetadata sourceMetadata = metadata.index(resizeSourceIndex.getName());
if (type == ResizeType.SHRINK) {
final List<String> nodesToAllocateOn = validateShrinkIndex(
currentState,
metadata,
clusterBlocks,
routingTable,
resizeSourceIndex.getName(),
resizeIntoName,
indexSettingsBuilder.build()
);
indexSettingsBuilder.put(INDEX_SHRINK_INITIAL_RECOVERY_KEY, Strings.arrayToCommaDelimitedString(nodesToAllocateOn.toArray()));
} else if (type == ResizeType.SPLIT) {
validateSplitIndex(currentState, resizeSourceIndex.getName(), resizeIntoName, indexSettingsBuilder.build());
validateSplitIndex(metadata, clusterBlocks, resizeSourceIndex.getName(), resizeIntoName, indexSettingsBuilder.build());
indexSettingsBuilder.putNull(INDEX_SHRINK_INITIAL_RECOVERY_KEY);
} else if (type == ResizeType.CLONE) {
validateCloneIndex(currentState, resizeSourceIndex.getName(), resizeIntoName, indexSettingsBuilder.build());
validateCloneIndex(metadata, clusterBlocks, resizeSourceIndex.getName(), resizeIntoName, indexSettingsBuilder.build());
indexSettingsBuilder.putNull(INDEX_SHRINK_INITIAL_RECOVERY_KEY);
} else {
throw new IllegalStateException("unknown resize type is " + type);
Expand Down
2 changes: 1 addition & 1 deletion server/src/main/java/org/elasticsearch/cluster/metadata/MetadataIndexStateService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1100,7 +1100,7 @@ private ClusterState openIndices(final Index[] indices, final ClusterState curre
}
}

shardLimitValidator.validateShardLimit(currentState, indices);
shardLimitValidator.validateShardLimit(currentState.nodes(), currentState.metadata(), indices);
if (indicesToOpen.isEmpty()) {
return currentState;
}
Expand Down
7 changes: 6 additions & 1 deletion server/src/main/java/org/elasticsearch/cluster/metadata/MetadataUpdateSettingsService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -255,7 +255,12 @@ ClusterState execute(ClusterState currentState) {
final int updatedNumberOfReplicas = IndexMetadata.INDEX_NUMBER_OF_REPLICAS_SETTING.get(openSettings);
if (preserveExisting == false) {
// Verify that this won't take us over the cluster shard limit.
shardLimitValidator.validateShardLimitOnReplicaUpdate(currentState, request.indices(), updatedNumberOfReplicas);
shardLimitValidator.validateShardLimitOnReplicaUpdate(
currentState.nodes(),
currentState.metadata(),
request.indices(),
updatedNumberOfReplicas
);

/*
* We do not update the in-sync allocation IDs as they will be removed upon the first index operation
Expand Down
36 changes: 29 additions & 7 deletions server/src/main/java/org/elasticsearch/health/node/ShardsCapacityHealthIndicatorService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,8 @@

package org.elasticsearch.health.node;

import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.metadata.Metadata;
import org.elasticsearch.cluster.node.DiscoveryNodes;
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.TriFunction;
import org.elasticsearch.common.settings.Setting;
Expand Down Expand Up @@ -124,8 +125,18 @@ public HealthIndicatorResult calculate(boolean verbose, int maxAffectedResources
var shardLimitsMetadata = healthMetadata.getShardLimitsMetadata();
return mergeIndicators(
verbose,
calculateFrom(shardLimitsMetadata.maxShardsPerNode(), state, ShardLimitValidator::checkShardLimitForNormalNodes),
calculateFrom(shardLimitsMetadata.maxShardsPerNodeFrozen(), state, ShardLimitValidator::checkShardLimitForFrozenNodes)
calculateFrom(
shardLimitsMetadata.maxShardsPerNode(),
state.nodes(),
state.metadata(),
ShardLimitValidator::checkShardLimitForNormalNodes
),
calculateFrom(
shardLimitsMetadata.maxShardsPerNodeFrozen(),
state.nodes(),
state.metadata(),
ShardLimitValidator::checkShardLimitForFrozenNodes
)
);
}

Expand Down Expand Up @@ -173,13 +184,18 @@ private HealthIndicatorResult mergeIndicators(boolean verbose, StatusResult data
);
}

static StatusResult calculateFrom(int maxShardsPerNodeSetting, ClusterState state, ShardsCapacityChecker checker) {
var result = checker.check(maxShardsPerNodeSetting, 5, 1, state);
static StatusResult calculateFrom(
int maxShardsPerNodeSetting,
DiscoveryNodes discoveryNodes,
Metadata metadata,
ShardsCapacityChecker checker
) {
var result = checker.check(maxShardsPerNodeSetting, 5, 1, discoveryNodes, metadata);
if (result.canAddShards() == false) {
return new StatusResult(HealthStatus.RED, result);
}

result = checker.check(maxShardsPerNodeSetting, 10, 1, state);
result = checker.check(maxShardsPerNodeSetting, 10, 1, discoveryNodes, metadata);
if (result.canAddShards() == false) {
return new StatusResult(HealthStatus.YELLOW, result);
}
Expand Down Expand Up @@ -225,6 +241,12 @@ record StatusResult(HealthStatus status, ShardLimitValidator.Result result) {}

@FunctionalInterface
interface ShardsCapacityChecker {
ShardLimitValidator.Result check(int maxConfiguredShardsPerNode, int numberOfNewShards, int replicas, ClusterState state);
ShardLimitValidator.Result check(
int maxConfiguredShardsPerNode,
int numberOfNewShards,
int replicas,
DiscoveryNodes discoveryNodes,
Metadata metadata
);
}
}
Loading

0 comments on commit 0aff606

Please sign in to comment.