Added min_state_age condition check for ISM

Signed-off-by: Tarun-kishore <[email protected]>

Author: Tarun Kishore

src/main/kotlin/org/opensearch/indexmanagement/indexstatemanagement/model/Transition.kt

@@ -81,10 +81,11 @@ data class Conditions(
     val cron: CronSchedule? = null,
     val rolloverAge: TimeValue? = null,
     val noAlias: Boolean? = null,
+    val minStateAge: TimeValue? = null,
 ) : ToXContentObject,
     Writeable {
     init {
-        val conditionsList = listOf(indexAge, docCount, size, cron, rolloverAge, noAlias)
+        val conditionsList = listOf(indexAge, docCount, size, cron, rolloverAge, noAlias, minStateAge)
         require(conditionsList.filterNotNull().size == 1) { "Cannot provide more than one Transition condition" }
 
         // Validate doc count condition
@@ -102,6 +103,7 @@ data class Conditions(
         if (cron != null) builder.field(CRON_FIELD, cron)
         if (rolloverAge != null) builder.field(MIN_ROLLOVER_AGE_FIELD, rolloverAge.stringRep)
         if (noAlias != null) builder.field(NO_ALIAS_FIELD, noAlias)
+        if (minStateAge != null) builder.field(MIN_STATE_AGE_FIELD, minStateAge.stringRep)
         return builder.endObject()
     }
 
@@ -113,6 +115,7 @@ data class Conditions(
         cron = sin.readOptionalWriteable(::CronSchedule),
         rolloverAge = sin.readOptionalTimeValue(),
         noAlias = sin.readOptionalBoolean(),
+        minStateAge = sin.readOptionalTimeValue(),
     )
 
     @Throws(IOException::class)
@@ -123,6 +126,7 @@ data class Conditions(
         out.writeOptionalWriteable(cron)
         out.writeOptionalTimeValue(rolloverAge)
         out.writeOptionalBoolean(noAlias)
+        out.writeOptionalTimeValue(minStateAge)
     }
 
     companion object {
@@ -132,6 +136,7 @@ data class Conditions(
         const val CRON_FIELD = "cron"
         const val MIN_ROLLOVER_AGE_FIELD = "min_rollover_age"
         const val NO_ALIAS_FIELD = "no_alias"
+        const val MIN_STATE_AGE_FIELD = "min_state_age"
 
         @JvmStatic
         @Throws(IOException::class)
@@ -142,6 +147,7 @@ data class Conditions(
             var cron: CronSchedule? = null
             var rolloverAge: TimeValue? = null
             var noAlias: Boolean? = null
+            var minStateAge: TimeValue? = null
 
             ensureExpectedToken(Token.START_OBJECT, xcp.currentToken(), xcp)
             while (xcp.nextToken() != Token.END_OBJECT) {
@@ -155,11 +161,12 @@ data class Conditions(
                     CRON_FIELD -> cron = ScheduleParser.parse(xcp) as? CronSchedule
                     MIN_ROLLOVER_AGE_FIELD -> rolloverAge = TimeValue.parseTimeValue(xcp.text(), MIN_ROLLOVER_AGE_FIELD)
                     NO_ALIAS_FIELD -> noAlias = xcp.booleanValue()
+                    MIN_STATE_AGE_FIELD -> minStateAge = TimeValue.parseTimeValue(xcp.text(), MIN_STATE_AGE_FIELD)
                     else -> throw IllegalArgumentException("Invalid field: [$fieldName] found in Conditions.")
                 }
             }
 
-            return Conditions(indexAge, docCount, size, cron, rolloverAge, noAlias)
+            return Conditions(indexAge, docCount, size, cron, rolloverAge, noAlias, minStateAge)
         }
     }
 }

src/main/kotlin/org/opensearch/indexmanagement/indexstatemanagement/step/transition/AttemptTransitionStep.kt

@@ -101,9 +101,11 @@ class AttemptTransitionStep(private val action: TransitionsAction) : Step(name)
 
             // Find the first transition that evaluates to true and get the state to transition to, otherwise return null if none are true
             val indexAliasesCount = indexMetadata?.aliases?.size ?: 0
+            val stateStartTime = context.metadata.stateMetaData?.startTime
+            val stateStartInstant = stateStartTime?.let { Instant.ofEpochMilli(it) }
             stateName =
                 transitions.find {
-                    it.evaluateConditions(indexCreationDateInstant, numDocs, indexSize, stepStartTime, rolloverDate, indexAliasesCount)
+                    it.evaluateConditions(indexCreationDateInstant, numDocs, indexSize, stepStartTime, rolloverDate, indexAliasesCount, stateStartInstant)
                 }?.stateName
             val message: String
             val stateName = stateName // shadowed on purpose to prevent var from changing

src/main/kotlin/org/opensearch/indexmanagement/indexstatemanagement/util/ManagedIndexUtils.kt

@@ -190,6 +190,7 @@ fun Transition.evaluateConditions(
     transitionStartTime: Instant,
     rolloverDate: Instant?,
     indexAliasesCount: Int? = null,
+    stateStartTime: Instant? = null,
 ): Boolean {
     // If there are no conditions, treat as always true
     if (this.conditions == null) return true
@@ -225,6 +226,11 @@ fun Transition.evaluateConditions(
             (!this.conditions.noAlias && indexAliasesCount > 0)
     }
 
+    if (this.conditions.minStateAge != null && stateStartTime != null) {
+        val elapsed = System.currentTimeMillis() - stateStartTime.toEpochMilli()
+        return elapsed >= this.conditions.minStateAge.millis
+    }
+
     // We should never reach this
     return false
 }

src/test/kotlin/org/opensearch/indexmanagement/indexstatemanagement/action/TransitionActionIT.kt

@@ -230,4 +230,49 @@ class TransitionActionIT : IndexStateManagementRestTestCase() {
             )
         }
     }
+
+    fun `test minStateAge transition occurs after elapsed time`() {
+        val indexName = "${testIndexName}_min_state_age"
+        val policyID = "${testIndexName}_min_state_age_policy"
+        val secondStateName = "second"
+        val states =
+            listOf(
+                State("first", listOf(), listOf(Transition(secondStateName, Conditions(minStateAge = TimeValue.timeValueSeconds(5))))),
+                State(secondStateName, listOf(), 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)
+        // Initialising policy
+        updateManagedIndexConfigStartTime(managedIndexConfig)
+        waitFor { assertEquals(policyID, getExplainManagedIndexMetaData(indexName).policyID) }
+        // should not transition immediately
+        updateManagedIndexConfigStartTime(managedIndexConfig)
+        waitFor {
+            assertEquals(
+                AttemptTransitionStep.getEvaluatingMessage(indexName),
+                getExplainManagedIndexMetaData(indexName).info?.get("message"),
+            )
+        }
+        // Wait for min_state_age to elapse
+        Thread.sleep(5500)
+        updateManagedIndexConfigStartTime(managedIndexConfig)
+        // Should transition now
+        waitFor {
+            assertEquals(
+                AttemptTransitionStep.getSuccessMessage(indexName, secondStateName),
+                getExplainManagedIndexMetaData(indexName).info?.get("message"),
+            )
+        }
+    }
 }