[Feature] Support Transform as an ISM action (opensearch-project#760)

* Initial impl Signed-off-by: Tanqiu Liu <[email protected]> * fix style Signed-off-by: Tanqiu Liu <[email protected]> * end to end functional Signed-off-by: Tanqiu Liu <[email protected]> * ISM transform unit tests & integ tests Signed-off-by: Tanqiu Liu <[email protected]> * Fix after core #8157 (opensearch-project#857) Signed-off-by: bowenlan-amzn <[email protected]> * Upgrade the backport workflow (opensearch-project#862) Signed-off-by: Ashish Agrawal <[email protected]> Signed-off-by: Tanqiu Liu <[email protected]> * Added 2.9 release notes. (opensearch-project#851) * Added 2.9 release notes. Signed-off-by: AWSHurneyt <[email protected]> * Added 2.9 release notes. Signed-off-by: AWSHurneyt <[email protected]> * Added 2.9 release notes. Signed-off-by: AWSHurneyt <[email protected]> --------- Signed-off-by: AWSHurneyt <[email protected]> Signed-off-by: Tanqiu Liu <[email protected]> * Handle NPE in isRollupIndex (opensearch-project#855) * Handle NPE in isRollupIndex `metadata.index()` can return `null`, so handle that case by returning `false`. Signed-off-by: Bryce Lampe <[email protected]> * unit test Signed-off-by: Bryce Lampe <[email protected]> --------- Signed-off-by: Bryce Lampe <[email protected]> Co-authored-by: bowenlan-amzn <[email protected]> Signed-off-by: Tanqiu Liu <[email protected]> * Fix core XcontentType refactor (opensearch-project#873) Signed-off-by: Hailong Cui <[email protected]> Signed-off-by: Tanqiu Liu <[email protected]> * fix for max & min aggregations when no metric property exist (opensearch-project#870) Signed-off-by: Subhobrata Dey <[email protected]> Signed-off-by: Tanqiu Liu <[email protected]> * core refactor change (opensearch-project#884) Signed-off-by: Hailong Cui <[email protected]> Signed-off-by: Tanqiu Liu <[email protected]> * update backport branch name (opensearch-project#885) Signed-off-by: Hailong Cui <[email protected]> Signed-off-by: Tanqiu Liu <[email protected]> * core refactor change (opensearch-project#887) Signed-off-by: Hailong Cui <[email protected]> Signed-off-by: Tanqiu Liu <[email protected]> * Fix breaking change by core refactor (opensearch-project#888) Signed-off-by: Hailong Cui <[email protected]> Signed-off-by: Tanqiu Liu <[email protected]> * fix core breaking (opensearch-project#906) Signed-off-by: bowenlan-amzn <[email protected]> Signed-off-by: Tanqiu Liu <[email protected]> * Support copy alias in rollover (opensearch-project#907) * Support copy alias in rollover Signed-off-by: bowenlan-amzn <[email protected]> * 2.10 Signed-off-by: bowenlan-amzn <[email protected]> --------- Signed-off-by: bowenlan-amzn <[email protected]> Signed-off-by: Tanqiu Liu <[email protected]> * Set preference to _primary when searching control-center index (opensearch-project#911) * Set preference to _primary when searching control-center index Signed-off-by: gaobinlong <[email protected]> * Use _primary_first instead Signed-off-by: gaobinlong <[email protected]> --------- Signed-off-by: gaobinlong <[email protected]> Signed-off-by: Tanqiu Liu <[email protected]> * Add primary first preference to all search requests (opensearch-project#912) Signed-off-by: Tanqiu Liu <[email protected]> * fix intelliJ IDEA gradle sync error (opensearch-project#916) Signed-off-by: Hailong Cui <[email protected]> Signed-off-by: Tanqiu Liu <[email protected]> * make control center index as system index (opensearch-project#919) Signed-off-by: Hailong Cui <[email protected]> Signed-off-by: Tanqiu Liu <[email protected]> * Updates demo certs used in integ tests (opensearch-project#921) Signed-off-by: Darshit Chanpura <[email protected]> Signed-off-by: Tanqiu Liu <[email protected]> * Added 2.10 release notes (opensearch-project#925) Signed-off-by: Ashish Agrawal <[email protected]> Signed-off-by: Tanqiu Liu <[email protected]> * Bump bwc version (opensearch-project#930) Signed-off-by: bowenlan-amzn <[email protected]> Signed-off-by: Tanqiu Liu <[email protected]> * fix integ tests; upgrade mappings versions Signed-off-by: Tanqiu Liu <[email protected]> * Fix DCO Signed-off-by: Tanqiu Liu <[email protected]> * Addressed pr comments; Add integ test case for re-execute the same transform action Signed-off-by: Tanqiu Liu <[email protected]> * Addressed detekt error Signed-off-by: Tanqiu Liu <[email protected]> * Added ISMTransform writeable test Signed-off-by: Tanqiu Liu <[email protected]> * Addressed comments; Moved updateTransformStartTime to IndexManagementRestTestCase Signed-off-by: Tanqiu Liu <[email protected]> --------- Signed-off-by: Tanqiu Liu <[email protected]> Signed-off-by: bowenlan-amzn <[email protected]> Signed-off-by: Ashish Agrawal <[email protected]> Signed-off-by: AWSHurneyt <[email protected]> Signed-off-by: Bryce Lampe <[email protected]> Signed-off-by: Hailong Cui <[email protected]> Signed-off-by: Subhobrata Dey <[email protected]> Signed-off-by: gaobinlong <[email protected]> Signed-off-by: Darshit Chanpura <[email protected]> Co-authored-by: bowenlan-amzn <[email protected]> Co-authored-by: Ashish Agrawal <[email protected]> Co-authored-by: AWSHurneyt <[email protected]> Co-authored-by: Bryce Lampe <[email protected]> Co-authored-by: Hailong Cui <[email protected]> Co-authored-by: Subhobrata Dey <[email protected]> Co-authored-by: gaobinlong <[email protected]> Co-authored-by: Darshit Chanpura <[email protected]>
bowenlan-amzn · Oct 11, 2023 · f9e75c0 · f9e75c0
1 parent c20a502
commit f9e75c0
Show file tree

Hide file tree

Showing 28 changed files with 1,631 additions and 61 deletions.
diff --git a/DEVELOPER_GUIDE.md b/DEVELOPER_GUIDE.md
@@ -48,7 +48,7 @@ However, to build the `index management` plugin project, we also use the OpenSea
 
 ### Building from the command line
 
-1. `./gradlew build` builds and tests project.
+1. `./gradlew build` builds and tests project. 
 2. `./gradlew run` launches a single node cluster with the index management (and job-scheduler) plugin installed.
 3. `./gradlew run -PnumNodes=3` launches a multi-node cluster with the index management (and job-scheduler) plugin installed.
 4. `./gradlew integTest` launches a single node cluster with the index management (and job-scheduler) plugin installed and runs all integ tests.

diff --git a/.../kotlin/org.opensearch.indexmanagement.spi/indexstatemanagement/model/ActionProperties.kt b/.../kotlin/org.opensearch.indexmanagement.spi/indexstatemanagement/model/ActionProperties.kt
@@ -23,7 +23,8 @@ data class ActionProperties(
     val snapshotName: String? = null,
     val rollupId: String? = null,
     val hasRollupFailed: Boolean? = null,
-    val shrinkActionProperties: ShrinkActionProperties? = null
+    val shrinkActionProperties: ShrinkActionProperties? = null,
+    val transformActionProperties: TransformActionProperties? = null
 ) : Writeable, ToXContentFragment {
 
     override fun writeTo(out: StreamOutput) {
@@ -32,6 +33,7 @@ data class ActionProperties(
         out.writeOptionalString(rollupId)
         out.writeOptionalBoolean(hasRollupFailed)
         out.writeOptionalWriteable(shrinkActionProperties)
+        out.writeOptionalWriteable(transformActionProperties)
     }
 
     override fun toXContent(builder: XContentBuilder, params: ToXContent.Params): XContentBuilder {
@@ -40,6 +42,7 @@ data class ActionProperties(
         if (rollupId != null) builder.field(Properties.ROLLUP_ID.key, rollupId)
         if (hasRollupFailed != null) builder.field(Properties.HAS_ROLLUP_FAILED.key, hasRollupFailed)
         if (shrinkActionProperties != null) builder.addObject(ShrinkActionProperties.SHRINK_ACTION_PROPERTIES, shrinkActionProperties, params)
+        if (transformActionProperties != null) builder.addObject(TransformActionProperties.TRANSFORM_ACTION_PROPERTIES, transformActionProperties, params)
         return builder
     }
 
@@ -52,7 +55,8 @@ data class ActionProperties(
             val rollupId: String? = si.readOptionalString()
             val hasRollupFailed: Boolean? = si.readOptionalBoolean()
             val shrinkActionProperties: ShrinkActionProperties? = si.readOptionalWriteable { ShrinkActionProperties.fromStreamInput(it) }
-            return ActionProperties(maxNumSegments, snapshotName, rollupId, hasRollupFailed, shrinkActionProperties)
+            val transformActionProperties: TransformActionProperties? = si.readOptionalWriteable { TransformActionProperties.fromStreamInput(it) }
+            return ActionProperties(maxNumSegments, snapshotName, rollupId, hasRollupFailed, shrinkActionProperties, transformActionProperties)
         }
 
         fun parse(xcp: XContentParser): ActionProperties {
@@ -61,6 +65,7 @@ data class ActionProperties(
             var rollupId: String? = null
             var hasRollupFailed: Boolean? = null
             var shrinkActionProperties: ShrinkActionProperties? = null
+            var transformActionProperties: TransformActionProperties? = null
 
             ensureExpectedToken(Token.START_OBJECT, xcp.currentToken(), xcp)
             while (xcp.nextToken() != Token.END_OBJECT) {
@@ -75,10 +80,13 @@ data class ActionProperties(
                     ShrinkActionProperties.SHRINK_ACTION_PROPERTIES -> {
                         shrinkActionProperties = if (xcp.currentToken() == Token.VALUE_NULL) null else ShrinkActionProperties.parse(xcp)
                     }
+                    TransformActionProperties.TRANSFORM_ACTION_PROPERTIES -> {
+                        transformActionProperties = if (xcp.currentToken() == Token.VALUE_NULL) null else TransformActionProperties.parse(xcp)
+                    }
                 }
             }
 
-            return ActionProperties(maxNumSegments, snapshotName, rollupId, hasRollupFailed, shrinkActionProperties)
+            return ActionProperties(maxNumSegments, snapshotName, rollupId, hasRollupFailed, shrinkActionProperties, transformActionProperties)
         }
     }
 

diff --git a/...rg.opensearch.indexmanagement.spi/indexstatemanagement/model/TransformActionProperties.kt b/...rg.opensearch.indexmanagement.spi/indexstatemanagement/model/TransformActionProperties.kt
@@ -0,0 +1,58 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.indexmanagement.spi.indexstatemanagement.model
+
+import org.opensearch.core.common.io.stream.StreamInput
+import org.opensearch.core.common.io.stream.StreamOutput
+import org.opensearch.core.common.io.stream.Writeable
+import org.opensearch.core.xcontent.XContentParserUtils.ensureExpectedToken
+import org.opensearch.core.xcontent.ToXContent
+import org.opensearch.core.xcontent.ToXContentFragment
+import org.opensearch.core.xcontent.XContentBuilder
+import org.opensearch.core.xcontent.XContentParser
+
+data class TransformActionProperties(
+    val transformId: String?
+) : Writeable, ToXContentFragment {
+
+    override fun writeTo(out: StreamOutput) {
+        out.writeOptionalString(transformId)
+    }
+
+    override fun toXContent(builder: XContentBuilder, params: ToXContent.Params?): XContentBuilder {
+        if (transformId != null) builder.field(Properties.TRANSFORM_ID.key, transformId)
+        return builder
+    }
+
+    companion object {
+        const val TRANSFORM_ACTION_PROPERTIES = "transform_action_properties"
+
+        fun fromStreamInput(sin: StreamInput): TransformActionProperties {
+            val transformId: String? = sin.readOptionalString()
+            return TransformActionProperties(transformId)
+        }
+
+        fun parse(xcp: XContentParser): TransformActionProperties {
+            var transformId: String? = null
+
+            ensureExpectedToken(XContentParser.Token.START_OBJECT, xcp.currentToken(), xcp)
+            while (xcp.nextToken() != XContentParser.Token.END_OBJECT) {
+                val fieldName = xcp.currentName()
+                xcp.nextToken()
+
+                when (fieldName) {
+                    Properties.TRANSFORM_ID.key -> transformId = xcp.text()
+                }
+            }
+
+            return TransformActionProperties(transformId)
+        }
+    }
+
+    enum class Properties(val key: String) {
+        TRANSFORM_ID("transform_id")
+    }
+}
diff --git a/src/main/kotlin/org/opensearch/indexmanagement/IndexManagementPlugin.kt b/src/main/kotlin/org/opensearch/indexmanagement/IndexManagementPlugin.kt
@@ -460,7 +460,7 @@ class IndexManagementPlugin : JobSchedulerExtension, NetworkPlugin, ActionPlugin
 
         val managedIndexCoordinator = ManagedIndexCoordinator(
             environment.settings(),
-            client, clusterService, threadPool, indexManagementIndices, metadataService, templateService, indexMetadataProvider
+            client, clusterService, threadPool, indexManagementIndices, metadataService, templateService, indexMetadataProvider, xContentRegistry
         )
 
         val smRunner = SMRunner.init(client, threadPool, settings, indexManagementIndices, clusterService)

diff --git a/src/main/kotlin/org/opensearch/indexmanagement/indexstatemanagement/ISMActionsParser.kt b/src/main/kotlin/org/opensearch/indexmanagement/indexstatemanagement/ISMActionsParser.kt
@@ -23,6 +23,7 @@ import org.opensearch.indexmanagement.indexstatemanagement.action.RolloverAction
 import org.opensearch.indexmanagement.indexstatemanagement.action.RollupActionParser
 import org.opensearch.indexmanagement.indexstatemanagement.action.ShrinkActionParser
 import org.opensearch.indexmanagement.indexstatemanagement.action.SnapshotActionParser
+import org.opensearch.indexmanagement.indexstatemanagement.action.TransformActionParser
 import org.opensearch.indexmanagement.spi.indexstatemanagement.Action
 import org.opensearch.indexmanagement.spi.indexstatemanagement.ActionParser
 import org.opensearch.indexmanagement.spi.indexstatemanagement.model.ActionRetry
@@ -49,7 +50,8 @@ class ISMActionsParser private constructor() {
         RollupActionParser(),
         RolloverActionParser(),
         ShrinkActionParser(),
-        SnapshotActionParser()
+        SnapshotActionParser(),
+        TransformActionParser(),
     )
 
     val customActionExtensionMap = mutableMapOf<String, String>()

diff --git a/...ain/kotlin/org/opensearch/indexmanagement/indexstatemanagement/ManagedIndexCoordinator.kt b/...ain/kotlin/org/opensearch/indexmanagement/indexstatemanagement/ManagedIndexCoordinator.kt
@@ -42,6 +42,7 @@ import org.opensearch.common.regex.Regex
 import org.opensearch.common.settings.Settings
 import org.opensearch.common.unit.TimeValue
 import org.opensearch.commons.authuser.User
+import org.opensearch.core.xcontent.NamedXContentRegistry
 import org.opensearch.core.index.Index
 import org.opensearch.index.query.QueryBuilders
 import org.opensearch.indexmanagement.IndexManagementIndices
@@ -117,7 +118,8 @@ class ManagedIndexCoordinator(
     indexManagementIndices: IndexManagementIndices,
     private val metadataService: MetadataService,
     private val templateService: ISMTemplateService,
-    private val indexMetadataProvider: IndexMetadataProvider
+    private val indexMetadataProvider: IndexMetadataProvider,
+    private val xContentRegistry: NamedXContentRegistry
 ) : ClusterStateListener,
     CoroutineScope by CoroutineScope(SupervisorJob() + Dispatchers.Default + CoroutineName("ManagedIndexCoordinator")),
     LifecycleListener() {
@@ -462,7 +464,7 @@ class ManagedIndexCoordinator(
 
         return try {
             val response: SearchResponse = client.suspendUntil { search(searchRequest, it) }
-            parseFromSearchResponse(response = response, parse = Policy.Companion::parse)
+            parseFromSearchResponse(response, xContentRegistry, Policy.Companion::parse)
         } catch (ex: IndexNotFoundException) {
             emptyList()
         } catch (ex: ClusterBlockException) {
@@ -713,7 +715,7 @@ class ManagedIndexCoordinator(
         }
         mRes.forEach {
             if (it.response.isExists) {
-                result[it.id] = contentParser(it.response.sourceAsBytesRef).parseWithType(
+                result[it.id] = contentParser(it.response.sourceAsBytesRef, xContentRegistry).parseWithType(
                     it.response.id, it.response.seqNo, it.response.primaryTerm, ManagedIndexConfig.Companion::parse
                 )
             }

diff --git a/...main/kotlin/org/opensearch/indexmanagement/indexstatemanagement/action/TransformAction.kt b/...main/kotlin/org/opensearch/indexmanagement/indexstatemanagement/action/TransformAction.kt
@@ -0,0 +1,63 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.indexmanagement.indexstatemanagement.action
+
+import org.opensearch.core.common.io.stream.StreamOutput
+import org.opensearch.core.xcontent.ToXContent
+import org.opensearch.core.xcontent.XContentBuilder
+import org.opensearch.indexmanagement.indexstatemanagement.step.transform.AttemptCreateTransformJobStep
+import org.opensearch.indexmanagement.indexstatemanagement.step.transform.WaitForTransformCompletionStep
+import org.opensearch.indexmanagement.spi.indexstatemanagement.Action
+import org.opensearch.indexmanagement.spi.indexstatemanagement.Step
+import org.opensearch.indexmanagement.spi.indexstatemanagement.model.StepContext
+import org.opensearch.indexmanagement.transform.model.ISMTransform
+
+class TransformAction(
+    val ismTransform: ISMTransform,
+    index: Int
+) : Action(name, index) {
+
+    companion object {
+        const val name = "transform"
+        const val ISM_TRANSFORM_FIELD = "ism_transform"
+    }
+
+    private val attemptCreateTransformJobStep = AttemptCreateTransformJobStep(this)
+    private val waitForTransformCompletionStep = WaitForTransformCompletionStep()
+    private val steps = listOf(attemptCreateTransformJobStep, waitForTransformCompletionStep)
+
+    @Suppress("ReturnCount")
+    override fun getStepToExecute(context: StepContext): Step {
+        // if stepMetaData is null, return first step
+        val stepMetaData = context.metadata.stepMetaData ?: return attemptCreateTransformJobStep
+
+        // if the current step has completed, return the next step
+        if (stepMetaData.stepStatus == Step.StepStatus.COMPLETED) {
+            return when (stepMetaData.name) {
+                AttemptCreateTransformJobStep.name -> waitForTransformCompletionStep
+                else -> attemptCreateTransformJobStep
+            }
+        }
+
+        return when (stepMetaData.name) {
+            AttemptCreateTransformJobStep.name -> attemptCreateTransformJobStep
+            else -> waitForTransformCompletionStep
+        }
+    }
+
+    override fun getSteps(): List<Step> = steps
+
+    override fun populateAction(builder: XContentBuilder, params: ToXContent.Params) {
+        builder.startObject(type)
+        builder.field(ISM_TRANSFORM_FIELD, ismTransform)
+        builder.endObject()
+    }
+
+    override fun populateAction(out: StreamOutput) {
+        ismTransform.writeTo(out)
+        out.writeInt(actionIndex)
+    }
+}
diff --git a/...otlin/org/opensearch/indexmanagement/indexstatemanagement/action/TransformActionParser.kt b/...otlin/org/opensearch/indexmanagement/indexstatemanagement/action/TransformActionParser.kt
@@ -0,0 +1,42 @@
+/*
+ * 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
+import org.opensearch.indexmanagement.spi.indexstatemanagement.Action
+import org.opensearch.indexmanagement.spi.indexstatemanagement.ActionParser
+import org.opensearch.indexmanagement.transform.model.ISMTransform
+
+class TransformActionParser : ActionParser() {
+    override fun fromStreamInput(sin: StreamInput): Action {
+        val ismTransform = ISMTransform(sin)
+        val index = sin.readInt()
+        return TransformAction(ismTransform, index)
+    }
+
+    override fun fromXContent(xcp: XContentParser, index: Int): Action {
+        var ismTransform: ISMTransform? = null
+
+        XContentParserUtils.ensureExpectedToken(XContentParser.Token.START_OBJECT, xcp.currentToken(), xcp)
+        while (xcp.nextToken() != XContentParser.Token.END_OBJECT) {
+            val fieldName = xcp.currentName()
+            xcp.nextToken()
+
+            when (fieldName) {
+                TransformAction.ISM_TRANSFORM_FIELD -> ismTransform = ISMTransform.parse(xcp)
+                else -> throw IllegalArgumentException("Invalid field: [$fieldName] found in TransformAction.")
+            }
+        }
+
+        return TransformAction(ismTransform = requireNotNull(ismTransform) { "TransformAction transform is null." }, index)
+    }
+
+    override fun getActionType(): String {
+        return TransformAction.name
+    }
+}