From 25633bee882d585969631277ae80041967680418 Mon Sep 17 00:00:00 2001 From: Gaurav Bafna <85113518+gbbafna@users.noreply.github.com> Date: Fri, 29 Jul 2022 15:21:56 +0530 Subject: [PATCH] Making shard copy count a multiple of attribute count (#3462) * Making all the copies a multiple of attribute count Signed-off-by: Gaurav Bafna (cherry picked from commit 82ff463fd1bc07c241c8674b72182f2875a60425) --- .../admin/indices/rollover/RolloverIT.java | 43 +++++++ .../allocation/AwarenessAllocationIT.java | 8 ++ .../settings/UpdateNumberOfReplicasIT.java | 46 ++++++- .../template/SimpleIndexTemplateIT.java | 47 ++++++-- .../snapshots/RestoreSnapshotIT.java | 66 ++++++++-- .../metadata/MetadataCreateIndexService.java | 31 ++++- .../MetadataIndexTemplateService.java | 6 +- .../MetadataUpdateSettingsService.java | 19 ++- .../allocation/AwarenessReplicaBalance.java | 114 ++++++++++++++++++ .../common/settings/ClusterSettings.java | 2 + .../main/java/org/opensearch/node/Node.java | 9 +- .../MetadataRolloverServiceTests.java | 10 +- .../MetadataCreateIndexServiceTests.java | 65 +++++++++- .../MetadataIndexTemplateServiceTests.java | 18 ++- .../allocation/AwarenessAllocationTests.java | 1 + .../AwarenessReplicaBalanceTests.java | 66 ++++++++++ .../indices/cluster/ClusterStateChanges.java | 11 +- .../snapshots/SnapshotResiliencyTests.java | 4 +- .../test/OpenSearchIntegTestCase.java | 23 ++++ 19 files changed, 551 insertions(+), 38 deletions(-) create mode 100644 server/src/main/java/org/opensearch/cluster/routing/allocation/AwarenessReplicaBalance.java create mode 100644 server/src/test/java/org/opensearch/cluster/routing/allocation/AwarenessReplicaBalanceTests.java diff --git a/server/src/internalClusterTest/java/org/opensearch/action/admin/indices/rollover/RolloverIT.java b/server/src/internalClusterTest/java/org/opensearch/action/admin/indices/rollover/RolloverIT.java index 178dc2c6ffa87..d0ff3ef19a028 100644 --- a/server/src/internalClusterTest/java/org/opensearch/action/admin/indices/rollover/RolloverIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/action/admin/indices/rollover/RolloverIT.java @@ -61,6 +61,7 @@ import java.util.List; import java.util.Set; +import static org.hamcrest.Matchers.containsString; import static org.opensearch.index.mapper.MapperService.SINGLE_MAPPING_NAME; import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked; import static org.hamcrest.Matchers.containsInAnyOrder; @@ -172,6 +173,7 @@ public void testRolloverWithNoWriteIndex() { } public void testRolloverWithIndexSettings() throws Exception { + Alias testAlias = new Alias("test_alias"); boolean explicitWriteIndex = randomBoolean(); if (explicitWriteIndex) { @@ -210,6 +212,47 @@ public void testRolloverWithIndexSettings() throws Exception { } } + public void testRolloverWithIndexSettingsBalancedReplica() throws Exception { + Alias testAlias = new Alias("test_alias"); + boolean explicitWriteIndex = randomBoolean(); + if (explicitWriteIndex) { + testAlias.writeIndex(true); + } + assertAcked(prepareCreate("test_index-2").addAlias(testAlias).get()); + manageReplicaBalanceSetting(true); + index("test_index-2", "type1", "1", "field", "value"); + flush("test_index-2"); + final Settings settings = Settings.builder() + .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1) + .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0) + .build(); + + final IllegalArgumentException restoreError = expectThrows( + IllegalArgumentException.class, + () -> client().admin().indices().prepareRolloverIndex("test_alias").settings(settings).alias(new Alias("extra_alias")).get() + ); + + assertThat( + restoreError.getMessage(), + containsString("expected total copies needs to be a multiple of total awareness attributes [2]") + ); + + final Settings balancedReplicaSettings = Settings.builder() + .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1) + .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 1) + .build(); + + client().admin() + .indices() + .prepareRolloverIndex("test_alias") + .settings(balancedReplicaSettings) + .alias(new Alias("extra_alias")) + .waitForActiveShards(0) + .get(); + + manageReplicaBalanceSetting(false); + } + public void testRolloverWithIndexSettingsWithoutPrefix() throws Exception { Alias testAlias = new Alias("test_alias"); boolean explicitWriteIndex = randomBoolean(); diff --git a/server/src/internalClusterTest/java/org/opensearch/cluster/allocation/AwarenessAllocationIT.java b/server/src/internalClusterTest/java/org/opensearch/cluster/allocation/AwarenessAllocationIT.java index 2b73c5da27606..fbcf5d3bc78f6 100644 --- a/server/src/internalClusterTest/java/org/opensearch/cluster/allocation/AwarenessAllocationIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/cluster/allocation/AwarenessAllocationIT.java @@ -36,12 +36,14 @@ import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.opensearch.action.admin.cluster.health.ClusterHealthResponse; +import org.opensearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest; import org.opensearch.cluster.ClusterState; import org.opensearch.cluster.metadata.IndexMetadata; import org.opensearch.cluster.metadata.IndexMetadata.State; import org.opensearch.cluster.routing.IndexRoutingTable; import org.opensearch.cluster.routing.IndexShardRoutingTable; import org.opensearch.cluster.routing.ShardRouting; +import org.opensearch.cluster.routing.allocation.AwarenessReplicaBalance; import org.opensearch.cluster.routing.allocation.decider.AwarenessAllocationDecider; import org.opensearch.common.Priority; import org.opensearch.common.settings.Settings; @@ -77,6 +79,12 @@ public void testSimpleAwareness() throws Exception { logger.info("--> starting 2 nodes on the same rack"); internalCluster().startNodes(2, Settings.builder().put(commonSettings).put("node.attr.rack_id", "rack_1").build()); + Settings settings = Settings.builder() + .put(AwarenessReplicaBalance.CLUSTER_ROUTING_ALLOCATION_AWARENESS_BALANCE_SETTING.getKey(), false) + .build(); + ClusterUpdateSettingsRequest updateSettingsRequest = new ClusterUpdateSettingsRequest(); + updateSettingsRequest.persistentSettings(settings); + createIndex("test1"); createIndex("test2"); diff --git a/server/src/internalClusterTest/java/org/opensearch/indices/settings/UpdateNumberOfReplicasIT.java b/server/src/internalClusterTest/java/org/opensearch/indices/settings/UpdateNumberOfReplicasIT.java index f78ecd82834c2..98001b447e8b2 100644 --- a/server/src/internalClusterTest/java/org/opensearch/indices/settings/UpdateNumberOfReplicasIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/indices/settings/UpdateNumberOfReplicasIT.java @@ -45,11 +45,11 @@ import java.io.IOException; import java.util.EnumSet; +import static org.hamcrest.Matchers.equalTo; import static org.opensearch.common.xcontent.XContentFactory.jsonBuilder; import static org.opensearch.index.query.QueryBuilders.matchAllQuery; import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked; import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertHitCount; -import static org.hamcrest.Matchers.equalTo; @OpenSearchIntegTestCase.ClusterScope(minNumDataNodes = 2) public class UpdateNumberOfReplicasIT extends OpenSearchIntegTestCase { @@ -606,4 +606,48 @@ public void testUpdateNumberOfReplicasAllowNoIndices() { assertThat(numberOfReplicas, equalTo(0)); } + public void testAwarenessReplicaBalance() { + createIndex("aware-replica", Settings.builder().put("index.number_of_replicas", 0).build()); + createIndex(".system-index", Settings.builder().put("index.number_of_replicas", 0).build()); + manageReplicaBalanceSetting(true); + int updated = 0; + + try { + // replica count of 1 is ideal + client().admin() + .indices() + .prepareUpdateSettings("aware-replica") + .setSettings(Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 1)) + .execute() + .actionGet(); + updated++; + + // system index - should be able to update + client().admin() + .indices() + .prepareUpdateSettings(".system-index") + .setSettings(Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 2)) + .execute() + .actionGet(); + updated++; + + client().admin() + .indices() + .prepareUpdateSettings("aware-replica") + .setSettings(Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 2)) + .execute() + .actionGet(); + fail("should have thrown an exception about the replica count"); + + } catch (IllegalArgumentException e) { + assertEquals( + "Validation Failed: 1: expected total copies needs to be a multiple of total awareness attributes [2];", + e.getMessage() + ); + assertEquals(2, updated); + } finally { + manageReplicaBalanceSetting(false); + } + } + } diff --git a/server/src/internalClusterTest/java/org/opensearch/indices/template/SimpleIndexTemplateIT.java b/server/src/internalClusterTest/java/org/opensearch/indices/template/SimpleIndexTemplateIT.java index 0e15a0c895432..42c0145676f2d 100644 --- a/server/src/internalClusterTest/java/org/opensearch/indices/template/SimpleIndexTemplateIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/indices/template/SimpleIndexTemplateIT.java @@ -32,18 +32,19 @@ package org.opensearch.indices.template; +import org.junit.After; import org.opensearch.action.ActionRequestValidationException; import org.opensearch.action.admin.indices.alias.Alias; import org.opensearch.action.admin.indices.alias.get.GetAliasesResponse; import org.opensearch.action.admin.indices.settings.get.GetSettingsResponse; import org.opensearch.action.admin.indices.template.get.GetIndexTemplatesResponse; import org.opensearch.action.admin.indices.template.put.PutIndexTemplateRequestBuilder; - import org.opensearch.action.bulk.BulkResponse; import org.opensearch.action.index.IndexRequest; import org.opensearch.action.search.SearchResponse; import org.opensearch.cluster.ClusterState; import org.opensearch.cluster.metadata.AliasMetadata; +import org.opensearch.cluster.metadata.IndexMetadata; import org.opensearch.common.ParsingException; import org.opensearch.common.bytes.BytesArray; import org.opensearch.common.settings.Settings; @@ -52,12 +53,11 @@ import org.opensearch.index.mapper.MapperParsingException; import org.opensearch.index.query.QueryBuilders; import org.opensearch.indices.InvalidAliasNameException; +import org.opensearch.indices.InvalidIndexTemplateException; import org.opensearch.plugins.Plugin; import org.opensearch.search.SearchHit; -import org.opensearch.test.OpenSearchIntegTestCase; import org.opensearch.test.InternalSettingsPlugin; - -import org.junit.After; +import org.opensearch.test.OpenSearchIntegTestCase; import java.io.IOException; import java.util.ArrayList; @@ -68,11 +68,6 @@ import java.util.List; import java.util.Set; -import static org.opensearch.action.support.WriteRequest.RefreshPolicy.IMMEDIATE; -import static org.opensearch.index.query.QueryBuilders.termQuery; -import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked; -import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertHitCount; -import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertRequestBuilderThrows; import static org.hamcrest.Matchers.anyOf; import static org.hamcrest.Matchers.containsInAnyOrder; import static org.hamcrest.Matchers.containsString; @@ -83,6 +78,11 @@ import static org.hamcrest.Matchers.is; import static org.hamcrest.Matchers.notNullValue; import static org.hamcrest.Matchers.nullValue; +import static org.opensearch.action.support.WriteRequest.RefreshPolicy.IMMEDIATE; +import static org.opensearch.index.query.QueryBuilders.termQuery; +import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked; +import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertHitCount; +import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertRequestBuilderThrows; public class SimpleIndexTemplateIT extends OpenSearchIntegTestCase { @@ -1029,4 +1029,33 @@ public void testPartitionedTemplate() throws Exception { GetSettingsResponse getSettingsResponse = client().admin().indices().prepareGetSettings("test_good").get(); assertEquals("6", getSettingsResponse.getIndexToSettings().get("test_good").get("index.routing_partition_size")); } + + public void testAwarenessReplicaBalance() throws IOException { + manageReplicaBalanceSetting(true); + try { + client().admin() + .indices() + .preparePutTemplate("template_1") + .setPatterns(Arrays.asList("a*", "b*")) + .setSettings(Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 1)) + .get(); + + client().admin() + .indices() + .preparePutTemplate("template_1") + .setPatterns(Arrays.asList("a*", "b*")) + .setSettings(Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)) + .get(); + + fail("should have thrown an exception about the replica count"); + } catch (InvalidIndexTemplateException e) { + assertEquals( + "index_template [template_1] invalid, cause [Validation Failed: 1: expected total copies needs to be a multiple of total awareness attributes [2];]", + e.getMessage() + ); + } finally { + manageReplicaBalanceSetting(false); + } + } + } diff --git a/server/src/internalClusterTest/java/org/opensearch/snapshots/RestoreSnapshotIT.java b/server/src/internalClusterTest/java/org/opensearch/snapshots/RestoreSnapshotIT.java index 3a7fdd4093657..a40378b9c2dfa 100644 --- a/server/src/internalClusterTest/java/org/opensearch/snapshots/RestoreSnapshotIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/snapshots/RestoreSnapshotIT.java @@ -59,6 +59,14 @@ import java.util.stream.Collectors; import java.util.stream.IntStream; +import static org.hamcrest.Matchers.allOf; +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.greaterThan; +import static org.hamcrest.Matchers.is; +import static org.hamcrest.Matchers.not; +import static org.hamcrest.Matchers.notNullValue; +import static org.hamcrest.Matchers.nullValue; import static org.opensearch.cluster.metadata.IndexMetadata.SETTING_NUMBER_OF_REPLICAS; import static org.opensearch.cluster.metadata.IndexMetadata.SETTING_NUMBER_OF_SHARDS; import static org.opensearch.index.IndexSettings.INDEX_REFRESH_INTERVAL_SETTING; @@ -70,14 +78,6 @@ import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertIndexTemplateExists; import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertIndexTemplateMissing; import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertRequestBuilderThrows; -import static org.hamcrest.Matchers.allOf; -import static org.hamcrest.Matchers.containsString; -import static org.hamcrest.Matchers.equalTo; -import static org.hamcrest.Matchers.greaterThan; -import static org.hamcrest.Matchers.is; -import static org.hamcrest.Matchers.not; -import static org.hamcrest.Matchers.notNullValue; -import static org.hamcrest.Matchers.nullValue; public class RestoreSnapshotIT extends AbstractSnapshotIntegTestCase { @@ -973,4 +973,54 @@ public void testForbidDisableSoftDeletesDuringRestore() throws Exception { ); assertThat(restoreError.getMessage(), containsString("cannot disable setting [index.soft_deletes.enabled] on restore")); } + + public void testRestoreBalancedReplica() { + try { + createRepository("test-repo", "fs"); + createIndex("test-index", Settings.builder().put("index.number_of_replicas", 0).build()); + createIndex(".system-index", Settings.builder().put("index.number_of_replicas", 0).build()); + ensureGreen(); + clusterAdmin().prepareCreateSnapshot("test-repo", "snapshot-0") + .setIndices("test-index", ".system-index") + .setWaitForCompletion(true) + .get(); + manageReplicaBalanceSetting(true); + + final IllegalArgumentException restoreError = expectThrows( + IllegalArgumentException.class, + () -> clusterAdmin().prepareRestoreSnapshot("test-repo", "snapshot-0") + .setRenamePattern("test-index") + .setRenameReplacement("new-index") + .setIndices("test-index") + .get() + ); + assertThat( + restoreError.getMessage(), + containsString("expected total copies needs to be a multiple of total awareness attributes [2]") + ); + + RestoreSnapshotResponse restoreSnapshotResponse = clusterAdmin().prepareRestoreSnapshot("test-repo", "snapshot-0") + .setRenamePattern(".system-index") + .setRenameReplacement(".system-index-restore-1") + .setWaitForCompletion(true) + .setIndices(".system-index") + .execute() + .actionGet(); + + assertThat(restoreSnapshotResponse.getRestoreInfo().totalShards(), greaterThan(0)); + + restoreSnapshotResponse = clusterAdmin().prepareRestoreSnapshot("test-repo", "snapshot-0") + .setRenamePattern("test-index") + .setRenameReplacement("new-index") + .setIndexSettings(Settings.builder().put("index.number_of_replicas", 1).build()) + .setWaitForCompletion(true) + .setIndices("test-index") + .execute() + .actionGet(); + + assertThat(restoreSnapshotResponse.getRestoreInfo().totalShards(), greaterThan(0)); + } finally { + manageReplicaBalanceSetting(false); + } + } } diff --git a/server/src/main/java/org/opensearch/cluster/metadata/MetadataCreateIndexService.java b/server/src/main/java/org/opensearch/cluster/metadata/MetadataCreateIndexService.java index c63e0a498ef6f..79419810bdebd 100644 --- a/server/src/main/java/org/opensearch/cluster/metadata/MetadataCreateIndexService.java +++ b/server/src/main/java/org/opensearch/cluster/metadata/MetadataCreateIndexService.java @@ -59,6 +59,7 @@ import org.opensearch.cluster.routing.ShardRouting; import org.opensearch.cluster.routing.ShardRoutingState; import org.opensearch.cluster.routing.allocation.AllocationService; +import org.opensearch.cluster.routing.allocation.AwarenessReplicaBalance; import org.opensearch.cluster.service.ClusterService; import org.opensearch.common.Nullable; import org.opensearch.common.Priority; @@ -102,6 +103,7 @@ import java.util.List; import java.util.Locale; import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiConsumer; @@ -145,6 +147,7 @@ public class MetadataCreateIndexService { private final ShardLimitValidator shardLimitValidator; private final boolean forbidPrivateIndexSettings; private final Set indexSettingProviders = new HashSet<>(); + private AwarenessReplicaBalance awarenessReplicaBalance; public MetadataCreateIndexService( final Settings settings, @@ -158,7 +161,8 @@ public MetadataCreateIndexService( final ThreadPool threadPool, final NamedXContentRegistry xContentRegistry, final SystemIndices systemIndices, - final boolean forbidPrivateIndexSettings + final boolean forbidPrivateIndexSettings, + final AwarenessReplicaBalance awarenessReplicaBalance ) { this.settings = settings; this.clusterService = clusterService; @@ -172,6 +176,7 @@ public MetadataCreateIndexService( this.systemIndices = systemIndices; this.forbidPrivateIndexSettings = forbidPrivateIndexSettings; this.shardLimitValidator = shardLimitValidator; + this.awarenessReplicaBalance = awarenessReplicaBalance; } /** @@ -1171,7 +1176,7 @@ private void validate(CreateIndexClusterStateUpdateRequest request, ClusterState public void validateIndexSettings(String indexName, final Settings settings, final boolean forbidPrivateIndexSettings) throws IndexCreationException { - List validationErrors = getIndexSettingsValidationErrors(settings, forbidPrivateIndexSettings); + List validationErrors = getIndexSettingsValidationErrors(settings, forbidPrivateIndexSettings, indexName); if (validationErrors.isEmpty() == false) { ValidationException validationException = new ValidationException(); @@ -1180,11 +1185,31 @@ public void validateIndexSettings(String indexName, final Settings settings, fin } } - List getIndexSettingsValidationErrors(final Settings settings, final boolean forbidPrivateIndexSettings) { + List getIndexSettingsValidationErrors(final Settings settings, final boolean forbidPrivateIndexSettings, String indexName) { + List validationErrors = getIndexSettingsValidationErrors(settings, forbidPrivateIndexSettings, Optional.of(indexName)); + return validationErrors; + } + + List getIndexSettingsValidationErrors( + final Settings settings, + final boolean forbidPrivateIndexSettings, + Optional indexName + ) { List validationErrors = validateIndexCustomPath(settings, env.sharedDataFile()); if (forbidPrivateIndexSettings) { validationErrors.addAll(validatePrivateSettingsNotExplicitlySet(settings, indexScopedSettings)); } + if (indexName.isEmpty() || indexName.get().charAt(0) != '.') { + // Apply aware replica balance only to non system indices + int replicaCount = settings.getAsInt( + IndexMetadata.SETTING_NUMBER_OF_REPLICAS, + INDEX_NUMBER_OF_REPLICAS_SETTING.getDefault(Settings.EMPTY) + ); + Optional error = awarenessReplicaBalance.validate(replicaCount); + if (error.isPresent()) { + validationErrors.add(error.get()); + } + } return validationErrors; } diff --git a/server/src/main/java/org/opensearch/cluster/metadata/MetadataIndexTemplateService.java b/server/src/main/java/org/opensearch/cluster/metadata/MetadataIndexTemplateService.java index 1a0f4f0f83e00..7e91b491a234c 100644 --- a/server/src/main/java/org/opensearch/cluster/metadata/MetadataIndexTemplateService.java +++ b/server/src/main/java/org/opensearch/cluster/metadata/MetadataIndexTemplateService.java @@ -1471,7 +1471,11 @@ private void validate(String name, @Nullable Settings settings, List ind validationErrors.add(t.getMessage()); } } - List indexSettingsValidation = metadataCreateIndexService.getIndexSettingsValidationErrors(settings, true); + List indexSettingsValidation = metadataCreateIndexService.getIndexSettingsValidationErrors( + settings, + true, + Optional.empty() + ); validationErrors.addAll(indexSettingsValidation); } diff --git a/server/src/main/java/org/opensearch/cluster/metadata/MetadataUpdateSettingsService.java b/server/src/main/java/org/opensearch/cluster/metadata/MetadataUpdateSettingsService.java index 1390860271577..eb142be815d27 100644 --- a/server/src/main/java/org/opensearch/cluster/metadata/MetadataUpdateSettingsService.java +++ b/server/src/main/java/org/opensearch/cluster/metadata/MetadataUpdateSettingsService.java @@ -46,6 +46,7 @@ import org.opensearch.cluster.block.ClusterBlocks; import org.opensearch.cluster.routing.RoutingTable; import org.opensearch.cluster.routing.allocation.AllocationService; +import org.opensearch.cluster.routing.allocation.AwarenessReplicaBalance; import org.opensearch.cluster.service.ClusterService; import org.opensearch.common.Priority; import org.opensearch.common.ValidationException; @@ -89,6 +90,8 @@ public class MetadataUpdateSettingsService { private final ShardLimitValidator shardLimitValidator; private final ThreadPool threadPool; + private AwarenessReplicaBalance awarenessReplicaBalance; + @Inject public MetadataUpdateSettingsService( ClusterService clusterService, @@ -96,7 +99,8 @@ public MetadataUpdateSettingsService( IndexScopedSettings indexScopedSettings, IndicesService indicesService, ShardLimitValidator shardLimitValidator, - ThreadPool threadPool + ThreadPool threadPool, + AwarenessReplicaBalance awarenessReplicaBalance ) { this.clusterService = clusterService; this.threadPool = threadPool; @@ -104,6 +108,7 @@ public MetadataUpdateSettingsService( this.indexScopedSettings = indexScopedSettings; this.indicesService = indicesService; this.shardLimitValidator = shardLimitValidator; + this.awarenessReplicaBalance = awarenessReplicaBalance; } public void updateSettings( @@ -193,6 +198,18 @@ public ClusterState execute(ClusterState currentState) { if (IndexMetadata.INDEX_NUMBER_OF_REPLICAS_SETTING.exists(openSettings)) { final int updatedNumberOfReplicas = IndexMetadata.INDEX_NUMBER_OF_REPLICAS_SETTING.get(openSettings); if (preserveExisting == false) { + for (Index index : request.indices()) { + if (index.getName().charAt(0) != '.') { + // No replica count validation for system indices + Optional error = awarenessReplicaBalance.validate(updatedNumberOfReplicas); + if (error.isPresent()) { + ValidationException ex = new ValidationException(); + ex.addValidationError(error.get()); + throw ex; + } + } + } + // Verify that this won't take us over the cluster shard limit. int totalNewShards = Arrays.stream(request.indices()) .mapToInt(i -> getTotalNewShards(i, currentState, updatedNumberOfReplicas)) diff --git a/server/src/main/java/org/opensearch/cluster/routing/allocation/AwarenessReplicaBalance.java b/server/src/main/java/org/opensearch/cluster/routing/allocation/AwarenessReplicaBalance.java new file mode 100644 index 0000000000000..accf0b69a4f0e --- /dev/null +++ b/server/src/main/java/org/opensearch/cluster/routing/allocation/AwarenessReplicaBalance.java @@ -0,0 +1,114 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.cluster.routing.allocation; + +import org.opensearch.common.settings.ClusterSettings; +import org.opensearch.common.settings.Setting; +import org.opensearch.common.settings.Settings; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +import static java.lang.Math.max; +import static org.opensearch.cluster.routing.allocation.decider.AwarenessAllocationDecider.CLUSTER_ROUTING_ALLOCATION_AWARENESS_ATTRIBUTE_SETTING; +import static org.opensearch.cluster.routing.allocation.decider.AwarenessAllocationDecider.CLUSTER_ROUTING_ALLOCATION_AWARENESS_FORCE_GROUP_SETTING; + +/** + * This {@link AwarenessReplicaBalance} gives total unique values of awareness attributes + * It takes in effect only iff cluster.routing.allocation.awareness.attributes and + * cluster.routing.allocation.awareness.force.zone.values both are specified. + * + * This is used in enforcing total copy of shard is a maximum of unique values of awareness attributes + * Helps in balancing shards across all awareness attributes and ensuring high availability of data. + */ +public class AwarenessReplicaBalance { + public static final Setting CLUSTER_ROUTING_ALLOCATION_AWARENESS_BALANCE_SETTING = Setting.boolSetting( + "cluster.routing.allocation.awareness.balance", + false, + Setting.Property.Dynamic, + Setting.Property.NodeScope + ); + + private volatile List awarenessAttributes; + + private volatile Map> forcedAwarenessAttributes; + + private volatile Boolean awarenessBalance; + + public AwarenessReplicaBalance(Settings settings, ClusterSettings clusterSettings) { + this.awarenessAttributes = CLUSTER_ROUTING_ALLOCATION_AWARENESS_ATTRIBUTE_SETTING.get(settings); + clusterSettings.addSettingsUpdateConsumer(CLUSTER_ROUTING_ALLOCATION_AWARENESS_ATTRIBUTE_SETTING, this::setAwarenessAttributes); + setForcedAwarenessAttributes(CLUSTER_ROUTING_ALLOCATION_AWARENESS_FORCE_GROUP_SETTING.get(settings)); + clusterSettings.addSettingsUpdateConsumer( + CLUSTER_ROUTING_ALLOCATION_AWARENESS_FORCE_GROUP_SETTING, + this::setForcedAwarenessAttributes + ); + setAwarenessBalance(CLUSTER_ROUTING_ALLOCATION_AWARENESS_BALANCE_SETTING.get(settings)); + clusterSettings.addSettingsUpdateConsumer(CLUSTER_ROUTING_ALLOCATION_AWARENESS_BALANCE_SETTING, this::setAwarenessBalance); + + } + + private void setAwarenessBalance(Boolean awarenessBalance) { + this.awarenessBalance = awarenessBalance; + } + + private void setForcedAwarenessAttributes(Settings forceSettings) { + Map> forcedAwarenessAttributes = new HashMap<>(); + Map forceGroups = forceSettings.getAsGroups(); + for (Map.Entry entry : forceGroups.entrySet()) { + List aValues = entry.getValue().getAsList("values"); + if (aValues.size() > 0) { + forcedAwarenessAttributes.put(entry.getKey(), aValues); + } + } + this.forcedAwarenessAttributes = forcedAwarenessAttributes; + } + + private void setAwarenessAttributes(List awarenessAttributes) { + this.awarenessAttributes = awarenessAttributes; + } + + /* + For a cluster having zone as awareness attribute , it will return the size of zones if set it forced awareness attributes + + If there are multiple forced awareness attributes, it will return size of the largest list, as all copies of data + is supposed to get distributed amongst those. + + cluster.routing.allocation.awareness.attributes: rack_id , zone + cluster.routing.allocation.awareness.force.zone.values: zone1, zone2 + cluster.routing.allocation.awareness.force.rack_id.values: rack_id1, rack_id2, rack_id3 + + In this case, awareness attributes would be 3. + */ + public int maxAwarenessAttributes() { + int awarenessAttributes = 1; + if (this.awarenessBalance == false) { + return awarenessAttributes; + } + for (String awarenessAttribute : this.awarenessAttributes) { + if (forcedAwarenessAttributes.containsKey(awarenessAttribute)) { + awarenessAttributes = max(awarenessAttributes, forcedAwarenessAttributes.get(awarenessAttribute).size()); + } + } + return awarenessAttributes; + } + + public Optional validate(int replicaCount) { + if ((replicaCount + 1) % maxAwarenessAttributes() != 0) { + String errorMessage = "expected total copies needs to be a multiple of total awareness attributes [" + + maxAwarenessAttributes() + + "]"; + return Optional.of(errorMessage); + } + return Optional.empty(); + } + +} diff --git a/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java b/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java index 16f2ac19833af..464b6e9ee39de 100644 --- a/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java +++ b/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java @@ -33,6 +33,7 @@ import org.apache.logging.log4j.LogManager; import org.opensearch.action.main.TransportMainAction; +import org.opensearch.cluster.routing.allocation.AwarenessReplicaBalance; import org.opensearch.cluster.routing.allocation.decider.NodeLoadAwareAllocationDecider; import org.opensearch.index.IndexModule; import org.opensearch.index.IndexSettings; @@ -219,6 +220,7 @@ public void apply(Settings value, Settings current, Settings previous) { Arrays.asList( AwarenessAllocationDecider.CLUSTER_ROUTING_ALLOCATION_AWARENESS_ATTRIBUTE_SETTING, AwarenessAllocationDecider.CLUSTER_ROUTING_ALLOCATION_AWARENESS_FORCE_GROUP_SETTING, + AwarenessReplicaBalance.CLUSTER_ROUTING_ALLOCATION_AWARENESS_BALANCE_SETTING, BalancedShardsAllocator.INDEX_BALANCE_FACTOR_SETTING, BalancedShardsAllocator.SHARD_BALANCE_FACTOR_SETTING, BalancedShardsAllocator.SHARD_MOVE_PRIMARY_FIRST_SETTING, diff --git a/server/src/main/java/org/opensearch/node/Node.java b/server/src/main/java/org/opensearch/node/Node.java index de558b2746c5e..47ac1b5058d74 100644 --- a/server/src/main/java/org/opensearch/node/Node.java +++ b/server/src/main/java/org/opensearch/node/Node.java @@ -36,6 +36,7 @@ import org.apache.logging.log4j.Logger; import org.apache.lucene.util.Constants; import org.apache.lucene.util.SetOnce; +import org.opensearch.cluster.routing.allocation.AwarenessReplicaBalance; import org.opensearch.index.IndexingPressureService; import org.opensearch.watcher.ResourceWatcherService; import org.opensearch.Assertions; @@ -643,6 +644,10 @@ protected Node( final AliasValidator aliasValidator = new AliasValidator(); final ShardLimitValidator shardLimitValidator = new ShardLimitValidator(settings, clusterService, systemIndices); + final AwarenessReplicaBalance awarenessReplicaBalance = new AwarenessReplicaBalance( + settings, + clusterService.getClusterSettings() + ); final MetadataCreateIndexService metadataCreateIndexService = new MetadataCreateIndexService( settings, clusterService, @@ -655,7 +660,8 @@ protected Node( threadPool, xContentRegistry, systemIndices, - forbidPrivateIndexSettings + forbidPrivateIndexSettings, + awarenessReplicaBalance ); pluginsService.filterPlugins(Plugin.class) .forEach( @@ -912,6 +918,7 @@ protected Node( b.bind(IndicesService.class).toInstance(indicesService); b.bind(AliasValidator.class).toInstance(aliasValidator); b.bind(MetadataCreateIndexService.class).toInstance(metadataCreateIndexService); + b.bind(AwarenessReplicaBalance.class).toInstance(awarenessReplicaBalance); b.bind(MetadataCreateDataStreamService.class).toInstance(metadataCreateDataStreamService); b.bind(SearchService.class).toInstance(searchService); b.bind(SearchTransportService.class).toInstance(searchTransportService); diff --git a/server/src/test/java/org/opensearch/action/admin/indices/rollover/MetadataRolloverServiceTests.java b/server/src/test/java/org/opensearch/action/admin/indices/rollover/MetadataRolloverServiceTests.java index 9ebfc33617920..2317edd37320c 100644 --- a/server/src/test/java/org/opensearch/action/admin/indices/rollover/MetadataRolloverServiceTests.java +++ b/server/src/test/java/org/opensearch/action/admin/indices/rollover/MetadataRolloverServiceTests.java @@ -57,6 +57,7 @@ import org.opensearch.cluster.metadata.MetadataIndexAliasesService; import org.opensearch.cluster.metadata.Template; import org.opensearch.cluster.routing.allocation.AllocationService; +import org.opensearch.cluster.routing.allocation.AwarenessReplicaBalance; import org.opensearch.cluster.service.ClusterService; import org.opensearch.common.CheckedFunction; import org.opensearch.common.Strings; @@ -617,7 +618,8 @@ public void testRolloverClusterState() throws Exception { testThreadPool, null, systemIndices, - false + false, + new AwarenessReplicaBalance(Settings.EMPTY, clusterService.getClusterSettings()) ); MetadataIndexAliasesService indexAliasesService = new MetadataIndexAliasesService( clusterService, @@ -754,7 +756,8 @@ public void testRolloverClusterStateForDataStream() throws Exception { testThreadPool, null, systemIndices, - false + false, + new AwarenessReplicaBalance(Settings.EMPTY, clusterService.getClusterSettings()) ); MetadataIndexAliasesService indexAliasesService = new MetadataIndexAliasesService( clusterService, @@ -931,7 +934,8 @@ public void testRolloverClusterStateForDataStreamNoTemplate() throws Exception { testThreadPool, null, new SystemIndices(emptyMap()), - false + false, + new AwarenessReplicaBalance(Settings.EMPTY, clusterService.getClusterSettings()) ); MetadataIndexAliasesService indexAliasesService = new MetadataIndexAliasesService( clusterService, diff --git a/server/src/test/java/org/opensearch/cluster/metadata/MetadataCreateIndexServiceTests.java b/server/src/test/java/org/opensearch/cluster/metadata/MetadataCreateIndexServiceTests.java index 5bea69c5bbd66..2d589cf58dc22 100644 --- a/server/src/test/java/org/opensearch/cluster/metadata/MetadataCreateIndexServiceTests.java +++ b/server/src/test/java/org/opensearch/cluster/metadata/MetadataCreateIndexServiceTests.java @@ -52,13 +52,16 @@ import org.opensearch.cluster.node.DiscoveryNodes; import org.opensearch.cluster.routing.RoutingTable; import org.opensearch.cluster.routing.allocation.AllocationService; +import org.opensearch.cluster.routing.allocation.AwarenessReplicaBalance; import org.opensearch.cluster.routing.allocation.allocator.BalancedShardsAllocator; import org.opensearch.cluster.routing.allocation.decider.AllocationDeciders; +import org.opensearch.cluster.routing.allocation.decider.AwarenessAllocationDecider; import org.opensearch.cluster.routing.allocation.decider.MaxRetryAllocationDecider; import org.opensearch.cluster.service.ClusterService; import org.opensearch.common.Strings; import org.opensearch.common.collect.ImmutableOpenMap; import org.opensearch.common.compress.CompressedXContent; +import org.opensearch.common.settings.ClusterSettings; import org.opensearch.common.settings.IndexScopedSettings; import org.opensearch.common.settings.Setting; import org.opensearch.common.settings.Settings; @@ -96,6 +99,7 @@ import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.UUID; import java.util.concurrent.atomic.AtomicBoolean; @@ -109,7 +113,6 @@ import static java.util.Collections.emptyMap; import static java.util.Collections.singleton; import static java.util.Collections.singletonList; -import static org.opensearch.index.IndexSettings.INDEX_SOFT_DELETES_SETTING; import static org.hamcrest.Matchers.endsWith; import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.hasKey; @@ -118,6 +121,8 @@ import static org.hamcrest.Matchers.notNullValue; import static org.hamcrest.Matchers.nullValue; import static org.hamcrest.Matchers.startsWith; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; import static org.opensearch.cluster.metadata.IndexMetadata.INDEX_NUMBER_OF_ROUTING_SHARDS_SETTING; import static org.opensearch.cluster.metadata.IndexMetadata.INDEX_NUMBER_OF_SHARDS_SETTING; import static org.opensearch.cluster.metadata.IndexMetadata.INDEX_READ_ONLY_BLOCK; @@ -131,6 +136,7 @@ import static org.opensearch.cluster.metadata.MetadataCreateIndexService.getIndexNumberOfRoutingShards; import static org.opensearch.cluster.metadata.MetadataCreateIndexService.parseV1Mappings; import static org.opensearch.cluster.metadata.MetadataCreateIndexService.resolveAndValidateAliases; +import static org.opensearch.index.IndexSettings.INDEX_SOFT_DELETES_SETTING; import static org.opensearch.indices.ShardLimitValidatorTests.createTestShardLimitService; public class MetadataCreateIndexServiceTests extends OpenSearchTestCase { @@ -601,7 +607,8 @@ public void testValidateIndexName() throws Exception { threadPool, null, new SystemIndices(Collections.emptyMap()), - false + false, + new AwarenessReplicaBalance(Settings.EMPTY, clusterService.getClusterSettings()) ); validateIndexName(checkerService, "index?name", "must not contain the following characters " + Strings.INVALID_FILENAME_CHARS); @@ -682,7 +689,8 @@ public void testValidateDotIndex() { threadPool, null, new SystemIndices(Collections.singletonMap("foo", systemIndexDescriptors)), - false + false, + new AwarenessReplicaBalance(Settings.EMPTY, clusterService.getClusterSettings()) ); // Check deprecations assertFalse(checkerService.validateDotIndex(".test2", false)); @@ -1073,6 +1081,52 @@ public void testParseMappingsWithTypelessTemplate() throws Exception { assertThat(mappings, Matchers.hasKey(MapperService.SINGLE_MAPPING_NAME)); } + public void testvalidateIndexSettings() { + ClusterService clusterService = mock(ClusterService.class); + ThreadPool threadPool = new TestThreadPool(getTestName()); + Settings settings = Settings.builder() + .put(AwarenessAllocationDecider.CLUSTER_ROUTING_ALLOCATION_AWARENESS_ATTRIBUTE_SETTING.getKey(), "zone, rack") + .put(AwarenessAllocationDecider.CLUSTER_ROUTING_ALLOCATION_AWARENESS_FORCE_GROUP_SETTING.getKey() + "zone.values", "a, b") + .put(AwarenessAllocationDecider.CLUSTER_ROUTING_ALLOCATION_AWARENESS_FORCE_GROUP_SETTING.getKey() + "rack.values", "c, d, e") + .put(AwarenessReplicaBalance.CLUSTER_ROUTING_ALLOCATION_AWARENESS_BALANCE_SETTING.getKey(), true) + .build(); + ClusterSettings clusterSettings = new ClusterSettings(settings, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS); + when(clusterService.getSettings()).thenReturn(settings); + when(clusterService.getClusterSettings()).thenReturn(clusterSettings); + MetadataCreateIndexService checkerService = new MetadataCreateIndexService( + settings, + clusterService, + null, + null, + null, + createTestShardLimitService(randomIntBetween(1, 1000), clusterService), + new Environment(Settings.builder().put("path.home", "dummy").build(), null), + IndexScopedSettings.DEFAULT_SCOPED_SETTINGS, + threadPool, + null, + new SystemIndices(Collections.emptyMap()), + true, + new AwarenessReplicaBalance(settings, clusterService.getClusterSettings()) + ); + + List validationErrors = checkerService.getIndexSettingsValidationErrors(settings, false, Optional.empty()); + assertThat(validationErrors.size(), is(1)); + assertThat(validationErrors.get(0), is("expected total copies needs to be a multiple of total awareness attributes [3]")); + + settings = Settings.builder() + .put(AwarenessAllocationDecider.CLUSTER_ROUTING_ALLOCATION_AWARENESS_ATTRIBUTE_SETTING.getKey(), "zone, rack") + .put(AwarenessAllocationDecider.CLUSTER_ROUTING_ALLOCATION_AWARENESS_FORCE_GROUP_SETTING.getKey() + "zone.values", "a, b") + .put(AwarenessAllocationDecider.CLUSTER_ROUTING_ALLOCATION_AWARENESS_FORCE_GROUP_SETTING.getKey() + "rack.values", "c, d, e") + .put(AwarenessReplicaBalance.CLUSTER_ROUTING_ALLOCATION_AWARENESS_BALANCE_SETTING.getKey(), true) + .put(SETTING_NUMBER_OF_REPLICAS, 2) + .build(); + + validationErrors = checkerService.getIndexSettingsValidationErrors(settings, false, Optional.empty()); + assertThat(validationErrors.size(), is(0)); + + threadPool.shutdown(); + } + public void testBuildIndexMetadata() { IndexMetadata sourceIndexMetadata = IndexMetadata.builder("parent") .settings(Settings.builder().put("index.version.created", Version.CURRENT).build()) @@ -1195,10 +1249,11 @@ public void testIndexLifecycleNameSetting() { threadPool, null, new SystemIndices(Collections.emptyMap()), - true + true, + new AwarenessReplicaBalance(Settings.EMPTY, clusterService.getClusterSettings()) ); - final List validationErrors = checkerService.getIndexSettingsValidationErrors(ilnSetting, true); + final List validationErrors = checkerService.getIndexSettingsValidationErrors(ilnSetting, true, Optional.empty()); assertThat(validationErrors.size(), is(1)); assertThat(validationErrors.get(0), is("expected [index.lifecycle.name] to be private but it was not")); })); diff --git a/server/src/test/java/org/opensearch/cluster/metadata/MetadataIndexTemplateServiceTests.java b/server/src/test/java/org/opensearch/cluster/metadata/MetadataIndexTemplateServiceTests.java index c3a16a1e25bc8..887d8469bd01c 100644 --- a/server/src/test/java/org/opensearch/cluster/metadata/MetadataIndexTemplateServiceTests.java +++ b/server/src/test/java/org/opensearch/cluster/metadata/MetadataIndexTemplateServiceTests.java @@ -38,9 +38,11 @@ import org.opensearch.action.support.master.AcknowledgedResponse; import org.opensearch.cluster.ClusterState; import org.opensearch.cluster.metadata.MetadataIndexTemplateService.PutRequest; +import org.opensearch.cluster.routing.allocation.AwarenessReplicaBalance; import org.opensearch.cluster.service.ClusterService; import org.opensearch.common.Strings; import org.opensearch.common.compress.CompressedXContent; +import org.opensearch.common.settings.ClusterSettings; import org.opensearch.common.settings.IndexScopedSettings; import org.opensearch.common.settings.Settings; import org.opensearch.common.unit.TimeValue; @@ -85,7 +87,10 @@ import static org.hamcrest.Matchers.empty; import static org.hamcrest.Matchers.is; import static org.hamcrest.Matchers.matchesRegex; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; import static org.opensearch.common.settings.Settings.builder; +import static org.opensearch.env.Environment.PATH_HOME_SETTING; import static org.opensearch.index.mapper.DataStreamFieldMapper.Defaults.TIMESTAMP_FIELD; import static org.opensearch.indices.ShardLimitValidatorTests.createTestShardLimitService; @@ -2084,9 +2089,14 @@ public void testLegacyNoopUpdate() { } private static List putTemplate(NamedXContentRegistry xContentRegistry, PutRequest request) { + ClusterService clusterService = mock(ClusterService.class); + Settings settings = Settings.builder().put(PATH_HOME_SETTING.getKey(), "dummy").build(); + ClusterSettings clusterSettings = new ClusterSettings(settings, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS); + when(clusterService.getSettings()).thenReturn(settings); + when(clusterService.getClusterSettings()).thenReturn(clusterSettings); MetadataCreateIndexService createIndexService = new MetadataCreateIndexService( Settings.EMPTY, - null, + clusterService, null, null, null, @@ -2096,7 +2106,8 @@ private static List putTemplate(NamedXContentRegistry xContentRegistr null, xContentRegistry, new SystemIndices(Collections.emptyMap()), - true + true, + new AwarenessReplicaBalance(Settings.EMPTY, clusterService.getClusterSettings()) ); MetadataIndexTemplateService service = new MetadataIndexTemplateService( null, @@ -2158,7 +2169,8 @@ private MetadataIndexTemplateService getMetadataIndexTemplateService() { null, xContentRegistry(), new SystemIndices(Collections.emptyMap()), - true + true, + new AwarenessReplicaBalance(Settings.EMPTY, clusterService.getClusterSettings()) ); return new MetadataIndexTemplateService( clusterService, diff --git a/server/src/test/java/org/opensearch/cluster/routing/allocation/AwarenessAllocationTests.java b/server/src/test/java/org/opensearch/cluster/routing/allocation/AwarenessAllocationTests.java index b2adcd21cd8c9..c0cec7e3201bb 100644 --- a/server/src/test/java/org/opensearch/cluster/routing/allocation/AwarenessAllocationTests.java +++ b/server/src/test/java/org/opensearch/cluster/routing/allocation/AwarenessAllocationTests.java @@ -64,6 +64,7 @@ import java.util.Map; import static java.util.Collections.singletonMap; +import static org.hamcrest.MatcherAssert.assertThat; import static org.opensearch.cluster.routing.ShardRoutingState.INITIALIZING; import static org.opensearch.cluster.routing.ShardRoutingState.RELOCATING; import static org.opensearch.cluster.routing.ShardRoutingState.STARTED; diff --git a/server/src/test/java/org/opensearch/cluster/routing/allocation/AwarenessReplicaBalanceTests.java b/server/src/test/java/org/opensearch/cluster/routing/allocation/AwarenessReplicaBalanceTests.java new file mode 100644 index 0000000000000..e2431765709e6 --- /dev/null +++ b/server/src/test/java/org/opensearch/cluster/routing/allocation/AwarenessReplicaBalanceTests.java @@ -0,0 +1,66 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.cluster.routing.allocation; + +import org.opensearch.cluster.OpenSearchAllocationTestCase; +import org.opensearch.cluster.routing.allocation.decider.AwarenessAllocationDecider; +import org.opensearch.common.settings.ClusterSettings; +import org.opensearch.common.settings.Settings; + +import java.util.Optional; + +import static org.hamcrest.Matchers.equalTo; + +public class AwarenessReplicaBalanceTests extends OpenSearchAllocationTestCase { + + private static final ClusterSettings EMPTY_CLUSTER_SETTINGS = new ClusterSettings( + Settings.EMPTY, + ClusterSettings.BUILT_IN_CLUSTER_SETTINGS + ); + + public void testNoForcedAwarenessAttribute() { + Settings settings = Settings.builder().put("cluster.routing.allocation.awareness.attributes", "rack_id").build(); + + AwarenessReplicaBalance awarenessReplicaBalance = new AwarenessReplicaBalance(settings, EMPTY_CLUSTER_SETTINGS); + assertThat(awarenessReplicaBalance.maxAwarenessAttributes(), equalTo(1)); + + assertEquals(awarenessReplicaBalance.validate(0), Optional.empty()); + assertEquals(awarenessReplicaBalance.validate(1), Optional.empty()); + } + + public void testForcedAwarenessAttribute() { + Settings settings = Settings.builder() + .put(AwarenessAllocationDecider.CLUSTER_ROUTING_ALLOCATION_AWARENESS_ATTRIBUTE_SETTING.getKey(), "zone, rack") + .put(AwarenessAllocationDecider.CLUSTER_ROUTING_ALLOCATION_AWARENESS_FORCE_GROUP_SETTING.getKey() + "zone.values", "a, b") + .put(AwarenessAllocationDecider.CLUSTER_ROUTING_ALLOCATION_AWARENESS_FORCE_GROUP_SETTING.getKey() + "rack.values", "c, d, e") + .put(AwarenessReplicaBalance.CLUSTER_ROUTING_ALLOCATION_AWARENESS_BALANCE_SETTING.getKey(), true) + .build(); + + AwarenessReplicaBalance awarenessReplicaBalance = new AwarenessReplicaBalance(settings, EMPTY_CLUSTER_SETTINGS); + assertThat(awarenessReplicaBalance.maxAwarenessAttributes(), equalTo(3)); + assertEquals(awarenessReplicaBalance.validate(2), Optional.empty()); + assertEquals( + awarenessReplicaBalance.validate(1), + Optional.of("expected total copies needs to be a multiple of total awareness attributes [3]") + ); + } + + public void testForcedAwarenessAttributeDisabled() { + Settings settings = Settings.builder() + .put(AwarenessAllocationDecider.CLUSTER_ROUTING_ALLOCATION_AWARENESS_ATTRIBUTE_SETTING.getKey(), "zone, rack") + .put(AwarenessReplicaBalance.CLUSTER_ROUTING_ALLOCATION_AWARENESS_BALANCE_SETTING.getKey(), true) + .build(); + + AwarenessReplicaBalance awarenessReplicaBalance = new AwarenessReplicaBalance(settings, EMPTY_CLUSTER_SETTINGS); + assertThat(awarenessReplicaBalance.maxAwarenessAttributes(), equalTo(1)); + assertEquals(awarenessReplicaBalance.validate(0), Optional.empty()); + assertEquals(awarenessReplicaBalance.validate(1), Optional.empty()); + } + +} diff --git a/server/src/test/java/org/opensearch/indices/cluster/ClusterStateChanges.java b/server/src/test/java/org/opensearch/indices/cluster/ClusterStateChanges.java index 769fdc220bec4..7fe17e570d157 100644 --- a/server/src/test/java/org/opensearch/indices/cluster/ClusterStateChanges.java +++ b/server/src/test/java/org/opensearch/indices/cluster/ClusterStateChanges.java @@ -81,6 +81,7 @@ import org.opensearch.cluster.node.DiscoveryNode; import org.opensearch.cluster.routing.ShardRouting; import org.opensearch.cluster.routing.allocation.AllocationService; +import org.opensearch.cluster.routing.allocation.AwarenessReplicaBalance; import org.opensearch.cluster.routing.allocation.FailedShard; import org.opensearch.cluster.routing.allocation.RandomAllocationDeciderTests; import org.opensearch.cluster.routing.allocation.allocator.BalancedShardsAllocator; @@ -190,6 +191,7 @@ public ClusterStateChanges(NamedXContentRegistry xContentRegistry, ThreadPool th // mocks clusterService = mock(ClusterService.class); when(clusterService.getClusterSettings()).thenReturn(clusterSettings); + when(clusterService.getSettings()).thenReturn(SETTINGS); IndicesService indicesService = mock(IndicesService.class); // MetadataCreateIndexService uses withTempIndexService to check mappings -> fake it here try { @@ -272,13 +274,17 @@ public IndexMetadata upgradeIndexMetadata(IndexMetadata indexMetadata, Version m transportVerifyShardIndexBlockAction ); MetadataDeleteIndexService deleteIndexService = new MetadataDeleteIndexService(SETTINGS, clusterService, allocationService); + + final AwarenessReplicaBalance awarenessReplicaBalance = new AwarenessReplicaBalance(SETTINGS, clusterService.getClusterSettings()); + MetadataUpdateSettingsService metadataUpdateSettingsService = new MetadataUpdateSettingsService( clusterService, allocationService, IndexScopedSettings.DEFAULT_SCOPED_SETTINGS, indicesService, shardLimitValidator, - threadPool + threadPool, + awarenessReplicaBalance ); MetadataCreateIndexService createIndexService = new MetadataCreateIndexService( SETTINGS, @@ -292,7 +298,8 @@ public IndexMetadata upgradeIndexMetadata(IndexMetadata indexMetadata, Version m threadPool, xContentRegistry, systemIndices, - true + true, + awarenessReplicaBalance ); transportCloseIndexAction = new TransportCloseIndexAction( diff --git a/server/src/test/java/org/opensearch/snapshots/SnapshotResiliencyTests.java b/server/src/test/java/org/opensearch/snapshots/SnapshotResiliencyTests.java index d24016922651c..6c79be3a8239d 100644 --- a/server/src/test/java/org/opensearch/snapshots/SnapshotResiliencyTests.java +++ b/server/src/test/java/org/opensearch/snapshots/SnapshotResiliencyTests.java @@ -141,6 +141,7 @@ import org.opensearch.cluster.routing.ShardRouting; import org.opensearch.cluster.routing.UnassignedInfo; import org.opensearch.cluster.routing.allocation.AllocationService; +import org.opensearch.cluster.routing.allocation.AwarenessReplicaBalance; import org.opensearch.cluster.routing.allocation.command.AllocateEmptyPrimaryAllocationCommand; import org.opensearch.cluster.service.ClusterApplierService; import org.opensearch.cluster.service.ClusterManagerService; @@ -1890,7 +1891,8 @@ public void onFailure(final Exception e) { threadPool, namedXContentRegistry, systemIndices, - false + false, + new AwarenessReplicaBalance(Settings.EMPTY, clusterService.getClusterSettings()) ); actions.put( CreateIndexAction.INSTANCE, diff --git a/test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java b/test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java index 73e47fcaedb23..fef69cde3cea0 100644 --- a/test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java +++ b/test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java @@ -49,6 +49,7 @@ import org.opensearch.action.admin.cluster.node.hotthreads.NodeHotThreads; import org.opensearch.action.admin.cluster.node.info.NodeInfo; import org.opensearch.action.admin.cluster.node.info.NodesInfoResponse; +import org.opensearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest; import org.opensearch.action.admin.cluster.state.ClusterStateResponse; import org.opensearch.action.admin.cluster.tasks.PendingClusterTasksResponse; import org.opensearch.action.admin.indices.create.CreateIndexRequestBuilder; @@ -91,7 +92,9 @@ import org.opensearch.cluster.routing.IndexShardRoutingTable; import org.opensearch.cluster.routing.ShardRouting; import org.opensearch.cluster.routing.UnassignedInfo; +import org.opensearch.cluster.routing.allocation.AwarenessReplicaBalance; import org.opensearch.cluster.routing.allocation.DiskThresholdSettings; +import org.opensearch.cluster.routing.allocation.decider.AwarenessAllocationDecider; import org.opensearch.cluster.routing.allocation.decider.EnableAllocationDecider; import org.opensearch.cluster.service.ClusterService; import org.opensearch.common.Nullable; @@ -2414,4 +2417,24 @@ protected boolean willSufferDebian8MemoryProblem() { boolean java15Plus = Runtime.version().compareTo(Version.parse("15")) >= 0; return anyDebian8Nodes && java15Plus == false; } + + public void manageReplicaBalanceSetting(boolean apply) { + Settings settings; + if (apply) { + settings = Settings.builder() + .put(AwarenessAllocationDecider.CLUSTER_ROUTING_ALLOCATION_AWARENESS_ATTRIBUTE_SETTING.getKey(), "zone") + .put(AwarenessAllocationDecider.CLUSTER_ROUTING_ALLOCATION_AWARENESS_FORCE_GROUP_SETTING.getKey() + "zone.values", "a, b") + .put(AwarenessReplicaBalance.CLUSTER_ROUTING_ALLOCATION_AWARENESS_BALANCE_SETTING.getKey(), true) + .build(); + } else { + settings = Settings.builder() + .putNull(AwarenessAllocationDecider.CLUSTER_ROUTING_ALLOCATION_AWARENESS_ATTRIBUTE_SETTING.getKey()) + .putNull(AwarenessAllocationDecider.CLUSTER_ROUTING_ALLOCATION_AWARENESS_FORCE_GROUP_SETTING.getKey() + "zone.values") + .putNull(AwarenessReplicaBalance.CLUSTER_ROUTING_ALLOCATION_AWARENESS_BALANCE_SETTING.getKey()) + .build(); + } + ClusterUpdateSettingsRequest updateSettingsRequest = new ClusterUpdateSettingsRequest(); + updateSettingsRequest.persistentSettings(settings); + assertAcked(client().admin().cluster().updateSettings(updateSettingsRequest).actionGet()); + } }