Adding unfollow action in ism to invoke stop replication for ccr #1198
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,32 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.indexmanagement.indexstatemanagement.action | ||
|
|
||
| import org.opensearch.indexmanagement.indexstatemanagement.step.unfollow.AttemptUnfollowStep | ||
| import org.opensearch.indexmanagement.spi.indexstatemanagement.Action | ||
| import org.opensearch.indexmanagement.spi.indexstatemanagement.Step | ||
| import org.opensearch.indexmanagement.spi.indexstatemanagement.model.StepContext | ||
|
|
||
| /** | ||
| * ISM action to stop replication on indices replicated on a follower cluster. | ||
| */ | ||
| class UnfollowAction( | ||
|
There was a problem hiding this comment. We probably want to use stop relication, which is the name of the API we have https://opensearch.org/docs/latest/tuning-your-cluster/replication-plugin/api/#stop-replication There was a problem hiding this comment. Oh, do you mean the name of the action in ism to be renamed? for ex. Sure 👍 With that, all the files could be renamed as StopReplicationAction.kt, AttemptStopReplicationStep.kt, ValidateStopReplication.kt etc. Shall I go ahead with these names? However, it would be tricky 🤔 to name the classes in ccr now - as the action-name indices:admin/plugins/replication/index/unfollow and TransportUnfollowIndexReplicationAction need to be renamed accordingly, to distinguish them from the existing stop REST API. There was a problem hiding this comment.
yes
may be There was a problem hiding this comment.
Yes, this also aligns with the action name we've decided "indices:internal/plugins/replication/index/stop". Thank you so much 🙂 I'll make these changes in subsequent pushes.. |
||
| index: Int, | ||
| ) : Action(name, index) { | ||
| companion object { | ||
| const val name = "unfollow" | ||
| } | ||
|
|
||
| private val attemptUnfollowStep = AttemptUnfollowStep() | ||
|
|
||
| private val steps = listOf(attemptUnfollowStep) | ||
|
|
||
| override fun getStepToExecute(context: StepContext): Step { | ||
| return attemptUnfollowStep | ||
| } | ||
|
|
||
| override fun getSteps(): List<Step> = steps | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,30 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.indexmanagement.indexstatemanagement.action | ||
|
|
||
| import org.opensearch.core.common.io.stream.StreamInput | ||
| import org.opensearch.core.xcontent.XContentParser | ||
| import org.opensearch.core.xcontent.XContentParserUtils.ensureExpectedToken | ||
| import org.opensearch.indexmanagement.spi.indexstatemanagement.Action | ||
| import org.opensearch.indexmanagement.spi.indexstatemanagement.ActionParser | ||
|
|
||
| class UnfollowActionParser : ActionParser() { | ||
| override fun fromStreamInput(sin: StreamInput): Action { | ||
| val index = sin.readInt() | ||
| return UnfollowAction(index) | ||
| } | ||
|
|
||
| override fun fromXContent(xcp: XContentParser, index: Int): Action { | ||
| ensureExpectedToken(XContentParser.Token.START_OBJECT, xcp.currentToken(), xcp) | ||
| ensureExpectedToken(XContentParser.Token.END_OBJECT, xcp.nextToken(), xcp) | ||
|
|
||
| return UnfollowAction(index) | ||
| } | ||
|
|
||
| override fun getActionType(): String { | ||
| return UnfollowAction.name | ||
| } | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,110 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.indexmanagement.indexstatemanagement.step.unfollow | ||
|
|
||
| import org.apache.logging.log4j.LogManager | ||
| import org.opensearch.ExceptionsHelper | ||
| import org.opensearch.action.support.master.AcknowledgedResponse | ||
| import org.opensearch.client.node.NodeClient | ||
| import org.opensearch.commons.replication.ReplicationPluginInterface | ||
| import org.opensearch.commons.replication.action.StopIndexReplicationRequest | ||
| import org.opensearch.indexmanagement.opensearchapi.suspendUntil | ||
| import org.opensearch.indexmanagement.spi.indexstatemanagement.Step | ||
| import org.opensearch.indexmanagement.spi.indexstatemanagement.model.ManagedIndexMetaData | ||
| import org.opensearch.indexmanagement.spi.indexstatemanagement.model.StepMetaData | ||
| import org.opensearch.snapshots.SnapshotInProgressException | ||
| import org.opensearch.transport.RemoteTransportException | ||
| class AttemptUnfollowStep : Step(name) { | ||
| private val logger = LogManager.getLogger(javaClass) | ||
| private var stepStatus = StepStatus.STARTING | ||
| private var info: Map<String, Any>? = null | ||
|
|
||
| override suspend fun execute(): Step { | ||
| val context = this.context ?: return this | ||
| val indexName = context.metadata.index | ||
| try { | ||
| val stopIndexReplicationRequestObj = StopIndexReplicationRequest(indexName) | ||
| /*val response: AcknowledgedResponse = | ||
| ReplicationPluginInterface.suspendUntil { | ||
| this.stopReplication( | ||
| context.client as NodeClient, | ||
| stopIndexReplicationRequestObj, | ||
| it, | ||
| ) | ||
| }*/ | ||
|
Comment on lines
+30
to
+37
There was a problem hiding this comment. Sure 👍 I had added this to get some suggestions on writing mock-uts for AttemptUnfollowStep.kt as I have tried with two different approaches using a function performStopAction() and directly invoking ReplicationPluginInterface.suspendUntil, but to no success. Do you have any suggestions on either partially mocking AttempUnfollowStep or mocking ReplicationPluginInterface.stopReplication in this case? There was a problem hiding this comment. Wondering did you try mock the client behavior like this, I suppose it would be pretty similar since both return acknowledge response There was a problem hiding this comment. Hey, AttemptRolloverStep executes the rollover like this In AttemptRolloverStepTests, context.client has been mocked in a way that when the rolloverIndex() is invoked by the test, it would return the value as directed by the mocked function. But in the AttemptUnfollowStep.kt class, the execute() method invokes it like this Here, I can mock the context.client and other params, but the test needs to mock ReplicationPluginInterface and ReplicationPluginInterface.suspendUntil to ultimately mock the response of stopReplication() function. If that does not happen, it would run the actual implementation and not the mocked one - which does not work in the UT. There was a problem hiding this comment. Hi @bowenlan-amzn , Any thoughts here? |
||
| val response = performStopAction(context.client as NodeClient, stopIndexReplicationRequestObj) | ||
|
There was a problem hiding this comment. I read the CCR doc and see if we stop relication, we won't be able to resume anymore. I'm wondering how do we know that so to prevent early stopping the replication. not requiring this since probably we can just mention this caveat in the documentation, and give a long waiting time before stop relication in the follower cluster There was a problem hiding this comment. Yes, replication cannot be resumed on an index after stopping (or pausing more than 12h). Hmm, the major use-case we had identified in ISM+CCR case was - say ISM is setup in both leader and follower clusters, for deletion and other housekeeping operations in the respective clusters. So in such cases, users could chain the actions to be preceeded by stop-replication first (before any other write actions). There was a problem hiding this comment. Understood. I'm just a little worried about one part: the follower index cannot know whether leader index finishes indexing, so it may stop relication early unexpectedly but this can be waited for community feedback whether it's needed |
||
| if (response.isAcknowledged) { | ||
| stepStatus = StepStatus.COMPLETED | ||
| info = mapOf("message" to getSuccessMessage(indexName)) | ||
| } else { | ||
| val message = getFailedMessage(indexName) | ||
| logger.warn(message) | ||
| stepStatus = StepStatus.FAILED | ||
| info = mapOf("message" to message) | ||
| } | ||
| } catch (e: RemoteTransportException) { | ||
| val cause = ExceptionsHelper.unwrapCause(e) | ||
| if (cause is SnapshotInProgressException) { | ||
| handleSnapshotException(indexName, cause) | ||
| } else { | ||
| handleException(indexName, cause as Exception) | ||
| } | ||
| } catch (e: SnapshotInProgressException) { | ||
| handleSnapshotException(indexName, e) | ||
| } catch (e: Exception) { | ||
| handleException(indexName, e) | ||
| } | ||
| return this | ||
| } | ||
|
|
||
| internal suspend fun performStopAction(client: NodeClient, request: StopIndexReplicationRequest): AcknowledgedResponse { | ||
| val response: AcknowledgedResponse = | ||
| ReplicationPluginInterface.suspendUntil { | ||
| this.stopReplication( | ||
| client, | ||
| request, | ||
| it, | ||
| ) | ||
| } | ||
| return response | ||
| } | ||
| private fun handleSnapshotException(indexName: String, e: SnapshotInProgressException) { | ||
| val message = getSnapshotMessage(indexName) | ||
| logger.warn(message, e) | ||
| stepStatus = StepStatus.CONDITION_NOT_MET | ||
| info = mapOf("message" to message) | ||
| } | ||
|
|
||
| private fun handleException(indexName: String, e: Exception) { | ||
| val message = getFailedMessage(indexName) | ||
| logger.error(message, e) | ||
| stepStatus = StepStatus.FAILED | ||
| val mutableInfo = mutableMapOf("message" to message) | ||
| val errorMessage = e.message | ||
| if (errorMessage != null) mutableInfo["cause"] = errorMessage | ||
| info = mutableInfo.toMap() | ||
| } | ||
|
|
||
| override fun getUpdatedManagedIndexMetadata(currentMetadata: ManagedIndexMetaData): ManagedIndexMetaData { | ||
| return currentMetadata.copy( | ||
| stepMetaData = StepMetaData(name, getStepStartTime(currentMetadata).toEpochMilli(), stepStatus), | ||
| transitionTo = null, | ||
| info = info, | ||
| ) | ||
| } | ||
|
|
||
| override fun isIdempotent() = true | ||
|
|
||
| companion object { | ||
| const val name = "attempt_unfollow" | ||
|
|
||
| fun getFailedMessage(index: String) = "Failed to unfollow index [index=$index]" | ||
|
|
||
| fun getSuccessMessage(index: String) = "Successfully unfollowed index [index=$index]" | ||
|
|
||
| fun getSnapshotMessage(index: String) = "Index had snapshot in progress, retrying unfollowing [index=$index]" | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,71 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.indexmanagement.indexstatemanagement.validation | ||
|
|
||
| import org.apache.logging.log4j.LogManager | ||
| import org.opensearch.cluster.metadata.MetadataCreateIndexService | ||
| import org.opensearch.cluster.service.ClusterService | ||
| import org.opensearch.common.settings.Settings | ||
| import org.opensearch.indexmanagement.spi.indexstatemanagement.Validate | ||
| import org.opensearch.indexmanagement.util.OpenForTesting | ||
| import org.opensearch.indices.InvalidIndexNameException | ||
| import org.opensearch.monitor.jvm.JvmService | ||
| @OpenForTesting | ||
| class ValidateUnfollow( | ||
| settings: Settings, | ||
| clusterService: ClusterService, | ||
| jvmService: JvmService, | ||
| ) : Validate(settings, clusterService, jvmService) { | ||
| private val logger = LogManager.getLogger(javaClass) | ||
|
|
||
| @Suppress("ReturnSuppressCount", "ReturnCount") | ||
| override fun execute(indexName: String): Validate { | ||
| // if these conditions are false, fail validation and do not execute unfollow action | ||
| if (!indexExists(indexName) || !validIndex(indexName)) { | ||
| validationStatus = ValidationStatus.FAILED | ||
| return this | ||
| } | ||
| validationMessage = getValidationPassedMessage(indexName) | ||
| return this | ||
| } | ||
|
|
||
| private fun indexExists(indexName: String): Boolean { | ||
| val isIndexExists = clusterService.state().metadata.indices.containsKey(indexName) | ||
| if (!isIndexExists) { | ||
| val message = getNoIndexMessage(indexName) | ||
| logger.warn(message) | ||
| validationMessage = message | ||
| return false | ||
| } | ||
| return true | ||
| } | ||
|
|
||
| private fun validIndex(indexName: String): Boolean { | ||
| val exceptionGenerator: (String, String) -> RuntimeException = { index_name, reason -> InvalidIndexNameException(index_name, reason) } | ||
| // If the index name is invalid for any reason, this will throw an exception giving the reason why in the message. | ||
| // That will be displayed to the user as the cause. | ||
| try { | ||
| MetadataCreateIndexService.validateIndexOrAliasName(indexName, exceptionGenerator) | ||
| } catch (e: Exception) { | ||
| val message = getIndexNotValidMessage(indexName) | ||
| logger.warn(message) | ||
| validationMessage = message | ||
| return false | ||
| } | ||
| return true | ||
| } | ||
|
|
||
| @Suppress("TooManyFunctions") | ||
| companion object { | ||
| const val name = "validate_unfollow" | ||
|
|
||
| fun getNoIndexMessage(index: String) = "No such index [index=$index] for unfollow action." | ||
|
|
||
| fun getIndexNotValidMessage(index: String) = "Index [index=$index] is not valid. Abort unfollow action on it." | ||
|
|
||
| fun getValidationPassedMessage(index: String) = "Unfollow action validation passed for [index=$index]" | ||
| } | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -167,6 +167,9 @@ | |
| } | ||
| } | ||
| }, | ||
| "unfollow": { | ||
| "type": "object" | ||
| }, | ||
| "delete": { | ||
| "type": "object" | ||
| }, | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I see we have ccr zip published https://aws.oss.sonatype.org/content/repositories/snapshots/org/opensearch/plugin/opensearch-cross-cluster-replication/3.0.0.0-SNAPSHOT/
can you follow this part
index-management/build.gradle
Lines 228 to 232 in dbd2bc2
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes thank you, I have followed this and will push this in the next commit.
However, for now, as my changes in ccr repo are still not merged, I cannot test the integration with the zip taken from repo, and am using the local zip itself.