Add boolean parameter to the ISM rollover action that prevents rollover on empty indices (#1545) (#1547)

* Add boolean parameter to the ISM rollover action that prevents rollover of indices with zero documents



* Add version checks



---------


(cherry picked from commit fe5da1d)

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

Author: tandonks

src/main/kotlin/org/opensearch/indexmanagement/indexstatemanagement/action/RolloverAction.kt

@@ -5,6 +5,7 @@
 
 package org.opensearch.indexmanagement.indexstatemanagement.action
 
+import org.opensearch.Version
 import org.opensearch.common.unit.TimeValue
 import org.opensearch.core.common.io.stream.StreamOutput
 import org.opensearch.core.common.unit.ByteSizeValue
@@ -21,6 +22,7 @@ class RolloverAction(
     val minAge: TimeValue?,
     val minPrimaryShardSize: ByteSizeValue?,
     val copyAlias: Boolean = false,
+    val preventEmptyRollover: Boolean = false,
     index: Int,
 ) : Action(name, index) {
     init {
@@ -48,6 +50,7 @@ class RolloverAction(
         if (minAge != null) builder.field(MIN_INDEX_AGE_FIELD, minAge.stringRep)
         if (minPrimaryShardSize != null) builder.field(MIN_PRIMARY_SHARD_SIZE_FIELD, minPrimaryShardSize.stringRep)
         builder.field(COPY_ALIAS_FIELD, copyAlias)
+        if (preventEmptyRollover) builder.field(PREVENT_EMPTY_ROLLOVER_FIELD, preventEmptyRollover)
         builder.endObject()
     }
 
@@ -57,6 +60,9 @@ class RolloverAction(
         out.writeOptionalTimeValue(minAge)
         out.writeOptionalWriteable(minPrimaryShardSize)
         out.writeBoolean(copyAlias)
+        if (out.version.onOrAfter(Version.V_3_4_0)) {
+            out.writeBoolean(preventEmptyRollover)
+        }
         out.writeInt(actionIndex)
     }
 
@@ -67,5 +73,6 @@ class RolloverAction(
         const val MIN_INDEX_AGE_FIELD = "min_index_age"
         const val MIN_PRIMARY_SHARD_SIZE_FIELD = "min_primary_shard_size"
         const val COPY_ALIAS_FIELD = "copy_alias"
+        const val PREVENT_EMPTY_ROLLOVER_FIELD = "prevent_empty_rollover"
     }
 }

src/main/kotlin/org/opensearch/indexmanagement/indexstatemanagement/action/RolloverActionParser.kt

@@ -5,6 +5,7 @@
 
 package org.opensearch.indexmanagement.indexstatemanagement.action
 
+import org.opensearch.Version
 import org.opensearch.common.unit.TimeValue
 import org.opensearch.core.common.io.stream.StreamInput
 import org.opensearch.core.common.unit.ByteSizeValue
@@ -20,9 +21,14 @@ class RolloverActionParser : ActionParser() {
         val minAge = sin.readOptionalTimeValue()
         val minPrimaryShardSize = sin.readOptionalWriteable(::ByteSizeValue)
         val copyAlias = sin.readBoolean()
+        val preventEmptyRollover = if (sin.version.onOrAfter(Version.V_3_4_0)) {
+            sin.readBoolean()
+        } else {
+            false
+        }
         val index = sin.readInt()
 
-        return RolloverAction(minSize, minDocs, minAge, minPrimaryShardSize, copyAlias, index)
+        return RolloverAction(minSize, minDocs, minAge, minPrimaryShardSize, copyAlias, preventEmptyRollover, index)
     }
 
     override fun fromXContent(xcp: XContentParser, index: Int): Action {
@@ -31,6 +37,7 @@ class RolloverActionParser : ActionParser() {
         var minAge: TimeValue? = null
         var minPrimaryShardSize: ByteSizeValue? = null
         var copyAlias = false
+        var preventEmptyRollover = false
 
         ensureExpectedToken(XContentParser.Token.START_OBJECT, xcp.currentToken(), xcp)
         while (xcp.nextToken() != XContentParser.Token.END_OBJECT) {
@@ -49,11 +56,14 @@ class RolloverActionParser : ActionParser() {
                                 .MIN_PRIMARY_SHARD_SIZE_FIELD,
                         )
                 RolloverAction.COPY_ALIAS_FIELD -> copyAlias = xcp.booleanValue()
+
+                RolloverAction.PREVENT_EMPTY_ROLLOVER_FIELD -> preventEmptyRollover = xcp.booleanValue()
+
                 else -> throw IllegalArgumentException("Invalid field: [$fieldName] found in RolloverAction.")
             }
         }
 
-        return RolloverAction(minSize, minDocs, minAge, minPrimaryShardSize, copyAlias, index)
+        return RolloverAction(minSize, minDocs, minAge, minPrimaryShardSize, copyAlias, preventEmptyRollover, index)
     }
 
     override fun getActionType(): String = RolloverAction.name

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

@@ -75,6 +75,25 @@ class AttemptRolloverStep(private val action: RolloverAction) : Step(name) {
         // If statsResponse is null we already updated failed info from getIndexStatsOrUpdateInfo and can return early
         statsResponse ?: return this
 
+        // Check for empty index if prevent_empty_rollover is enabled
+        if (action.preventEmptyRollover) {
+            val numDocs = statsResponse.primaries.docs?.count ?: 0
+            if (numDocs == 0L) {
+                stepStatus = StepStatus.CONDITION_NOT_MET
+                info = mapOf(
+                    "message" to getPreventedEmptyRolloverMessage(indexName),
+                    "conditions" to mapOf(
+                        "prevent_empty_rollover" to mapOf(
+                            "condition" to "index must have at least 1 document",
+                            "current" to 0,
+                        ),
+                    ),
+                )
+                logger.info("$indexName rollover prevented: index is empty (0 documents)")
+                return this
+            }
+        }
+
         val indexCreationDate = clusterService.state().metadata().index(indexName).creationDate
         val indexAgeTimeValue =
             if (indexCreationDate == -1L) {
@@ -426,5 +445,8 @@ class AttemptRolloverStep(private val action: RolloverAction) : Step(name) {
 
         fun getCopyAliasRolledOverIndexNotFoundMessage(index: String?) =
             "Successfully rolled over [index=$index] but ISM cannot find rolled over index from metadata to copy aliases to, please manually copy"
+
+        fun getPreventedEmptyRolloverMessage(index: String) =
+            "Rollover prevented: index is empty (0 documents) [index=$index]"
     }
 }

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

@@ -1,6 +1,6 @@
 {
   "_meta" : {
-    "schema_version": 24
+    "schema_version": 25
   },
   "dynamic": "strict",
   "properties": {
@@ -245,6 +245,9 @@
                     },
                     "copy_alias": {
                       "type": "boolean"
+                    },
+                    "prevent_empty_rollover": {
+                      "type": "boolean"
                     }
                   }
                 },

src/test/kotlin/org/opensearch/indexmanagement/IndexManagementRestTestCase.kt

@@ -41,7 +41,7 @@ import javax.management.remote.JMXConnectorFactory
 import javax.management.remote.JMXServiceURL
 
 abstract class IndexManagementRestTestCase : ODFERestTestCase() {
-    val configSchemaVersion = 24
+    val configSchemaVersion = 25
     val historySchemaVersion = 7
 
     // Having issues with tests leaking into other tests and mappings being incorrect and they are not caught by any pending task wait check as

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

@@ -133,6 +133,7 @@ fun randomRolloverActionConfig(
     minDocs = minDocs,
     minAge = minAge,
     minPrimaryShardSize = minPrimaryShardSize,
+    preventEmptyRollover = false,
     index = 0,
 )