Backport bug fixes/improvements to opendistro-1.2 (#137)

* Make rollover condition distinct and change ManagedIndexConfig to use PolicySeqNo and PolicyPrimaryTerm (#121) * Make rollover condition distinct and change ManagedIndexConfig to use PolicySeqNo and PolicyPrimaryTerm * add rollover test case for no condition * remove 'which' in the comment * Adds null check in cluster changed event sweep (#125) * Fixes history index iteration breaking instead of continuing (#134) * Fixes issue where action timeout was using start_time from previous action (#133) * Bumps plugin version and notification dependency * Updates usage of LockService to new async API (#107) Co-authored-by: Jinsoo <[email protected]>
opendistro-for-elasticsearch · Jan 17, 2020 · 981a54a · 981a54a
1 parent 7b5f3f2
commit 981a54a
Show file tree

Hide file tree

Showing 11 changed files with 274 additions and 19 deletions.
diff --git a/build.gradle b/build.gradle
@@ -73,12 +73,12 @@ detekt {
 
 dependencies {
     compileOnly "org.elasticsearch:elasticsearch:${es_version}"
-    compileOnly "com.amazon.opendistroforelasticsearch:opendistro-job-scheduler-spi:1.2.1.0"
+    compileOnly "com.amazon.opendistroforelasticsearch:opendistro-job-scheduler-spi:1.2.1.1"
     compile "org.jetbrains.kotlin:kotlin-stdlib:${kotlin_version}"
     compile "org.jetbrains.kotlin:kotlin-stdlib-common:${kotlin_version}"
     compile 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.2.1'
     compile "org.jetbrains:annotations:13.0"
-    compile "com.amazon.opendistroforelasticsearch:notification:1.2.0.0"
+    compile "com.amazon.opendistroforelasticsearch:notification:1.2.1.0"
 
     testCompile "org.elasticsearch.test:framework:${es_version}"
     testCompile "org.jetbrains.kotlin:kotlin-test:${kotlin_version}"
@@ -100,7 +100,7 @@ ext {
 }
 
 group = "com.amazon.opendistroforelasticsearch"
-version = "${opendistroVersion}.0"
+version = "${opendistroVersion}.1"
 
 if (isSnapshot) {
     version += "-SNAPSHOT"

diff --git a/...com/amazon/opendistroforelasticsearch/indexstatemanagement/IndexStateManagementHistory.kt b/...com/amazon/opendistroforelasticsearch/indexstatemanagement/IndexStateManagementHistory.kt
@@ -134,7 +134,7 @@ class IndexStateManagementHistory(
                 val alias = indexMetaData.aliases.firstOrNull { IndexStateManagementIndices.HISTORY_WRITE_INDEX_ALIAS == it.value.alias }
                 if (alias != null && historyEnabled) {
                     // If index has write alias and history is enable, don't delete the index.
-                    break
+                    continue
                 }
 
                 indexToDelete.add(indexMetaData.index.name)

diff --git a/...lin/com/amazon/opendistroforelasticsearch/indexstatemanagement/ManagedIndexCoordinator.kt b/...lin/com/amazon/opendistroforelasticsearch/indexstatemanagement/ManagedIndexCoordinator.kt
@@ -231,7 +231,7 @@ class ManagedIndexCoordinator(
     @OpenForTesting
     suspend fun sweepClusterChangedEvent(event: ClusterChangedEvent) {
         val indicesDeletedRequests = event.indicesDeleted()
-                    .filter { event.previousState().metaData().index(it).getPolicyID() != null }
+                    .filter { event.previousState().metaData().index(it)?.getPolicyID() != null }
                     .map { deleteManagedIndexRequest(it.uuid) }
 
         /*

diff --git a/...n/kotlin/com/amazon/opendistroforelasticsearch/indexstatemanagement/ManagedIndexRunner.kt b/...n/kotlin/com/amazon/opendistroforelasticsearch/indexstatemanagement/ManagedIndexRunner.kt
@@ -45,6 +45,7 @@ import com.amazon.opendistroforelasticsearch.indexstatemanagement.util.getAction
 import com.amazon.opendistroforelasticsearch.indexstatemanagement.util.getStartingManagedIndexMetaData
 import com.amazon.opendistroforelasticsearch.indexstatemanagement.util.getStateToExecute
 import com.amazon.opendistroforelasticsearch.indexstatemanagement.util.getCompletedManagedIndexMetaData
+import com.amazon.opendistroforelasticsearch.indexstatemanagement.util.getUpdatedActionMetaData
 import com.amazon.opendistroforelasticsearch.indexstatemanagement.util.hasDifferentJobInterval
 import com.amazon.opendistroforelasticsearch.indexstatemanagement.util.hasTimedOut
 import com.amazon.opendistroforelasticsearch.indexstatemanagement.util.hasVersionConflict
@@ -174,13 +175,13 @@ object ManagedIndexRunner : ScheduledJobRunner,
 
         launch {
             // Attempt to acquire lock
-            val lock: LockModel? = withContext(Dispatchers.IO) { context.lockService.acquireLock(job, context) }
+            val lock: LockModel? = context.lockService.suspendUntil { acquireLock(job, context, it) }
             if (lock == null) {
                 logger.debug("Could not acquire lock for ${job.index}")
             } else {
                 runManagedIndexConfig(job)
                 // Release lock
-                val released = withContext(Dispatchers.IO) { context.lockService.release(lock) }
+                val released: Boolean = context.lockService.suspendUntil { release(lock, it) }
                 if (!released) {
                     logger.debug("Could not release lock for ${job.index}")
                 }
@@ -233,6 +234,7 @@ object ManagedIndexRunner : ScheduledJobRunner,
         val state = policy.getStateToExecute(managedIndexMetaData)
         val action: Action? = state?.getActionToExecute(clusterService, scriptService, client, managedIndexMetaData)
         val step: Step? = action?.getStepToExecute()
+        val currentActionMetaData = action?.getUpdatedActionMetaData(managedIndexMetaData, state)
 
         // If Index State Management is disabled and the current step is not null and safe to disable on
         // then disable the job and return early
@@ -241,11 +243,11 @@ object ManagedIndexRunner : ScheduledJobRunner,
             return
         }
 
-        if (action?.hasTimedOut(managedIndexMetaData.actionMetaData) == true) {
+        if (action?.hasTimedOut(currentActionMetaData) == true) {
             val info = mapOf("message" to "Action timed out")
             logger.error("Action=${action.type.type} has timed out")
             val updated = updateManagedIndexMetaData(managedIndexMetaData
-                .copy(actionMetaData = managedIndexMetaData.actionMetaData?.copy(failed = true), info = info))
+                .copy(actionMetaData = currentActionMetaData?.copy(failed = true), info = info))
             if (updated) disableManagedIndexConfig(managedIndexConfig)
             return
         }
@@ -255,7 +257,7 @@ object ManagedIndexRunner : ScheduledJobRunner,
             return
         }
 
-        val shouldBackOff = action?.shouldBackoff(managedIndexMetaData.actionMetaData, action.config.configRetry)
+        val shouldBackOff = action?.shouldBackoff(currentActionMetaData, action.config.configRetry)
         if (shouldBackOff?.first == true) {
             // If we should back off then exit early.
             logger.info("Backoff for retrying. Remaining time ${shouldBackOff.second}")
@@ -282,9 +284,7 @@ object ManagedIndexRunner : ScheduledJobRunner,
         val startingManagedIndexMetaData = managedIndexMetaData.getStartingManagedIndexMetaData(state, action, step)
         val updateResult = updateManagedIndexMetaData(startingManagedIndexMetaData)
 
-        val actionMetaData = startingManagedIndexMetaData.actionMetaData
-
-        if (updateResult && state != null && action != null && step != null && actionMetaData != null) {
+        if (updateResult && state != null && action != null && step != null && currentActionMetaData != null) {
             // Step null check is done in getStartingManagedIndexMetaData
             step.execute()
             var executedManagedIndexMetaData = startingManagedIndexMetaData.getCompletedManagedIndexMetaData(action, step)

diff --git a/...om/amazon/opendistroforelasticsearch/indexstatemanagement/elasticapi/ElasticExtensions.kt b/...om/amazon/opendistroforelasticsearch/indexstatemanagement/elasticapi/ElasticExtensions.kt
@@ -20,6 +20,7 @@ package com.amazon.opendistroforelasticsearch.indexstatemanagement.elasticapi
 import com.amazon.opendistroforelasticsearch.indexstatemanagement.model.ManagedIndexMetaData
 import com.amazon.opendistroforelasticsearch.indexstatemanagement.model.coordinator.ClusterStateManagedIndexConfig
 import com.amazon.opendistroforelasticsearch.indexstatemanagement.settings.ManagedIndexSettings
+import com.amazon.opendistroforelasticsearch.jobscheduler.spi.utils.LockService
 import kotlinx.coroutines.delay
 import org.apache.logging.log4j.Logger
 import org.elasticsearch.ElasticsearchException
@@ -124,6 +125,20 @@ suspend fun <C : ElasticsearchClient, T> C.suspendUntil(block: C.(ActionListener
             })
         }
 
+/**
+ * Converts [LockService] methods that take a callback into a kotlin suspending function.
+ *
+ * @param block - a block of code that is passed an [ActionListener] that should be passed to the LockService API.
+ */
+suspend fun <T> LockService.suspendUntil(block: LockService.(ActionListener<T>) -> Unit): T =
+    suspendCoroutine { cont ->
+        block(object : ActionListener<T> {
+            override fun onResponse(response: T) = cont.resume(response)
+
+            override fun onFailure(e: Exception) = cont.resumeWithException(e)
+        })
+    }
+
 /**
  * Compares current and previous IndexMetaData to determine if we should create [ManagedIndexConfig].
  *

diff --git a/...in/com/amazon/opendistroforelasticsearch/indexstatemanagement/model/ManagedIndexConfig.kt b/...in/com/amazon/opendistroforelasticsearch/indexstatemanagement/model/ManagedIndexConfig.kt
@@ -177,7 +177,8 @@ data class ManagedIndexConfig(
                 policyID = requireNotNull(policyID) { "ManagedIndexConfig policy id is null" },
                 policySeqNo = policySeqNo,
                 policyPrimaryTerm = policyPrimaryTerm,
-                policy = policy,
+                policy = policy?.copy(seqNo = policySeqNo ?: SequenceNumbers.UNASSIGNED_SEQ_NO,
+                    primaryTerm = policyPrimaryTerm ?: SequenceNumbers.UNASSIGNED_PRIMARY_TERM),
                 changePolicy = changePolicy
             )
         }

diff --git a/...tlin/com/amazon/opendistroforelasticsearch/indexstatemanagement/util/ManagedIndexUtils.kt b/...tlin/com/amazon/opendistroforelasticsearch/indexstatemanagement/util/ManagedIndexUtils.kt
@@ -218,22 +218,28 @@ fun RolloverActionConfig.evaluateConditions(
     numDocs: Long,
     indexSize: ByteSizeValue
 ): Boolean {
+    if (this.minDocs == null &&
+        this.minAge == null &&
+        this.minSize == null) {
+        // If no conditions specified we default to true
+        return true
+    }
 
     if (this.minDocs != null) {
-        return this.minDocs <= numDocs
+        if (this.minDocs <= numDocs) return true
     }
 
     if (this.minAge != null) {
         val elapsedTime = Instant.now().toEpochMilli() - indexCreationDate.toEpochMilli()
-        return this.minAge.millis <= elapsedTime
+        if (this.minAge.millis <= elapsedTime) return true
     }
 
     if (this.minSize != null) {
-        return this.minSize <= indexSize
+        if (this.minSize <= indexSize) return true
     }
 
-    // If no conditions specified we default to true
-    return true
+    // return false if non of the conditions were true.
+    return false
 }
 
 fun Policy.getStateToExecute(managedIndexMetaData: ManagedIndexMetaData): State? {

diff --git a/...tlin/com/amazon/opendistroforelasticsearch/indexstatemanagement/action/ActionTimeoutIT.kt b/...tlin/com/amazon/opendistroforelasticsearch/indexstatemanagement/action/ActionTimeoutIT.kt
@@ -76,4 +76,60 @@ class ActionTimeoutIT : IndexStateManagementRestTestCase() {
             )
         }
     }
+
+    // https://github.com/opendistro-for-elasticsearch/index-management/issues/130
+    fun `test action timeout doesn't bleed over into next action`() {
+        val indexName = "${testIndexName}_index_1"
+        val policyID = "${testIndexName}_testPolicyName_1"
+        val testPolicy = """
+        {"policy":{"description":"Default policy","default_state":"rolloverstate","states":[
+        {"name":"rolloverstate","actions":[{"timeout": "5s","open":{}},{"timeout":"1s","rollover":{"min_doc_count":100}}],
+        "transitions":[]}]}}
+        """.trimIndent()
+
+        createPolicyJson(testPolicy, policyID)
+
+        createIndex(indexName, policyID, "some_alias")
+
+        val managedIndexConfig = getExistingManagedIndexConfig(indexName)
+        // Change the start time so the job will trigger in 2 seconds.
+        // First execution. We need to initialize the policy.
+        updateManagedIndexConfigStartTime(managedIndexConfig)
+
+        waitFor {
+            assertPredicatesOnMetaData(
+                listOf(indexName to listOf(ManagedIndexMetaData.POLICY_ID to policyID::equals)),
+                getExplainMap(indexName),
+                strict = false
+            )
+        }
+
+        // the second execution we move into open action, we won't hit the timeout as this is the execution that sets the startTime
+        updateManagedIndexConfigStartTime(managedIndexConfig)
+
+        val expectedOpenInfoString = mapOf("message" to "Successfully opened index").toString()
+        waitFor {
+            assertPredicatesOnMetaData(
+                listOf(indexName to listOf(ManagedIndexMetaData.INFO to fun(info: Any?): Boolean = expectedOpenInfoString == info.toString())),
+                getExplainMap(indexName),
+                strict = false
+            )
+        }
+
+        // wait 5 seconds for the timeout from the first action to have passed
+        Thread.sleep(5000L)
+
+        // the third execution we move into rollover action, we should not hit the timeout yet because its the first execution of rollover
+        // but there was a bug before where it would use the startTime from the previous actions metadata and immediately fail
+        updateManagedIndexConfigStartTime(managedIndexConfig)
+
+        val expectedRolloverInfoString = mapOf("message" to "Attempting to rollover").toString()
+        waitFor {
+            assertPredicatesOnMetaData(
+                listOf(indexName to listOf(ManagedIndexMetaData.INFO to fun(info: Any?): Boolean = expectedRolloverInfoString == info.toString())),
+                getExplainMap(indexName),
+                strict = false
+            )
+        }
+    }
 }