Allows out of band rollovers on an index without causing ISM to fail (#180)

* Allows out of band rollovers on an index without causing ISM to fail

Signed-off-by: Drew Baugher <46505179+dbbaughe@users.noreply.github.com>

* Fixes detekt issue

Signed-off-by: Drew Baugher <46505179+dbbaughe@users.noreply.github.com>

Author: dbbaughe

src/main/kotlin/org/opensearch/indexmanagement/indexstatemanagement/step/rollover/AttemptRolloverStep.kt

@@ -72,18 +72,16 @@ class AttemptRolloverStep(
             return this
         }
 
-        // If we have already rolled over this index then fail as we only allow an index to be rolled over once
-        if (managedIndexMetaData.rolledOver == true) {
-            logger.warn("$indexName was already rolled over, cannot execute rollover step")
-            stepStatus = StepStatus.FAILED
-            info = mapOf("message" to getFailedDuplicateRolloverMessage(indexName))
-            return this
-        }
-
         val (rolloverTarget, isDataStream) = getRolloverTargetOrUpdateInfo()
         // If the rolloverTarget is null, we would've already updated the failed info from getRolloverTargetOrUpdateInfo and can return early
         rolloverTarget ?: return this
 
+        if (clusterService.state().metadata.index(indexName).rolloverInfos.containsKey(rolloverTarget)) {
+            stepStatus = StepStatus.COMPLETED
+            info = mapOf("message" to getAlreadyRolledOverMessage(indexName, rolloverTarget))
+            return this
+        }
+
         if (!isDataStream && !preCheckIndexAlias(rolloverTarget)) {
             stepStatus = StepStatus.FAILED
             info = mapOf("message" to getFailedPreCheckMessage(indexName))
@@ -285,13 +283,14 @@ class AttemptRolloverStep(
             "New index created, but failed to update alias [index=$index, newIndex=$newIndex]"
         fun getFailedDataStreamRolloverMessage(dataStream: String) = "Failed to rollover data stream [data_stream=$dataStream]"
         fun getFailedNoValidAliasMessage(index: String) = "Missing rollover_alias index setting [index=$index]"
-        fun getFailedDuplicateRolloverMessage(index: String) = "Index has already been rolled over [index=$index]"
         fun getFailedEvaluateMessage(index: String) = "Failed to evaluate conditions for rollover [index=$index]"
         fun getPendingMessage(index: String) = "Pending rollover of index [index=$index]"
         fun getSuccessMessage(index: String) = "Successfully rolled over index [index=$index]"
         fun getSuccessDataStreamRolloverMessage(dataStream: String, index: String) =
             "Successfully rolled over data stream [data_stream=$dataStream index=$index]"
         fun getFailedPreCheckMessage(index: String) = "Missing alias or not the write index when rollover [index=$index]"
         fun getSkipRolloverMessage(index: String) = "Skipped rollover action for [index=$index]"
+        fun getAlreadyRolledOverMessage(index: String, alias: String) =
+            "This index has already been rolled over using this alias, treating as a success [index=$index, alias=$alias]"
     }
 }

src/test/kotlin/org/opensearch/indexmanagement/indexstatemanagement/IndexStateManagementRestTestCase.kt

@@ -595,6 +595,16 @@ abstract class IndexStateManagementRestTestCase : IndexManagementRestTestCase()
         return metadata
     }
 
+    protected fun rolloverIndex(index: String) {
+        val response = client().performRequest(
+            Request(
+                "POST",
+                "/$index/_rollover"
+            )
+        )
+        assertEquals(response.statusLine.statusCode, RestStatus.OK.status)
+    }
+
     protected fun createRepository(
         repository: String
     ) {

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

@@ -42,6 +42,7 @@ import org.opensearch.indexmanagement.indexstatemanagement.model.action.Rollover
 import org.opensearch.indexmanagement.indexstatemanagement.randomErrorNotification
 import org.opensearch.indexmanagement.indexstatemanagement.resthandler.RestRetryFailedManagedIndexAction
 import org.opensearch.indexmanagement.indexstatemanagement.settings.ManagedIndexSettings
+import org.opensearch.indexmanagement.indexstatemanagement.step.Step
 import org.opensearch.indexmanagement.indexstatemanagement.step.rollover.AttemptRolloverStep
 import org.opensearch.indexmanagement.makeRequest
 import org.opensearch.indexmanagement.waitFor
@@ -496,4 +497,46 @@ class RolloverActionIT : IndexStateManagementRestTestCase() {
         val secondIndexName = DataStream.getDefaultBackingIndexName(dataStreamName, 2L)
         Assert.assertTrue("New rollover index does not exist.", indexExists(secondIndexName))
     }
+
+    @Suppress("UNCHECKED_CAST")
+    fun `test rollover from outside ISM doesn't fail ISM job`() {
+        val aliasName = "${testIndexName}_alias"
+        val indexNameBase = "${testIndexName}_index"
+        val firstIndex = "$indexNameBase-1"
+        val policyID = "${testIndexName}_testPolicyName_1"
+        val actionConfig = RolloverActionConfig(null, null, null, 0)
+        val states = listOf(State(name = "RolloverAction", actions = listOf(actionConfig), transitions = 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)
+        // create index defaults
+        createIndex(firstIndex, policyID, aliasName)
+
+        val managedIndexConfig = getExistingManagedIndexConfig(firstIndex)
+
+        // Change the start time so the job will trigger in 2 seconds, this will trigger the first initialization of the policy
+        updateManagedIndexConfigStartTime(managedIndexConfig)
+        waitFor { assertEquals(policyID, getExplainManagedIndexMetaData(firstIndex).policyID) }
+
+        // Rollover the alias manually before ISM tries to roll it over
+        rolloverIndex(aliasName)
+
+        // Need to speed up to second execution where it will trigger the first execution of the action
+        updateManagedIndexConfigStartTime(managedIndexConfig)
+        waitFor {
+            val info = getExplainManagedIndexMetaData(firstIndex).info as Map<String, Any?>
+            val stepMetadata = getExplainManagedIndexMetaData(firstIndex).stepMetaData
+            assertEquals("Index should succeed if already rolled over.", AttemptRolloverStep.getAlreadyRolledOverMessage(firstIndex, aliasName), info["message"])
+            assertEquals("Index should succeed if already rolled over.", Step.StepStatus.COMPLETED, stepMetadata?.stepStatus)
+        }
+        assertTrue("New rollover index does not exist.", indexExists("$indexNameBase-000002"))
+    }
 }