-
Notifications
You must be signed in to change notification settings - Fork 116
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[2.x] Adding new stop-replication action in ism
Signed-off-by: aggarwalShivani <shivani.aggarwal@nokia.com>
- Loading branch information
1 parent
efb8cb6
commit af906fc
Showing
12 changed files
with
505 additions
and
3 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
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
30 changes: 30 additions & 0 deletions
30
...otlin/org/opensearch/indexmanagement/indexstatemanagement/action/StopReplicationAction.kt
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,30 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.indexmanagement.indexstatemanagement.action | ||
|
|
||
| import org.opensearch.indexmanagement.indexstatemanagement.step.stopreplication.AttemptStopReplicationStep | ||
| 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 StopReplicationAction( | ||
| index: Int, | ||
| ) : Action(name, index) { | ||
| companion object { | ||
| const val name = "stop_replication" | ||
| } | ||
|
|
||
| private val attemptStopReplicationStep = AttemptStopReplicationStep() | ||
|
|
||
| private val steps = listOf(attemptStopReplicationStep) | ||
|
|
||
| override fun getStepToExecute(context: StepContext): Step = attemptStopReplicationStep | ||
|
|
||
| override fun getSteps(): List<Step> = steps | ||
| } |
28 changes: 28 additions & 0 deletions
28
...org/opensearch/indexmanagement/indexstatemanagement/action/StopReplicationActionParser.kt
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,28 @@ | ||
| /* | ||
| * 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 StopReplicationActionParser : ActionParser() { | ||
| override fun fromStreamInput(sin: StreamInput): Action { | ||
| val index = sin.readInt() | ||
| return StopReplicationAction(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 StopReplicationAction(index) | ||
| } | ||
|
|
||
| override fun getActionType(): String = StopReplicationAction.name | ||
| } |
99 changes: 99 additions & 0 deletions
99
...h/indexmanagement/indexstatemanagement/step/stopreplication/AttemptStopReplicationStep.kt
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,99 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.indexmanagement.indexstatemanagement.step.stopreplication | ||
|
|
||
| import org.apache.logging.log4j.LogManager | ||
| import org.opensearch.ExceptionsHelper | ||
| import org.opensearch.action.support.master.AcknowledgedResponse | ||
| 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 AttemptStopReplicationStep : Step(name) { | ||
| private val logger = LogManager.getLogger(javaClass) | ||
| private var stepStatus = StepStatus.STARTING | ||
| private var info: Map<String, Any>? = null | ||
| private var replicationPluginInterface: ReplicationPluginInterface = ReplicationPluginInterface() | ||
| fun setReplicationPluginInterface(replicationPluginInterface: ReplicationPluginInterface) { | ||
| this.replicationPluginInterface = replicationPluginInterface | ||
| } | ||
|
|
||
| override suspend fun execute(): Step { | ||
| val context = this.context ?: return this | ||
| val indexName = context.metadata.index | ||
| try { | ||
| val stopIndexReplicationRequestObj = StopIndexReplicationRequest(indexName) | ||
| val response: AcknowledgedResponse = context.client.admin().indices().suspendUntil { | ||
| replicationPluginInterface.stopReplication( | ||
| context.client, | ||
| stopIndexReplicationRequestObj, | ||
| it, | ||
| ) | ||
| } | ||
| 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 | ||
| } | ||
|
|
||
| private fun handleSnapshotException(indexName: String, e: SnapshotInProgressException) { | ||
| val message = getSnapshotMessage(indexName) | ||
| logger.error(message, e) | ||
| stepStatus = StepStatus.FAILED | ||
| 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 = currentMetadata.copy( | ||
| stepMetaData = StepMetaData(name, getStepStartTime(currentMetadata).toEpochMilli(), stepStatus), | ||
| transitionTo = null, | ||
| info = info, | ||
| ) | ||
|
|
||
| override fun isIdempotent() = false | ||
|
|
||
| companion object { | ||
| const val name = "attempt_stop_replication" | ||
|
|
||
| fun getFailedMessage(index: String) = "Failed to stop replication [index=$index]" | ||
|
|
||
| fun getSuccessMessage(index: String) = "Successfully stopped replication [index=$index]" | ||
|
|
||
| fun getSnapshotMessage(index: String) = "Index had snapshot in progress, retrying stop replication [index=$index]" | ||
| } | ||
| } |
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
72 changes: 72 additions & 0 deletions
72
...org/opensearch/indexmanagement/indexstatemanagement/validation/ValidateStopReplication.kt
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,72 @@ | ||
| /* | ||
| * 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 ValidateStopReplication( | ||
| 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 stop_replication 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_stop_replication" | ||
|
|
||
| fun getNoIndexMessage(index: String) = "No such index [index=$index] for stop replication action." | ||
|
|
||
| fun getIndexNotValidMessage(index: String) = "Index [index=$index] is not valid. Abort stop replication action on it." | ||
|
|
||
| fun getValidationPassedMessage(index: String) = "Stop replication action validation passed for [index=$index]" | ||
| } | ||
| } |
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
57 changes: 57 additions & 0 deletions
57
...lin/org/opensearch/indexmanagement/indexstatemanagement/action/StopReplicationActionIT.kt
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,57 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.indexmanagement.indexstatemanagement.action | ||
|
|
||
| import org.opensearch.indexmanagement.indexstatemanagement.IndexStateManagementRestTestCase | ||
| import org.opensearch.indexmanagement.indexstatemanagement.model.Policy | ||
| import org.opensearch.indexmanagement.indexstatemanagement.model.State | ||
| import org.opensearch.indexmanagement.indexstatemanagement.randomErrorNotification | ||
| import org.opensearch.indexmanagement.waitFor | ||
| import java.time.Instant | ||
| import java.time.temporal.ChronoUnit | ||
| import java.util.Locale | ||
|
|
||
| class StopReplicationActionIT : IndexStateManagementRestTestCase() { | ||
| private val testIndexName = javaClass.simpleName.lowercase(Locale.ROOT) | ||
|
|
||
| fun `test stop_replication on a non-replicated index`() { | ||
| val indexName = "${testIndexName}_index_1" | ||
| val policyID = "${testIndexName}_testPolicyName_1" | ||
| val actionConfig = StopReplicationAction(0) | ||
| val states = | ||
| listOf( | ||
| State("StopReplicationState", listOf(actionConfig), listOf()), | ||
| ) | ||
|
|
||
| val policy = | ||
| Policy( | ||
| id = policyID, | ||
| description = "$testIndexName description", | ||
| schemaVersion = 1L, | ||
| lastUpdatedTime = Instant.now().truncatedTo(ChronoUnit.MILLIS), | ||
| errorNotification = randomErrorNotification(), | ||
| defaultState = states[0].name, | ||
| states = states, | ||
| ) | ||
| createPolicy(policy, policyID) | ||
| createIndex(indexName, policyID) | ||
| val managedIndexConfig = getExistingManagedIndexConfig(indexName) | ||
| // Change the start time so the job will trigger in 2 seconds. | ||
| updateManagedIndexConfigStartTime(managedIndexConfig) | ||
|
|
||
| waitFor { assertEquals(policyID, getExplainManagedIndexMetaData(indexName).policyID) } | ||
|
|
||
| // Need to wait two cycles. | ||
| // Change the start time so the job will trigger in 2 seconds. | ||
| updateManagedIndexConfigStartTime(managedIndexConfig) | ||
| waitFor { | ||
| val metadataInfo = getExplainManagedIndexMetaData(indexName).info.toString() | ||
| assertTrue( | ||
| metadataInfo.contains("cause=No replication in progress for index:" + indexName), | ||
| ) | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.