Fixes force merge failing on long executions, changes some action mes… (#267)

* Fixes force merge failing on long executions, changes some action messaging, adds better try/catch on actions to deal with remote transport exceptions

* Adds unit tests for failures in steps

* Adds more tests

* Updates Allocation step with message and execute changes

* Updates Allocation test message

* Updates jacoco source so the source files show up in the report

* Adds tests for wait for snapshot step

* Adds tests for attempt snapshot step and some missing license headers

* Addressing comments on naming and spacing

Author: dbbaughe

README.md

@@ -64,15 +64,6 @@ The project in this package uses the [Gradle](https://docs.gradle.org/current/us
 
 However, to build the `index management` plugin project, we also use the Elastic build tools for Gradle.  These tools are idiosyncratic and don't always follow the conventions and instructions for building regular Java code using Gradle. Not everything in `index management` will work the way it's described in the Gradle documentation. If you encounter such a situation, the Elastic build tools [source code](https://github.com/elastic/elasticsearch/tree/master/buildSrc/src/main/groovy/org/elasticsearch/gradle) is your best bet for figuring out what's going on.
 
-This project currently uses the Notification subproject from the [Alerting plugin](https://github.com/opendistro-for-elasticsearch/alerting). There is an [open PR](https://github.com/opendistro-for-elasticsearch/alerting/pull/97) that introduces the maven publish task in Alerting for publishing the Notification jars. Until this PR is fully merged and jars published you will need to pull down the PR yourself and publish the jars to your local maven repository in order to build Index Management.
-
-1. Visit the PR [here](https://github.com/opendistro-for-elasticsearch/alerting/pull/97) and pull down the Alerting plugin along with the PR changes
-2. You may need to cherry-pick the changes into a separate branch if you require a specific version to be published
-3. Build the Alerting plugin (w/ the changes in PR) and publish the artifacts to your local maven repository
-     1. `./gradlew clean`
-     2. `./gradlew build` or `./gradlew assemble` build will run the tests and build artifacts, assemble will only build the artifacts
-     3. `./gradlew publishToMavenLocal` publishes artifacts to your local maven repository
-
 ### Building from the command line
 
 1. `./gradlew build` builds and tests project.

build-tools/esplugin-coverage.gradle

@@ -58,7 +58,7 @@ integTest.runner {
 jacocoTestReport {
     dependsOn integTest, test
     executionData dummyTest.jacoco.destinationFile, dummyIntegTest.jacoco.destinationFile
-    sourceDirectories.from = sourceSets.main.allSource
+    sourceDirectories.from = "src/main/kotlin"
     classDirectories.from = sourceSets.main.output
     reports {
         html.enabled = true // human readable

src/main/kotlin/com/amazon/opendistroforelasticsearch/indexstatemanagement/ManagedIndexRunner.kt

@@ -290,7 +290,7 @@ object ManagedIndexRunner : ScheduledJobRunner,
 
         if (updateResult && state != null && action != null && step != null && currentActionMetaData != null) {
             // Step null check is done in getStartingManagedIndexMetaData
-            step.execute()
+            step.preExecute(logger).execute().postExecute(logger)
             var executedManagedIndexMetaData = startingManagedIndexMetaData.getCompletedManagedIndexMetaData(action, step)
 
             if (executedManagedIndexMetaData.isFailed) {

src/main/kotlin/com/amazon/opendistroforelasticsearch/indexstatemanagement/elasticapi/ElasticExtensions.kt

@@ -24,8 +24,10 @@ import com.amazon.opendistroforelasticsearch.jobscheduler.spi.utils.LockService
 import kotlinx.coroutines.delay
 import org.apache.logging.log4j.Logger
 import org.elasticsearch.ElasticsearchException
+import org.elasticsearch.ExceptionsHelper
 import org.elasticsearch.action.ActionListener
 import org.elasticsearch.action.bulk.BackoffPolicy
+import org.elasticsearch.action.support.DefaultShardOperationFailedException
 import org.elasticsearch.client.ElasticsearchClient
 import org.elasticsearch.cluster.metadata.IndexMetadata
 import org.elasticsearch.common.bytes.BytesReference
@@ -36,6 +38,7 @@ import org.elasticsearch.common.xcontent.XContentParser
 import org.elasticsearch.common.xcontent.XContentParserUtils
 import org.elasticsearch.common.xcontent.XContentType
 import org.elasticsearch.rest.RestStatus
+import org.elasticsearch.transport.RemoteTransportException
 import java.time.Instant
 import kotlin.coroutines.resume
 import kotlin.coroutines.resumeWithException
@@ -202,9 +205,7 @@ fun IndexMetadata.getRolloverAlias(): String? {
 fun IndexMetadata.getClusterStateManagedIndexConfig(): ClusterStateManagedIndexConfig? {
     val index = this.index.name
     val uuid = this.index.uuid
-    val policyID = this.getPolicyID()
-
-    if (policyID == null) return null
+    val policyID = this.getPolicyID() ?: return null
 
     return ClusterStateManagedIndexConfig(index = index, uuid = uuid, policyID = policyID)
 }
@@ -217,3 +218,13 @@ fun IndexMetadata.getManagedIndexMetaData(): ManagedIndexMetaData? {
     }
     return null
 }
+
+fun Throwable.findRemoteTransportException(): RemoteTransportException? {
+    if (this is RemoteTransportException) return this
+    return this.cause?.findRemoteTransportException()
+}
+
+fun DefaultShardOperationFailedException.getUsefulCauseString(): String {
+    val rte = this.cause?.findRemoteTransportException()
+    return if (rte == null) this.toString() else ExceptionsHelper.unwrapCause(rte).toString()
+}

src/main/kotlin/com/amazon/opendistroforelasticsearch/indexstatemanagement/model/coordinator/SweptManagedIndexConfig.kt

@@ -40,6 +40,7 @@ data class SweptManagedIndexConfig(
 ) {
 
     companion object {
+        @Suppress("ComplexMethod")
         @JvmStatic
         @Throws(IOException::class)
         fun parse(xcp: XContentParser, seqNo: Long, primaryTerm: Long): SweptManagedIndexConfig {

src/main/kotlin/com/amazon/opendistroforelasticsearch/indexstatemanagement/step/Step.kt

@@ -17,6 +17,7 @@ package com.amazon.opendistroforelasticsearch.indexstatemanagement.step
 
 import com.amazon.opendistroforelasticsearch.indexstatemanagement.model.ManagedIndexMetaData
 import com.amazon.opendistroforelasticsearch.indexstatemanagement.model.managedindexmetadata.StepMetaData
+import org.apache.logging.log4j.Logger
 import org.elasticsearch.common.io.stream.StreamInput
 import org.elasticsearch.common.io.stream.StreamOutput
 import org.elasticsearch.common.io.stream.Writeable
@@ -25,7 +26,17 @@ import java.util.Locale
 
 abstract class Step(val name: String, val managedIndexMetaData: ManagedIndexMetaData, val isSafeToDisableOn: Boolean = true) {
 
-    abstract suspend fun execute()
+    fun preExecute(logger: Logger): Step {
+        logger.info("Executing $name for ${managedIndexMetaData.index}")
+        return this
+    }
+
+    abstract suspend fun execute(): Step
+
+    fun postExecute(logger: Logger): Step {
+        logger.info("Finished executing $name for ${managedIndexMetaData.index}")
+        return this
+    }
 
     abstract fun getUpdatedManagedIndexMetaData(currentMetaData: ManagedIndexMetaData): ManagedIndexMetaData
 
@@ -44,9 +55,7 @@ abstract class Step(val name: String, val managedIndexMetaData: ManagedIndexMeta
      */
     abstract fun isIdempotent(): Boolean
 
-    fun getStartingStepMetaData(): StepMetaData {
-        return StepMetaData(name, getStepStartTime().toEpochMilli(), StepStatus.STARTING)
-    }
+    fun getStartingStepMetaData(): StepMetaData = StepMetaData(name, getStepStartTime().toEpochMilli(), StepStatus.STARTING)
 
     fun getStepStartTime(): Instant {
         if (managedIndexMetaData.stepMetaData == null || managedIndexMetaData.stepMetaData.name != this.name) {
@@ -55,6 +64,8 @@ abstract class Step(val name: String, val managedIndexMetaData: ManagedIndexMeta
         return Instant.ofEpochMilli(managedIndexMetaData.stepMetaData.startTime)
     }
 
+    protected val indexName: String = managedIndexMetaData.index
+
     enum class StepStatus(val status: String) : Writeable {
         STARTING("starting"),
         CONDITION_NOT_MET("condition_not_met"),

src/main/kotlin/com/amazon/opendistroforelasticsearch/indexstatemanagement/step/allocation/AttemptAllocationStep.kt

@@ -39,20 +39,17 @@ class AttemptAllocationStep(
 
     override fun isIdempotent() = true
 
-    override suspend fun execute() {
+    override suspend fun execute(): AttemptAllocationStep {
         try {
             val response: AcknowledgedResponse = client.admin()
                 .indices()
                 .suspendUntil { updateSettings(UpdateSettingsRequest(buildSettings(), managedIndexMetaData.index), it) }
             handleResponse(response)
         } catch (e: Exception) {
-            logger.error(ERROR_MESSAGE, e)
-            stepStatus = StepStatus.FAILED
-            val mutableInfo = mutableMapOf("message" to ERROR_MESSAGE)
-            val errorMessage = e.message
-            if (errorMessage != null) mutableInfo["cause"] = errorMessage
-            info = mutableInfo.toMap()
+            handleException(e)
         }
+
+        return this
     }
 
     private fun buildSettings(): Settings {
@@ -63,13 +60,23 @@ class AttemptAllocationStep(
         return builder.build()
     }
 
+    private fun handleException(e: Exception) {
+        val message = getFailedMessage(indexName)
+        logger.error(message, e)
+        stepStatus = StepStatus.FAILED
+        val mutableInfo = mutableMapOf("message" to message)
+        val errorMessage = e.message
+        if (errorMessage != null) mutableInfo["cause"] = errorMessage
+        info = mutableInfo.toMap()
+    }
+
     private fun handleResponse(response: AcknowledgedResponse) {
         if (response.isAcknowledged) {
             stepStatus = StepStatus.COMPLETED
-            info = mapOf("message" to "Updated settings with allocation.")
+            info = mapOf("message" to getSuccessMessage(indexName))
         } else {
             stepStatus = StepStatus.FAILED
-            info = mapOf("message" to ERROR_MESSAGE)
+            info = mapOf("message" to getFailedMessage(indexName))
         }
     }
 
@@ -82,7 +89,8 @@ class AttemptAllocationStep(
     }
 
     companion object {
-        private const val ERROR_MESSAGE = "Failed to update settings with allocation."
         private const val SETTINGS_PREFIX = "index.routing.allocation."
+        fun getFailedMessage(index: String) = "Failed to update allocation setting [index=$index]"
+        fun getSuccessMessage(index: String) = "Successfully updated allocation setting [index=$index]"
     }
 }

src/main/kotlin/com/amazon/opendistroforelasticsearch/indexstatemanagement/step/close/AttemptCloseStep.kt

@@ -21,11 +21,13 @@ import com.amazon.opendistroforelasticsearch.indexstatemanagement.model.action.C
 import com.amazon.opendistroforelasticsearch.indexstatemanagement.model.managedindexmetadata.StepMetaData
 import com.amazon.opendistroforelasticsearch.indexstatemanagement.step.Step
 import org.apache.logging.log4j.LogManager
+import org.elasticsearch.ExceptionsHelper
 import org.elasticsearch.action.admin.indices.close.CloseIndexRequest
 import org.elasticsearch.action.admin.indices.close.CloseIndexResponse
 import org.elasticsearch.client.Client
 import org.elasticsearch.cluster.service.ClusterService
 import org.elasticsearch.snapshots.SnapshotInProgressException
+import org.elasticsearch.transport.RemoteTransportException
 
 class AttemptCloseStep(
     val clusterService: ClusterService,
@@ -41,34 +43,52 @@ class AttemptCloseStep(
     override fun isIdempotent() = true
 
     @Suppress("TooGenericExceptionCaught")
-    override suspend fun execute() {
-        val index = managedIndexMetaData.index
+    override suspend fun execute(): AttemptCloseStep {
         try {
-            logger.info("Executing close on $index")
             val closeIndexRequest = CloseIndexRequest()
-                .indices(index)
+                .indices(indexName)
 
             val response: CloseIndexResponse = client.admin().indices().suspendUntil { close(closeIndexRequest, it) }
-            logger.info("Close index for $index was acknowledged=${response.isAcknowledged}")
             if (response.isAcknowledged) {
                 stepStatus = StepStatus.COMPLETED
-                info = mapOf("message" to "Successfully closed index")
+                info = mapOf("message" to getSuccessMessage(indexName))
             } else {
+                val message = getFailedMessage(indexName)
+                logger.warn(message)
                 stepStatus = StepStatus.FAILED
-                info = mapOf("message" to "Failed to close index")
+                info = mapOf("message" to message)
+            }
+        } catch (e: RemoteTransportException) {
+            val cause = ExceptionsHelper.unwrapCause(e)
+            if (cause is SnapshotInProgressException) {
+                handleSnapshotException(cause)
+            } else {
+                handleException(cause as Exception)
             }
         } catch (e: SnapshotInProgressException) {
-            logger.warn("Failed to close index [index=$index] with snapshot in progress")
-            stepStatus = StepStatus.CONDITION_NOT_MET
-            info = mapOf("message" to "Index had snapshot in progress, retrying closing")
+            handleSnapshotException(e)
         } catch (e: Exception) {
-            logger.error("Failed to set index to close [index=$index]", e)
-            stepStatus = StepStatus.FAILED
-            val mutableInfo = mutableMapOf("message" to "Failed to set index to close")
-            val errorMessage = e.message
-            if (errorMessage != null) mutableInfo["cause"] = errorMessage
-            info = mutableInfo.toMap()
+            handleException(e)
         }
+
+        return this
+    }
+
+    private fun handleSnapshotException(e: SnapshotInProgressException) {
+        val message = getSnapshotMessage(indexName)
+        logger.warn(message, e)
+        stepStatus = StepStatus.CONDITION_NOT_MET
+        info = mapOf("message" to message)
+    }
+
+    private fun handleException(e: Exception) {
+        val message = getFailedMessage(indexName)
+        logger.error(message, e)
+        stepStatus = StepStatus.FAILED
+        val mutableInfo = mutableMapOf("message" to message)
+        val errorMessage = e.message
+        if (errorMessage != null) mutableInfo["cause"] = errorMessage
+        info = mutableInfo.toMap()
     }
 
     override fun getUpdatedManagedIndexMetaData(currentMetaData: ManagedIndexMetaData): ManagedIndexMetaData {
@@ -78,4 +98,10 @@ class AttemptCloseStep(
             info = info
         )
     }
+
+    companion object {
+        fun getFailedMessage(index: String) = "Failed to close index [index=$index]"
+        fun getSuccessMessage(index: String) = "Successfully closed index [index=$index]"
+        fun getSnapshotMessage(index: String) = "Index had snapshot in progress, retrying closing [index=$index]"
+    }
 }

src/main/kotlin/com/amazon/opendistroforelasticsearch/indexstatemanagement/step/delete/AttemptDeleteStep.kt

@@ -21,11 +21,13 @@ import com.amazon.opendistroforelasticsearch.indexstatemanagement.model.action.D
 import com.amazon.opendistroforelasticsearch.indexstatemanagement.model.managedindexmetadata.StepMetaData
 import com.amazon.opendistroforelasticsearch.indexstatemanagement.step.Step
 import org.apache.logging.log4j.LogManager
+import org.elasticsearch.ExceptionsHelper
 import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest
 import org.elasticsearch.action.support.master.AcknowledgedResponse
 import org.elasticsearch.client.Client
 import org.elasticsearch.cluster.service.ClusterService
 import org.elasticsearch.snapshots.SnapshotInProgressException
+import org.elasticsearch.transport.RemoteTransportException
 import java.lang.Exception
 
 class AttemptDeleteStep(
@@ -42,30 +44,51 @@ class AttemptDeleteStep(
     override fun isIdempotent() = true
 
     @Suppress("TooGenericExceptionCaught")
-    override suspend fun execute() {
+    override suspend fun execute(): AttemptDeleteStep {
         try {
             val response: AcknowledgedResponse = client.admin().indices()
-                .suspendUntil { delete(DeleteIndexRequest(managedIndexMetaData.index), it) }
+                .suspendUntil { delete(DeleteIndexRequest(indexName), it) }
 
             if (response.isAcknowledged) {
                 stepStatus = StepStatus.COMPLETED
-                info = mapOf("message" to "Deleted index")
+                info = mapOf("message" to getSuccessMessage(indexName))
             } else {
+                val message = getFailedMessage(indexName)
+                logger.warn(message)
                 stepStatus = StepStatus.FAILED
-                info = mapOf("message" to "Failed to delete index")
+                info = mapOf("message" to message)
+            }
+        } catch (e: RemoteTransportException) {
+            val cause = ExceptionsHelper.unwrapCause(e)
+            if (cause is SnapshotInProgressException) {
+                handleSnapshotException(cause)
+            } else {
+                handleException(cause as Exception)
             }
         } catch (e: SnapshotInProgressException) {
-            logger.warn("Failed to delete index [index=${managedIndexMetaData.index}] with snapshot in progress")
-            stepStatus = StepStatus.CONDITION_NOT_MET
-            info = mapOf("message" to "Index had snapshot in progress, retrying deletion")
+            handleSnapshotException(e)
         } catch (e: Exception) {
-            logger.error("Failed to delete index [index=${managedIndexMetaData.index}]", e)
-            stepStatus = StepStatus.FAILED
-            val mutableInfo = mutableMapOf("message" to "Failed to delete index")
-            val errorMessage = e.message
-            if (errorMessage != null) mutableInfo["cause"] = errorMessage
-            info = mutableInfo.toMap()
+            handleException(e)
         }
+
+        return this
+    }
+
+    private fun handleSnapshotException(e: SnapshotInProgressException) {
+        val message = getSnapshotMessage(indexName)
+        logger.warn(message, e)
+        stepStatus = StepStatus.CONDITION_NOT_MET
+        info = mapOf("message" to message)
+    }
+
+    private fun handleException(e: Exception) {
+        val message = getFailedMessage(indexName)
+        logger.error(message, e)
+        stepStatus = StepStatus.FAILED
+        val mutableInfo = mutableMapOf("message" to message)
+        val errorMessage = e.message
+        if (errorMessage != null) mutableInfo["cause"] = errorMessage
+        info = mutableInfo.toMap()
     }
 
     override fun getUpdatedManagedIndexMetaData(currentMetaData: ManagedIndexMetaData): ManagedIndexMetaData {
@@ -78,5 +101,8 @@ class AttemptDeleteStep(
 
     companion object {
         const val name = "attempt_delete"
+        fun getFailedMessage(index: String) = "Failed to delete index [index=$index]"
+        fun getSuccessMessage(index: String) = "Successfully deleted index [index=$index]"
+        fun getSnapshotMessage(index: String) = "Index had snapshot in progress, retrying deletion [index=$index]"
     }
 }