Add support for source index in ISM Policy Rollup Action to help perform multi-tier rollups

Signed-off-by: Kshitij Tandon <[email protected]>

Author: tandonks

src/main/kotlin/org/opensearch/indexmanagement/indexstatemanagement/step/rollup/AttemptCreateRollupJobStep.kt

@@ -18,6 +18,7 @@ import org.opensearch.indexmanagement.rollup.action.index.IndexRollupRequest
 import org.opensearch.indexmanagement.rollup.action.index.IndexRollupResponse
 import org.opensearch.indexmanagement.rollup.action.start.StartRollupAction
 import org.opensearch.indexmanagement.rollup.action.start.StartRollupRequest
+import org.opensearch.indexmanagement.rollup.util.RollupFieldValueExpressionResolver
 import org.opensearch.indexmanagement.spi.indexstatemanagement.Step
 import org.opensearch.indexmanagement.spi.indexstatemanagement.model.ActionProperties
 import org.opensearch.indexmanagement.spi.indexstatemanagement.model.ManagedIndexMetaData
@@ -39,39 +40,110 @@ class AttemptCreateRollupJobStep(private val action: RollupAction) : Step(name)
         val previousRunRollupId = managedIndexMetadata.actionMetaData?.actionProperties?.rollupId
         val hasPreviousRollupAttemptFailed = managedIndexMetadata.actionMetaData?.actionProperties?.hasRollupFailed
 
-        // Creating a rollup job
-        val rollup = action.ismRollup.toRollup(indexName, context.user)
-        rollupId = rollup.id
-        logger.info("Attempting to create a rollup job $rollupId for index $indexName")
-
-        val indexRollupRequest = IndexRollupRequest(rollup, WriteRequest.RefreshPolicy.IMMEDIATE)
-
         try {
+            // Create a temporary rollup object for template resolution context.
+            // This provides the rollup's source_index as {{ctx.source_index}} in templates.
+            val tempRollup = action.ismRollup.toRollup(indexName, context.user)
+
+            // Resolve source_index template if provided, else use managed index name.
+            // This enables patterns like:
+            // - source_index: "{{ctx.index}}" -> resolves to the managed index name
+            // - source_index: null -> defaults to the managed index name (backward compatible)
+            val resolvedSourceIndex = if (action.ismRollup.sourceIndex != null) {
+                RollupFieldValueExpressionResolver.resolve(
+                    tempRollup,
+                    action.ismRollup.sourceIndex,
+                    indexName,
+                )
+            } else {
+                indexName
+            }
+
+            // Resolve target_index template.
+            // Common patterns:
+            // - "rollup_{{ctx.index}}" -> "rollup_logs-2024-01"
+            // - "rollup_tier2_{{ctx.index}}" -> "rollup_tier2_rollup_tier1_logs-2024-01"
+            val resolvedTargetIndex = RollupFieldValueExpressionResolver.resolve(
+                tempRollup,
+                action.ismRollup.targetIndex,
+                indexName,
+            )
+
+            // Validate resolved indices to ensure they are valid and different.
+            // This catches configuration errors early before attempting to create the rollup job.
+            validateResolvedIndices(resolvedSourceIndex, resolvedTargetIndex)
+
+            logger.info(
+                "Executing rollup from source [$resolvedSourceIndex] to target [$resolvedTargetIndex] " +
+                    "for managed index [$indexName]",
+            )
+
+            // Create the final rollup job with resolved source_index and target_index.
+            // The copy() ensures that template variables are replaced with actual index names
+            // before the rollup job is persisted and executed.
+            val rollup = action.ismRollup.toRollup(indexName, context.user)
+                .copy(sourceIndex = resolvedSourceIndex, targetIndex = resolvedTargetIndex)
+            rollupId = rollup.id
+            logger.info("Attempting to create a rollup job $rollupId for index $indexName")
+
+            val indexRollupRequest = IndexRollupRequest(rollup, WriteRequest.RefreshPolicy.IMMEDIATE)
             val response: IndexRollupResponse = context.client.suspendUntil { execute(IndexRollupAction.INSTANCE, indexRollupRequest, it) }
             logger.info("Received status ${response.status.status} on trying to create rollup job $rollupId")
 
             stepStatus = StepStatus.COMPLETED
             info = mapOf("message" to getSuccessMessage(rollup.id, indexName))
+        } catch (e: IllegalArgumentException) {
+            val message = "Failed to validate resolved indices for rollup job"
+            logger.error(message, e)
+            stepStatus = StepStatus.FAILED
+            info = mapOf("message" to message, "cause" to "${e.message}")
         } catch (e: VersionConflictEngineException) {
-            val message = getFailedJobExistsMessage(rollup.id, indexName)
+            val message = getFailedJobExistsMessage(rollupId ?: "unknown", indexName)
             logger.info(message)
             if (rollupId == previousRunRollupId && hasPreviousRollupAttemptFailed == true) {
-                startRollupJob(rollup.id, context)
+                startRollupJob(rollupId ?: "unknown", context)
             } else {
                 stepStatus = StepStatus.COMPLETED
                 info = mapOf("info" to message)
             }
         } catch (e: RemoteTransportException) {
-            processFailure(rollup.id, indexName, ExceptionsHelper.unwrapCause(e) as Exception)
+            processFailure(rollupId ?: "unknown", indexName, ExceptionsHelper.unwrapCause(e) as Exception)
         } catch (e: OpenSearchException) {
-            processFailure(rollup.id, indexName, e)
+            processFailure(rollupId ?: "unknown", indexName, e)
         } catch (e: Exception) {
-            processFailure(rollup.id, indexName, e)
+            val message = "Failed to create rollup job"
+            logger.error(message, e)
+            stepStatus = StepStatus.FAILED
+            info = mapOf("message" to message, "cause" to "${e.message}")
         }
 
         return this
     }
 
+    /**
+     * Validates that resolved source and target indices are valid and different.
+     *
+     * This validation ensures that:
+     * 1. The resolved source_index is not empty or whitespace-only
+     * 2. The resolved target_index is not empty or whitespace-only
+     * 3. The source and target indices are different (prevents self-rollup)
+     *
+     * @param sourceIndex The resolved source index name (after template resolution)
+     * @param targetIndex The resolved target index name (after template resolution)
+     * @throws IllegalArgumentException if any validation rule fails, with a descriptive error message
+     */
+    private fun validateResolvedIndices(sourceIndex: String, targetIndex: String) {
+        require(sourceIndex.isNotBlank()) {
+            "Resolved source_index cannot be empty"
+        }
+        require(targetIndex.isNotBlank()) {
+            "Resolved target_index cannot be empty"
+        }
+        require(sourceIndex != targetIndex) {
+            "Source and target indices must be different: $sourceIndex"
+        }
+    }
+
     fun processFailure(rollupId: String, indexName: String, e: Exception) {
         val message = getFailedMessage(rollupId, indexName)
         logger.error(message, e)

src/main/kotlin/org/opensearch/indexmanagement/rollup/model/ISMRollup.kt

@@ -35,6 +35,7 @@ data class ISMRollup(
     val pageSize: Int,
     val dimensions: List<Dimension>,
     val metrics: List<RollupMetrics>,
+    val sourceIndex: String? = null,
 ) : ToXContentObject,
     Writeable {
     // TODO: This can be moved to a common place, since this is shared between Rollup and ISMRollup
@@ -58,6 +59,9 @@ data class ISMRollup(
             .field(Rollup.PAGE_SIZE_FIELD, pageSize)
             .field(Rollup.DIMENSIONS_FIELD, dimensions)
             .field(Rollup.METRICS_FIELD, metrics)
+        if (sourceIndex != null) {
+            builder.field(Rollup.SOURCE_INDEX_FIELD, sourceIndex)
+        }
         if (targetIndexSettings != null) {
             builder.startObject(Rollup.TARGET_INDEX_SETTINGS_FIELD)
             targetIndexSettings.toXContent(builder, params)
@@ -67,8 +71,20 @@ data class ISMRollup(
         return builder
     }
 
+    /**
+     * Converts ISMRollup configuration to a Rollup job.
+     *
+     * Source index resolution priority:
+     * 1. If ISMRollup.sourceIndex is set, use it (for multi-tier rollups)
+     * 2. Otherwise, use the sourceIndex parameter (managed index from ISM context)
+     *
+     * @param sourceIndex The managed index from ISM context (fallback if ISMRollup.sourceIndex is null)
+     * @param user Optional user context for the rollup job
+     * @return Rollup job configuration
+     */
     fun toRollup(sourceIndex: String, user: User? = null): Rollup {
-        val id = sourceIndex + toString()
+        val resolvedSourceIndex = this.sourceIndex ?: sourceIndex
+        val id = resolvedSourceIndex + toString()
         val currentTime = Instant.now()
         return Rollup(
             id = DigestUtils.sha1Hex(id),
@@ -80,7 +96,7 @@ data class ISMRollup(
             jobLastUpdatedTime = currentTime,
             jobEnabledTime = currentTime,
             description = this.description,
-            sourceIndex = sourceIndex,
+            sourceIndex = resolvedSourceIndex,
             targetIndex = this.targetIndex,
             targetIndexSettings = this.targetIndexSettings,
             metadataID = null,
@@ -120,6 +136,11 @@ data class ISMRollup(
             dimensionsList.toList()
         },
         metrics = sin.readList(::RollupMetrics),
+        sourceIndex = if (sin.version.onOrAfter(Version.V_3_0_0) && sin.readBoolean()) {
+            sin.readString()
+        } else {
+            null
+        },
     )
 
     override fun toString(): String {
@@ -159,6 +180,10 @@ data class ISMRollup(
             }
         }
         out.writeCollection(metrics)
+        if (out.version.onOrAfter(Version.V_3_0_0)) {
+            out.writeBoolean(sourceIndex != null)
+            if (sourceIndex != null) out.writeString(sourceIndex)
+        }
     }
 
     companion object {
@@ -170,6 +195,7 @@ data class ISMRollup(
         ): ISMRollup {
             var description = ""
             var targetIndex = ""
+            var sourceIndex: String? = null
             var targetIndexSettings: Settings? = null
             var pageSize = 0
             val dimensions = mutableListOf<Dimension>()
@@ -184,6 +210,7 @@ data class ISMRollup(
                 when (fieldName) {
                     Rollup.DESCRIPTION_FIELD -> description = xcp.text()
                     Rollup.TARGET_INDEX_FIELD -> targetIndex = xcp.text()
+                    Rollup.SOURCE_INDEX_FIELD -> sourceIndex = xcp.text()
                     Rollup.TARGET_INDEX_SETTINGS_FIELD -> {
                         XContentParserUtils.ensureExpectedToken(
                             XContentParser.Token.START_OBJECT,
@@ -224,6 +251,7 @@ data class ISMRollup(
                 metrics = metrics,
                 targetIndex = targetIndex,
                 targetIndexSettings = targetIndexSettings,
+                sourceIndex = sourceIndex,
             )
         }
     }

src/main/kotlin/org/opensearch/indexmanagement/rollup/util/RollupFieldValueExpressionResolver.kt

@@ -24,6 +24,15 @@ object RollupFieldValueExpressionResolver {
     private lateinit var clusterService: ClusterService
     lateinit var indexAliasUtils: IndexAliasUtils
 
+    /**
+     * Resolves template variables in a field value using the rollup object as context.
+     * This method is kept for backward compatibility and only supports {{ctx.source_index}}.
+     *
+     * @param rollup The rollup object providing context for template resolution
+     * @param fieldValue The field value that may contain template variables (e.g., "{{ctx.source_index}}")
+     * @return The resolved field value with variables replaced, or the original value if resolution
+     *         produces an empty/null result
+     */
     fun resolve(rollup: Rollup, fieldValue: String): String {
         val script = Script(ScriptType.INLINE, Script.DEFAULT_TEMPLATE_LANG, fieldValue, mapOf())
 
@@ -44,6 +53,48 @@ object RollupFieldValueExpressionResolver {
         return if (compiledValue.isNullOrBlank()) fieldValue else compiledValue
     }
 
+    /**
+     * Resolves template variables in a field value using the rollup object and managed index name as context.
+     * This is the primary method used by ISM rollup actions to resolve source_index and target_index templates.
+     *
+     * This method extends the basic resolve() by adding the managed index name to the template context,
+     * enabling the use of {{ctx.index}} in addition to {{ctx.source_index}}.
+     *
+     * @param rollup The rollup object providing context for template resolution. The rollup's source_index
+     *               field is made available as {{ctx.source_index}} in templates.
+     * @param fieldValue The field value that may contain Mustache template variables. Common patterns:
+     *                   - "{{ctx.index}}" - resolves to the managed index name
+     *                   - "{{ctx.source_index}}" - resolves to the rollup's source index
+     *                   - "rollup_{{ctx.index}}" - adds a prefix to the managed index name
+     *                   - Literal values without templates are returned unchanged
+     * @param managedIndexName The name of the index being managed by ISM. This is the index to which
+     *                         the ISM policy is applied, and it's made available as {{ctx.index}} in templates.
+     *                         For multi-tier rollups, this would be the output of the previous tier.
+     * @return The resolved field value with all template variables replaced. If the resolved value is an alias,
+     *         it's automatically resolved to the write index name. If resolution produces an empty or null result,
+     *         the original fieldValue is returned unchanged.
+     */
+    fun resolve(rollup: Rollup, fieldValue: String, managedIndexName: String): String {
+        val script = Script(ScriptType.INLINE, Script.DEFAULT_TEMPLATE_LANG, fieldValue, mapOf())
+
+        val contextMap =
+            rollup.toXContent(XContentFactory.jsonBuilder(), XCONTENT_WITHOUT_TYPE)
+                .toMap()
+                .filterKeys { key -> key in validTopContextFields }
+                .plus("index" to managedIndexName)
+
+        var compiledValue =
+            scriptService.compile(script, TemplateScript.CONTEXT)
+                .newInstance(script.params + mapOf("ctx" to contextMap))
+                .execute()
+
+        if (indexAliasUtils.isAlias(compiledValue)) {
+            compiledValue = indexAliasUtils.getWriteIndexNameForAlias(compiledValue)
+        }
+
+        return if (compiledValue.isNullOrBlank()) fieldValue else compiledValue
+    }
+
     fun registerServices(scriptService: ScriptService, clusterService: ClusterService) {
         this.scriptService = scriptService
         this.clusterService = clusterService

src/main/resources/mappings/opendistro-ism-config.json

@@ -1,6 +1,6 @@
 {
   "_meta" : {
-    "schema_version": 24
+    "schema_version": 25
   },
   "dynamic": "strict",
   "properties": {
@@ -374,6 +374,15 @@
                         "description": {
                           "type": "text"
                         },
+                        "source_index": {
+                          "type": "text",
+                          "fields": {
+                            "keyword": {
+                              "type": "keyword",
+                              "ignore_above": 256
+                            }
+                          }
+                        },
                         "target_index": {
                           "type": "text",
                           "fields": {