-
Notifications
You must be signed in to change notification settings - Fork 1.8k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add canRemain method to TargetPoolAllocationDecider to move shards from
local to remote pool for hot to warm tiering Signed-off-by: Neetika Singhal <[email protected]>
- Loading branch information
1 parent
6dbb079
commit dce0342
Showing
6 changed files
with
274 additions
and
4 deletions.
There are no files selected for viewing
82 changes: 82 additions & 0 deletions
82
server/src/main/java/org/opensearch/action/admin/indices/tiering/TieringUtils.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,82 @@ | ||
| /* | ||
| * 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.action.admin.indices.tiering; | ||
|
|
||
| import org.opensearch.cluster.metadata.IndexMetadata; | ||
| import org.opensearch.cluster.routing.ShardRouting; | ||
| import org.opensearch.cluster.routing.allocation.RoutingAllocation; | ||
| import org.opensearch.common.annotation.ExperimentalApi; | ||
| import org.opensearch.common.util.FeatureFlags; | ||
| import org.opensearch.core.index.Index; | ||
| import org.opensearch.index.IndexModule; | ||
|
|
||
| import java.util.LinkedList; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
|
|
||
| import static org.opensearch.index.IndexModule.INDEX_TIERING_STATE; | ||
|
|
||
| /** | ||
| * Utility class for tiering operations | ||
| * | ||
| * @opensearch.internal | ||
| */ | ||
| @ExperimentalApi | ||
| public class TieringUtils { | ||
|
|
||
| /** | ||
| * Constructs a HotToWarmTieringResponse from the rejected indices map | ||
| * | ||
| * @param rejectedIndices the rejected indices map | ||
| * @return the HotToWarmTieringResponse object | ||
| */ | ||
| public static HotToWarmTieringResponse constructToHotToWarmTieringResponse(final Map<Index, String> rejectedIndices) { | ||
| final List<HotToWarmTieringResponse.IndexResult> indicesResult = new LinkedList<>(); | ||
| for (Map.Entry<Index, String> rejectedIndex : rejectedIndices.entrySet()) { | ||
| indicesResult.add(new HotToWarmTieringResponse.IndexResult(rejectedIndex.getKey().getName(), rejectedIndex.getValue())); | ||
| } | ||
| return new HotToWarmTieringResponse(true, indicesResult); | ||
| } | ||
|
|
||
| /** | ||
| * Checks if the specified index is in the "hot" tiering state. | ||
| * | ||
| * @param indexMetadata the metadata of the index | ||
| * @return true if the index is in the "hot" tiering state, false otherwise | ||
| */ | ||
| public static boolean isHotIndex(final IndexMetadata indexMetadata) { | ||
| return IndexModule.TieringState.HOT.name().equals(INDEX_TIERING_STATE.get(indexMetadata.getSettings())); | ||
| } | ||
|
|
||
| /** | ||
| * Checks if the specified shard is a partial shard by | ||
| * checking the INDEX_STORE_LOCALITY_SETTING for its index. | ||
| * see {@link #isPartialIndex(IndexMetadata)} | ||
| * @param shard ShardRouting object representing the shard | ||
| * @param allocation RoutingAllocation object representing the allocation | ||
| * @return true if the shard is a partial shard, false otherwise | ||
| */ | ||
| public static boolean isPartialShard(ShardRouting shard, RoutingAllocation allocation) { | ||
| IndexMetadata indexMetadata = allocation.metadata().getIndexSafe(shard.index()); | ||
| return isPartialIndex(indexMetadata); | ||
| } | ||
|
|
||
| /** | ||
| * Checks if the specified index is a partial index by | ||
| * checking the INDEX_STORE_LOCALITY_SETTING for the index. | ||
| * | ||
| * @param indexMetadata the metadata of the index | ||
| * @return true if the index is a partial index, false otherwise | ||
| */ | ||
| public static boolean isPartialIndex(final IndexMetadata indexMetadata) { | ||
| return FeatureFlags.isEnabled(FeatureFlags.TIERED_REMOTE_INDEX) | ||
| && IndexModule.DataLocalityType.PARTIAL.name() | ||
| .equals(indexMetadata.getSettings().get(IndexModule.INDEX_STORE_LOCALITY_SETTING.getKey())); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
100 changes: 100 additions & 0 deletions
100
...src/test/java/org/opensearch/cluster/routing/allocation/ShardsTieringAllocationTests.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,100 @@ | ||
| /* | ||
| * 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.Version; | ||
| import org.opensearch.cluster.ClusterState; | ||
| import org.opensearch.cluster.metadata.IndexMetadata; | ||
| import org.opensearch.cluster.routing.RoutingNode; | ||
| import org.opensearch.cluster.routing.RoutingNodes; | ||
| import org.opensearch.cluster.routing.RoutingPool; | ||
| import org.opensearch.cluster.routing.ShardRouting; | ||
| import org.opensearch.index.IndexModule; | ||
|
|
||
| import static org.opensearch.cluster.routing.RoutingPool.LOCAL_ONLY; | ||
| import static org.opensearch.cluster.routing.RoutingPool.REMOTE_CAPABLE; | ||
| import static org.opensearch.cluster.routing.RoutingPool.getIndexPool; | ||
| import static org.opensearch.index.IndexModule.INDEX_STORE_LOCALITY_SETTING; | ||
|
|
||
| public class ShardsTieringAllocationTests extends TieringAllocationBaseTestCase { | ||
|
|
||
| public void testShardsWithNoTiering() { | ||
| int localOnlyNodes = 5; | ||
| int remoteCapableNodes = 3; | ||
| int localIndices = 5; | ||
| int remoteIndices = 0; | ||
| ClusterState clusterState = createInitialCluster(localOnlyNodes, remoteCapableNodes, localIndices, remoteIndices); | ||
| 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()) { | ||
| assertFalse(shard.unassigned()); | ||
| RoutingPool shardPool = RoutingPool.getShardPool(shard, allocation); | ||
| assertEquals(LOCAL_ONLY, shardPool); | ||
| } | ||
| } | ||
|
|
||
| public void testShardsWithTiering() { | ||
| int localOnlyNodes = 15; | ||
| int remoteCapableNodes = 13; | ||
| int localIndices = 10; | ||
| int remoteIndices = 0; | ||
| ClusterState clusterState = createInitialCluster(localOnlyNodes, remoteCapableNodes, localIndices, remoteIndices); | ||
| AllocationService service = this.createRemoteCapableAllocationService(); | ||
| // assign shards to respective nodes | ||
| clusterState = allocateShardsAndBalance(clusterState, service); | ||
| // put indices in the hot to warm tiering state | ||
| clusterState = updateIndexMetadataForTiering( | ||
| clusterState, | ||
| localIndices, | ||
| IndexModule.TieringState.HOT_TO_WARM.name(), | ||
| IndexModule.DataLocalityType.PARTIAL.name() | ||
| ); | ||
| // trigger shard relocation | ||
| clusterState = allocateShardsAndBalance(clusterState, service); | ||
| RoutingNodes routingNodes = clusterState.getRoutingNodes(); | ||
| RoutingAllocation allocation = getRoutingAllocation(clusterState, routingNodes); | ||
| assertEquals(0, routingNodes.unassigned().size()); | ||
|
|
||
| for (ShardRouting shard : clusterState.getRoutingTable().allShards()) { | ||
| assertFalse(shard.unassigned()); | ||
| RoutingNode node = routingNodes.node(shard.currentNodeId()); | ||
| RoutingPool nodePool = RoutingPool.getNodePool(node); | ||
| RoutingPool shardPool = RoutingPool.getShardPool(shard, allocation); | ||
| assertEquals(RoutingPool.REMOTE_CAPABLE, shardPool); | ||
| assertEquals(nodePool, shardPool); | ||
| } | ||
| } | ||
|
|
||
| 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())) | ||
| .numberOfShards(PRIMARIES) | ||
| .numberOfReplicas(REPLICAS) | ||
| .build(); | ||
| RoutingPool indexPool = getIndexPool(indexMetadata); | ||
| assertEquals(REMOTE_CAPABLE, indexPool); | ||
| } | ||
|
|
||
| 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())) | ||
| .numberOfShards(PRIMARIES) | ||
| .numberOfReplicas(REPLICAS) | ||
| .build(); | ||
| RoutingPool indexPool = getIndexPool(indexMetadata); | ||
| assertEquals(LOCAL_ONLY, indexPool); | ||
| } | ||
| } |
60 changes: 60 additions & 0 deletions
60
...rc/test/java/org/opensearch/cluster/routing/allocation/TieringAllocationBaseTestCase.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,60 @@ | ||
| /* | ||
| * 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.ClusterState; | ||
| import org.opensearch.cluster.metadata.IndexMetadata; | ||
| import org.opensearch.cluster.metadata.Metadata; | ||
| import org.opensearch.common.SuppressForbidden; | ||
| import org.opensearch.common.settings.Settings; | ||
| import org.opensearch.common.util.FeatureFlags; | ||
| import org.junit.AfterClass; | ||
| import org.junit.BeforeClass; | ||
|
|
||
| import static org.opensearch.index.IndexModule.INDEX_STORE_LOCALITY_SETTING; | ||
| import static org.opensearch.index.IndexModule.INDEX_TIERING_STATE; | ||
|
|
||
| @SuppressForbidden(reason = "feature flag overrides") | ||
| public abstract class TieringAllocationBaseTestCase extends RemoteShardsBalancerBaseTestCase { | ||
|
|
||
| @BeforeClass | ||
| public static void setup() { | ||
| System.setProperty(FeatureFlags.TIERED_REMOTE_INDEX, "true"); | ||
| } | ||
|
|
||
| @AfterClass | ||
| public static void teardown() { | ||
| System.setProperty(FeatureFlags.TIERED_REMOTE_INDEX, "false"); | ||
| } | ||
|
|
||
| public ClusterState updateIndexMetadataForTiering( | ||
| ClusterState clusterState, | ||
| int localIndices, | ||
| String tieringState, | ||
| String dataLocality | ||
| ) { | ||
| Metadata.Builder mb = Metadata.builder(clusterState.metadata()); | ||
| for (int i = 0; i < localIndices; i++) { | ||
| IndexMetadata indexMetadata = clusterState.metadata().index(getIndexName(i, false)); | ||
| Settings settings = indexMetadata.getSettings(); | ||
| mb.put( | ||
| IndexMetadata.builder(indexMetadata) | ||
| .settings( | ||
| Settings.builder() | ||
| .put(settings) | ||
| .put(settings) | ||
| .put(INDEX_TIERING_STATE.getKey(), tieringState) | ||
| .put(INDEX_STORE_LOCALITY_SETTING.getKey(), dataLocality) | ||
| ) | ||
| ); | ||
| } | ||
| Metadata metadata = mb.build(); | ||
| return ClusterState.builder(clusterState).metadata(metadata).build(); | ||
| } | ||
| } |