Add more error notification at fail points (opensearch-project#1000)

* Add more error notification at fail points Signed-off-by: bowenlan-amzn <[email protected]> * Handle exception gracefully Signed-off-by: bowenlan-amzn <[email protected]> * small fix Signed-off-by: bowenlan-amzn <[email protected]> --------- Signed-off-by: bowenlan-amzn <[email protected]>
bowenlan-amzn · Oct 13, 2023 · b98d962 · b98d962
1 parent c20a502
commit b98d962
Showing 1 changed file with 48 additions and 20 deletions.
diff --git a/src/main/kotlin/org/opensearch/indexmanagement/indexstatemanagement/ManagedIndexRunner.kt b/src/main/kotlin/org/opensearch/indexmanagement/indexstatemanagement/ManagedIndexRunner.kt
@@ -313,6 +313,7 @@ object ManagedIndexRunner :
             )
             if (result.metadataSaved) {
                 disableManagedIndexConfig(managedIndexConfig)
+                publishErrorNotification(policy, managedIndexMetaData)
             }
             return
         }
@@ -339,7 +340,10 @@ object ManagedIndexRunner :
                 managedIndexMetaData
                     .copy(actionMetaData = currentActionMetaData?.copy(failed = true), info = info)
             )
-            if (updated.metadataSaved) disableManagedIndexConfig(managedIndexConfig)
+            if (updated.metadataSaved) {
+                disableManagedIndexConfig(managedIndexConfig)
+                publishErrorNotification(policy, managedIndexMetaData)
+            }
             return
         }
 
@@ -365,7 +369,10 @@ object ManagedIndexRunner :
                         policyRetryInfo = PolicyRetryInfoMetaData(true, 0), info = info
                     )
                 )
-                if (updated.metadataSaved) disableManagedIndexConfig(managedIndexConfig)
+                if (updated.metadataSaved) {
+                    disableManagedIndexConfig(managedIndexConfig)
+                    publishErrorNotification(policy, managedIndexMetaData)
+                }
                 return
             }
         }
@@ -379,7 +386,10 @@ object ManagedIndexRunner :
                     policyRetryInfo = PolicyRetryInfoMetaData(true, 0), info = info
                 )
             )
-            if (updated.metadataSaved) disableManagedIndexConfig(managedIndexConfig)
+            if (updated.metadataSaved) {
+                disableManagedIndexConfig(managedIndexConfig)
+                publishErrorNotification(policy, managedIndexMetaData)
+            }
             return
         }
 
@@ -392,7 +402,10 @@ object ManagedIndexRunner :
                     policyRetryInfo = PolicyRetryInfoMetaData(true, 0), info = info
                 )
             )
-            if (updated.metadataSaved) disableManagedIndexConfig(managedIndexConfig)
+            if (updated.metadataSaved) {
+                disableManagedIndexConfig(managedIndexConfig)
+                publishErrorNotification(policy, managedIndexMetaData)
+            }
             return
         }
 
@@ -434,16 +447,7 @@ object ManagedIndexRunner :
             var executedManagedIndexMetaData = startingManagedIndexMetaData.getCompletedManagedIndexMetaData(action, step)
 
             if (executedManagedIndexMetaData.isFailed) {
-                try {
-                    // if the policy has no error_notification this will do nothing otherwise it will try to send the configured error message
-                    publishErrorNotification(policy, executedManagedIndexMetaData)
-                } catch (e: Exception) {
-                    logger.error("Failed to publish error notification", e)
-                    val errorMessage = e.message ?: "Failed to publish error notification"
-                    val mutableInfo = executedManagedIndexMetaData.info?.toMutableMap() ?: mutableMapOf()
-                    mutableInfo["errorNotificationFailure"] = errorMessage
-                    executedManagedIndexMetaData = executedManagedIndexMetaData.copy(info = mutableInfo.toMap())
-                }
+                executedManagedIndexMetaData = publishErrorNotification(policy, executedManagedIndexMetaData)
             }
 
             if (executedManagedIndexMetaData.isSuccessfulDelete) {
@@ -810,13 +814,37 @@ object ManagedIndexRunner :
         }
     }
 
-    private suspend fun publishErrorNotification(policy: Policy, managedIndexMetaData: ManagedIndexMetaData) {
-        policy.errorNotification?.run {
-            errorNotificationRetryPolicy.retry(logger) {
-                val compiledMessage = compileTemplate(messageTemplate, managedIndexMetaData)
-                destination?.buildLegacyBaseMessage(null, compiledMessage)?.publishLegacyNotification(client)
-                channel?.sendNotification(client, ErrorNotification.CHANNEL_TITLE, managedIndexMetaData, compiledMessage, policy.user)
+    private suspend fun publishErrorNotification(policy: Policy, metadata: ManagedIndexMetaData): ManagedIndexMetaData {
+        return try {
+            val errorNotification = policy.errorNotification
+            if (errorNotification != null) {
+                policy.errorNotification.run {
+                    errorNotificationRetryPolicy.retry(logger) {
+                        val compiledMessage = compileTemplate(messageTemplate, metadata)
+                        destination?.buildLegacyBaseMessage(null, compiledMessage)?.publishLegacyNotification(client)
+                        channel?.sendNotification(
+                            client,
+                            ErrorNotification.CHANNEL_TITLE,
+                            metadata,
+                            compiledMessage,
+                            policy.user
+                        )
+                    }
+                }
+                val message = "Successfully published error notification [index = ${metadata.index}]"
+                logger.info(message)
+                val mutableInfo = metadata.info?.toMutableMap() ?: mutableMapOf()
+                mutableInfo["error_notification"] = message
+                metadata.copy(info = mutableInfo.toMap())
+            } else {
+                return metadata
             }
+        } catch (e: Exception) {
+            logger.error("Failed to publish error notification", e)
+            val errorMessage = e.message ?: "Failed to publish error notification"
+            val mutableInfo = metadata.info?.toMutableMap() ?: mutableMapOf()
+            mutableInfo["error_notification"] = errorMessage
+            metadata.copy(info = mutableInfo.toMap())
         }
     }