Merge branch 'main' into patch-1

Author: Hailong-am

build.gradle

@@ -239,7 +239,7 @@ dependencies {
     testImplementation "com.nhaarman.mockitokotlin2:mockito-kotlin:2.2.0"
     testImplementation "org.mockito:mockito-core:${versions.mockito}"
 
-    add("ktlint", "com.pinterest.ktlint:ktlint-cli:1.5.0") {
+    add("ktlint", "com.pinterest.ktlint:ktlint-cli:1.8.0") {
         attributes {
             attribute(Bundling.BUNDLING_ATTRIBUTE, objects.named(Bundling, Bundling.EXTERNAL))
         }

spi/src/main/kotlin/org.opensearch.indexmanagement.spi/indexstatemanagement/Action.kt

@@ -61,8 +61,10 @@ abstract class Action(
             // start a new action
             stateMetaData?.name != stateName ->
                 ActionMetaData(this.type, Instant.now().toEpochMilli(), this.actionIndex, false, 0, 0, null)
+
             actionMetaData?.index != this.actionIndex ->
                 ActionMetaData(this.type, Instant.now().toEpochMilli(), this.actionIndex, false, 0, 0, null)
+
             // RetryAPI will reset startTime to null for actionMetaData and we'll reset it to "now" here
             else -> actionMetaData.copy(startTime = actionMetaData.startTime ?: Instant.now().toEpochMilli())
         }

spi/src/main/kotlin/org.opensearch.indexmanagement.spi/indexstatemanagement/Step.kt

@@ -137,11 +137,14 @@ abstract class Step(val name: String, val isSafeToDisableOn: Boolean = true) {
 
     final fun getStepStartTime(metadata: ManagedIndexMetaData): Instant = when {
         metadata.stepMetaData == null -> Instant.now()
+
         metadata.stepMetaData.name != this.name -> Instant.now()
+
         // The managed index metadata is a historical snapshot of the metadata and refers to what has happened from the previous
         // execution, so if we ever see it as COMPLETED it means we are always going to be in a new step, this specifically
         // helps with the Transition -> Transition (empty state) sequence which the above do not capture
         metadata.stepMetaData.stepStatus == StepStatus.COMPLETED -> Instant.now()
+
         else -> Instant.ofEpochMilli(metadata.stepMetaData.startTime)
     }
 

spi/src/main/kotlin/org.opensearch.indexmanagement.spi/indexstatemanagement/model/ActionMetaData.kt

@@ -120,11 +120,17 @@ data class ActionMetaData(
 
                 when (fieldName) {
                     NAME -> name = xcp.text()
+
                     START_TIME -> startTime = if (xcp.currentToken() == XContentParser.Token.VALUE_NULL) null else xcp.longValue()
+
                     INDEX -> index = xcp.intValue()
+
                     FAILED -> failed = xcp.booleanValue()
+
                     CONSUMED_RETRIES -> consumedRetries = xcp.intValue()
+
                     LAST_RETRY_TIME -> lastRetryTime = if (xcp.currentToken() == XContentParser.Token.VALUE_NULL) null else xcp.longValue()
+
                     ActionProperties.ACTION_PROPERTIES ->
                         actionProperties = if (xcp.currentToken() == XContentParser.Token.VALUE_NULL) null else ActionProperties.parse(xcp)
                 }

spi/src/main/kotlin/org.opensearch.indexmanagement.spi/indexstatemanagement/model/ActionProperties.kt

@@ -75,12 +75,17 @@ data class ActionProperties(
 
                 when (fieldName) {
                     Properties.MAX_NUM_SEGMENTS.key -> maxNumSegments = xcp.intValue()
+
                     Properties.SNAPSHOT_NAME.key -> snapshotName = xcp.text()
+
                     Properties.ROLLUP_ID.key -> rollupId = xcp.text()
+
                     Properties.HAS_ROLLUP_FAILED.key -> hasRollupFailed = xcp.booleanValue()
+
                     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)
                     }

spi/src/main/kotlin/org.opensearch.indexmanagement.spi/indexstatemanagement/model/ManagedIndexMetaData.kt

@@ -259,27 +259,41 @@ data class ManagedIndexMetaData(
 
                 when (fieldName) {
                     INDEX -> index = xcp.text()
+
                     INDEX_UUID -> indexUuid = xcp.text()
+
                     POLICY_ID -> policyID = xcp.text()
+
                     POLICY_SEQ_NO -> policySeqNo = if (xcp.currentToken() == XContentParser.Token.VALUE_NULL) null else xcp.longValue()
+
                     POLICY_PRIMARY_TERM -> policyPrimaryTerm = if (xcp.currentToken() == XContentParser.Token.VALUE_NULL) null else xcp.longValue()
+
                     POLICY_COMPLETED -> policyCompleted = if (xcp.currentToken() == XContentParser.Token.VALUE_NULL) null else xcp.booleanValue()
+
                     ROLLED_OVER -> rolledOver = if (xcp.currentToken() == XContentParser.Token.VALUE_NULL) null else xcp.booleanValue()
+
                     ROLLED_OVER_INDEX_NAME -> rolledOverIndexName = if (xcp.currentToken() == XContentParser.Token.VALUE_NULL) null else xcp.text()
+
                     INDEX_CREATION_DATE -> indexCreationDate = if (xcp.currentToken() == XContentParser.Token.VALUE_NULL) null else xcp.longValue()
+
                     TRANSITION_TO -> transitionTo = if (xcp.currentToken() == XContentParser.Token.VALUE_NULL) null else xcp.text()
+
                     StateMetaData.STATE -> {
                         state = if (xcp.currentToken() == XContentParser.Token.VALUE_NULL) null else StateMetaData.parse(xcp)
                     }
+
                     ActionMetaData.ACTION -> {
                         action = if (xcp.currentToken() == XContentParser.Token.VALUE_NULL) null else ActionMetaData.parse(xcp)
                     }
+
                     StepMetaData.STEP -> {
                         step = if (xcp.currentToken() == XContentParser.Token.VALUE_NULL) null else StepMetaData.parse(xcp)
                     }
+
                     PolicyRetryInfoMetaData.RETRY_INFO -> {
                         retryInfo = PolicyRetryInfoMetaData.parse(xcp)
                     }
+
                     INFO -> info = xcp.map()
                     // below line will break when getting metadata for explain or history
                     // else -> throw IllegalArgumentException("Invalid field: [$fieldName] found in ManagedIndexMetaData.")

src/main/kotlin/org/opensearch/indexmanagement/IndexManagementPlugin.kt

@@ -269,30 +269,39 @@ class IndexManagementPlugin :
                     ManagedIndexConfig.MANAGED_INDEX_TYPE -> {
                         return@ScheduledJobParser ManagedIndexConfig.parse(xcp, id, jobDocVersion.seqNo, jobDocVersion.primaryTerm)
                     }
+
                     Policy.POLICY_TYPE -> {
                         return@ScheduledJobParser null
                     }
+
                     Rollup.ROLLUP_TYPE -> {
                         return@ScheduledJobParser Rollup.parse(xcp, id, jobDocVersion.seqNo, jobDocVersion.primaryTerm)
                     }
+
                     RollupMetadata.ROLLUP_METADATA_TYPE -> {
                         return@ScheduledJobParser null
                     }
+
                     Transform.TRANSFORM_TYPE -> {
                         return@ScheduledJobParser Transform.parse(xcp, id, jobDocVersion.seqNo, jobDocVersion.primaryTerm)
                     }
+
                     TransformMetadata.TRANSFORM_METADATA_TYPE -> {
                         return@ScheduledJobParser null
                     }
+
                     ManagedIndexMetaData.MANAGED_INDEX_METADATA_TYPE -> {
                         return@ScheduledJobParser null
                     }
+
                     SMPolicy.SM_TYPE -> {
                         return@ScheduledJobParser SMPolicy.parse(xcp, id, jobDocVersion.seqNo, jobDocVersion.primaryTerm)
                     }
+
                     SMMetadata.SM_METADATA_TYPE -> {
                         return@ScheduledJobParser null
                     }
+
                     else -> {
                         logger.warn("Unsupported document was indexed in $INDEX_MANAGEMENT_INDEX with type: $fieldName")
                         xcp.skipChildren()

src/main/kotlin/org/opensearch/indexmanagement/IndexManagementRunner.kt

@@ -24,9 +24,13 @@ object IndexManagementRunner : ScheduledJobRunner {
     override fun runJob(job: ScheduledJobParameter, context: JobExecutionContext) {
         when (job) {
             is ManagedIndexConfig -> ManagedIndexRunner.runJob(job, context)
+
             is Rollup -> RollupRunner.runJob(job, context)
+
             is Transform -> TransformRunner.runJob(job, context)
+
             is SMPolicy -> SMRunner.runJob(job, context)
+
             else -> {
                 val errorMessage = "Invalid job type, found ${job.javaClass.simpleName} with id: ${context.jobId}"
                 logger.error(errorMessage)

src/main/kotlin/org/opensearch/indexmanagement/common/model/notification/Channel.kt

@@ -59,6 +59,7 @@ data class Channel(val id: String) :
                 xcp.nextToken()
                 when (fieldName) {
                     ID -> id = xcp.text()
+
                     else -> {
                         error("Unexpected field: $fieldName, while parsing Channel destination")
                     }

src/main/kotlin/org/opensearch/indexmanagement/controlcenter/notification/filter/NotificationActionListener.kt

@@ -288,10 +288,13 @@ class NotificationActionListener<Request : ActionRequest, Response : ActionRespo
         if (runtimeConfig != null) channels.add(runtimeConfig)
         if (defaultConfig != null) channels.add(defaultConfig)
 
-        return channels.filter { ch ->
-            val condition = ch.lronConfig.lronCondition
-            condition.success && result == OperationResult.COMPLETE || condition.failure && result != OperationResult.COMPLETE
-        }.toSet()
+        return channels
+            .filter { ch ->
+                val condition = ch.lronConfig.lronCondition
+                (condition.success && result == OperationResult.COMPLETE) ||
+                    (condition.failure && result != OperationResult.COMPLETE)
+            }
+            .toSet()
     }
 
     fun escapeQueryString(query: String): String = query.replace("/", "\\/").replace(":", "\\:")