-
Notifications
You must be signed in to change notification settings - Fork 2k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Signed-off-by: Prudhvi Godithi <pgodithi@amazon.com>
- Loading branch information
1 parent
db5212b
commit 6b8e897
Showing
31 changed files
with
1,953 additions
and
1,041 deletions.
There are no files selected for viewing
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
87 changes: 87 additions & 0 deletions
87
...main/java/org/opensearch/action/admin/indices/scale/searchonly/NodeSearchOnlyRequest.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,87 @@ | ||
| /* | ||
| * 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.scale.searchonly; | ||
|
|
||
| import org.opensearch.core.common.io.stream.StreamInput; | ||
| import org.opensearch.core.common.io.stream.StreamOutput; | ||
| import org.opensearch.core.index.shard.ShardId; | ||
| import org.opensearch.transport.TransportRequest; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.List; | ||
|
|
||
| /** | ||
| * A transport request sent to nodes to facilitate shard synchronization during search-only scaling operations. | ||
| * <p> | ||
| * This request is sent from the cluster manager to data nodes that host primary shards for the target index | ||
| * during scale operations. It contains the index name and a list of shard IDs that need to be synchronized | ||
| * before completing a scale-down operation. | ||
| * <p> | ||
| * When a node receives this request, it performs final sync and flush operations on the specified shards, | ||
| * ensuring all operations are committed and the remote store is synced. This is a crucial step in | ||
| * the scale-down process to ensure no data loss occurs when the index transitions to search-only mode. | ||
| */ | ||
| class NodeSearchOnlyRequest extends TransportRequest { | ||
| private final String index; | ||
| private final List<ShardId> shardIds; | ||
|
|
||
| /** | ||
| * Constructs a new NodeSearchOnlyRequest. | ||
| * | ||
| * @param index the name of the index being scaled | ||
| * @param shardIds the list of shard IDs to be synchronized on the target node | ||
| */ | ||
| NodeSearchOnlyRequest(String index, List<ShardId> shardIds) { | ||
| this.index = index; | ||
| this.shardIds = shardIds; | ||
| } | ||
|
|
||
| /** | ||
| * Deserialization constructor. | ||
| * | ||
| * @param in the stream input to read from | ||
| * @throws IOException if there is an I/O error during deserialization | ||
| */ | ||
| NodeSearchOnlyRequest(StreamInput in) throws IOException { | ||
| super(in); | ||
| this.index = in.readString(); | ||
| this.shardIds = in.readList(ShardId::new); | ||
| } | ||
|
|
||
| /** | ||
| * Serializes this request to the given output stream. | ||
| * | ||
| * @param out the output stream to write to | ||
| * @throws IOException if there is an I/O error during serialization | ||
| */ | ||
| @Override | ||
| public void writeTo(StreamOutput out) throws IOException { | ||
| super.writeTo(out); | ||
| out.writeString(index); | ||
| out.writeList(shardIds); | ||
| } | ||
|
|
||
| /** | ||
| * Returns the index name associated with this request. | ||
| * | ||
| * @return the index name | ||
| */ | ||
| String getIndex() { | ||
| return index; | ||
| } | ||
|
|
||
| /** | ||
| * Returns the list of shard IDs to be synchronized. | ||
| * | ||
| * @return the list of shard IDs | ||
| */ | ||
| List<ShardId> getShardIds() { | ||
| return shardIds; | ||
| } | ||
| } |
89 changes: 89 additions & 0 deletions
89
...ain/java/org/opensearch/action/admin/indices/scale/searchonly/NodeSearchOnlyResponse.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,89 @@ | ||
| /* | ||
| * 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.scale.searchonly; | ||
|
|
||
| import org.opensearch.cluster.node.DiscoveryNode; | ||
| import org.opensearch.core.common.io.stream.StreamInput; | ||
| import org.opensearch.core.common.io.stream.StreamOutput; | ||
| import org.opensearch.core.transport.TransportResponse; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.List; | ||
|
|
||
| /** | ||
| * Response sent from nodes after processing a {@link NodeSearchOnlyRequest} during search-only scaling operations. | ||
| * <p> | ||
| * This response contains information about the node that processed the request and the results of | ||
| * synchronization attempts for each requested shard. The cluster manager uses these responses to | ||
| * determine whether it's safe to proceed with finalizing a scale-down operation. | ||
| * <p> | ||
| * Each response includes details about whether shards have any uncommitted operations or need | ||
| * additional synchronization, which would indicate the scale operation should be delayed until | ||
| * the cluster reaches a stable state. | ||
| */ | ||
| class NodeSearchOnlyResponse extends TransportResponse { | ||
| private final DiscoveryNode node; | ||
| private final List<ShardSearchOnlyResponse> shardResponses; | ||
|
|
||
| /** | ||
| * Constructs a new NodeSearchOnlyResponse. | ||
| * | ||
| * @param node the node that processed the synchronization request | ||
| * @param shardResponses the list of responses from individual shard synchronization attempts | ||
| */ | ||
| NodeSearchOnlyResponse(DiscoveryNode node, List<ShardSearchOnlyResponse> shardResponses) { | ||
| this.node = node; | ||
| this.shardResponses = shardResponses; | ||
| } | ||
|
|
||
| /** | ||
| * Deserialization constructor. | ||
| * | ||
| * @param in the stream input to read from | ||
| * @throws IOException if there is an I/O error during deserialization | ||
| */ | ||
| NodeSearchOnlyResponse(StreamInput in) throws IOException { | ||
| node = new DiscoveryNode(in); | ||
| shardResponses = in.readList(ShardSearchOnlyResponse::new); | ||
| } | ||
|
|
||
| /** | ||
| * Serializes this response to the given output stream. | ||
| * | ||
| * @param out the output stream to write to | ||
| * @throws IOException if there is an I/O error during serialization | ||
| */ | ||
| @Override | ||
| public void writeTo(StreamOutput out) throws IOException { | ||
| node.writeTo(out); | ||
| out.writeList(shardResponses); | ||
| } | ||
|
|
||
| /** | ||
| * Returns the node that processed the synchronization request. | ||
| * | ||
| * @return the discovery node information | ||
| */ | ||
| public DiscoveryNode getNode() { | ||
| return node; | ||
| } | ||
|
|
||
| /** | ||
| * Returns the list of shard-level synchronization responses. | ||
| * <p> | ||
| * These responses contain critical information about the state of each shard, | ||
| * including whether there are uncommitted operations or if additional synchronization | ||
| * is needed before the scale operation can safely proceed. | ||
| * | ||
| * @return the list of shard responses | ||
| */ | ||
| public List<ShardSearchOnlyResponse> getShardResponses() { | ||
| return shardResponses; | ||
| } | ||
| } |
87 changes: 87 additions & 0 deletions
87
...in/java/org/opensearch/action/admin/indices/scale/searchonly/ScaleOperationValidator.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,87 @@ | ||
| /* | ||
| * 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.scale.searchonly; | ||
|
|
||
| import org.opensearch.action.support.clustermanager.AcknowledgedResponse; | ||
| import org.opensearch.cluster.metadata.IndexMetadata; | ||
| import org.opensearch.core.action.ActionListener; | ||
| import org.opensearch.indices.replication.common.ReplicationType; | ||
|
|
||
| /** | ||
| * Validates that indices meet the prerequisites for search-only scale operations. | ||
| * <p> | ||
| * This validator ensures that indexes being scaled up or down satisfy all the | ||
| * necessary conditions for a safe scaling operation. It checks for required settings, | ||
| * index state compatibility, and configuration prerequisites such as remote store | ||
| * and segment replication settings. | ||
| */ | ||
| class ScaleOperationValidator { | ||
|
|
||
| /** | ||
| * Validates that the given index meets the prerequisites for the scale operation. | ||
| * <p> | ||
| * For scale-down operations, this method verifies: | ||
| * <ul> | ||
| * <li>The index exists</li> | ||
| * <li>The index is not already in search-only mode</li> | ||
| * <li>The index has at least one search-only replica configured</li> | ||
| * <li>Remote store is enabled for the index</li> | ||
| * <li>Segment replication is enabled for the index</li> | ||
| * </ul> | ||
| * <p> | ||
| * For scale-up operations, this method verifies: | ||
| * <ul> | ||
| * <li>The index exists</li> | ||
| * <li>The index is currently in search-only mode</li> | ||
| * </ul> | ||
| * | ||
| * @param indexMetadata the metadata of the index to validate | ||
| * @param index the name of the index being validated | ||
| * @param listener the action listener to notify in case of validation failure | ||
| * @param isScaleDown true if validating for scale-down, false for scale-up | ||
| * @return true if validation succeeds, false if validation fails (and listener is notified) | ||
| */ | ||
| boolean validateScalePrerequisites( | ||
| IndexMetadata indexMetadata, | ||
| String index, | ||
| ActionListener<AcknowledgedResponse> listener, | ||
| boolean isScaleDown | ||
| ) { | ||
| try { | ||
| if (indexMetadata == null) { | ||
| throw new IllegalArgumentException("Index [" + index + "] not found"); | ||
| } | ||
| if (isScaleDown) { | ||
| if (indexMetadata.getSettings().getAsBoolean(IndexMetadata.INDEX_BLOCKS_SEARCH_ONLY_SETTING.getKey(), false)) { | ||
| throw new IllegalStateException("Index [" + index + "] is already in search-only mode"); | ||
| } | ||
|
|
||
| if (indexMetadata.getNumberOfSearchOnlyReplicas() == 0) { | ||
| throw new IllegalArgumentException("Cannot scale to zero without search replicas for index: " + index); | ||
| } | ||
| if (!indexMetadata.getSettings().getAsBoolean(IndexMetadata.SETTING_REMOTE_STORE_ENABLED, false)) { | ||
| throw new IllegalArgumentException( | ||
| "To scale to zero, " + IndexMetadata.SETTING_REMOTE_STORE_ENABLED + " must be enabled for index: " + index | ||
| ); | ||
| } | ||
| if (!ReplicationType.SEGMENT.toString().equals(indexMetadata.getSettings().get(IndexMetadata.SETTING_REPLICATION_TYPE))) { | ||
| throw new IllegalArgumentException("To scale to zero, segment replication must be enabled for index: " + index); | ||
| } | ||
| } else { | ||
| if (!indexMetadata.getSettings().getAsBoolean(IndexMetadata.INDEX_BLOCKS_SEARCH_ONLY_SETTING.getKey(), false)) { | ||
| throw new IllegalStateException("Index [" + index + "] is not in search-only mode"); | ||
| } | ||
| } | ||
| return true; | ||
| } catch (Exception e) { | ||
| listener.onFailure(e); | ||
| return false; | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.