separate doc-level monitor query indices for externally defined monitors

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>
opensearch-project · Sep 25, 2024 · b280761 · b280761
1 parent 452a51c
commit b280761
Show file tree

Hide file tree

Showing 5 changed files with 11 additions and 13 deletions.
diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/DocumentLevelMonitorRunner.kt b/alerting/src/main/kotlin/org/opensearch/alerting/DocumentLevelMonitorRunner.kt
@@ -375,12 +375,6 @@ class DocumentLevelMonitorRunner : MonitorRunner() {
                 monitorCtx.docLevelMonitorQueries!!.deleteDocLevelQueriesOnDryRun(monitorMetadata)
             }
 
-            if (monitor.dataSources.queryIndex.contains("optimized")) {
-                val ack = monitorCtx.docLevelMonitorQueries!!.deleteDocLevelQueryIndex(monitor.dataSources)
-                if (!ack) {
-                    logger.error("Deletion of concrete queryIndex:${monitor.dataSources.queryIndex} is not ack'd!")
-                }
-            }
             // TODO: Update the Document as part of the Trigger and return back the trigger action result
             return monitorResult.copy(triggerResults = triggerResults, inputResults = inputRunResults)
         } catch (e: Exception) {
@@ -392,16 +386,16 @@ class DocumentLevelMonitorRunner : MonitorRunner() {
                 RestStatus.INTERNAL_SERVER_ERROR,
                 e
             )
-            if (monitor.dataSources.queryIndex.contains("optimized") &&
+            return monitorResult.copy(error = alertingException, inputResults = InputRunResults(emptyList(), alertingException))
+        } finally {
+            if (monitor.deleteQueryIndexInEveryRun == true &&
                 monitorCtx.docLevelMonitorQueries!!.docLevelQueryIndexExists(monitor.dataSources)
             ) {
                 val ack = monitorCtx.docLevelMonitorQueries!!.deleteDocLevelQueryIndex(monitor.dataSources)
                 if (!ack) {
-                    logger.error("Retry deletion of concrete queryIndex:${monitor.dataSources.queryIndex} is not ack'd!")
+                    logger.error("Deletion of concrete queryIndex:${monitor.dataSources.queryIndex} is not ack'd! for monitor ${monitor.id}")
                 }
             }
-            return monitorResult.copy(error = alertingException, inputResults = InputRunResults(emptyList(), alertingException))
-        } finally {
             val endTime = System.currentTimeMillis()
             totalTimeTakenStat = endTime - startTime
             logger.debug(

diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/service/DeleteMonitorService.kt b/alerting/src/main/kotlin/org/opensearch/alerting/service/DeleteMonitorService.kt
@@ -95,7 +95,7 @@ object DeleteMonitorService :
 
     private suspend fun deleteDocLevelMonitorQueriesAndIndices(monitor: Monitor) {
         try {
-            if (monitor.owner == "alerting") {
+            if (monitor.deleteQueryIndexInEveryRun == false) {
                 val metadata = MonitorMetadataService.getMetadata(monitor)
                 metadata?.sourceToQueryIndexMapping?.forEach { (_, queryIndex) ->
 

diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportIndexMonitorAction.kt b/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportIndexMonitorAction.kt
@@ -538,7 +538,7 @@ class TransportIndexMonitorAction @Inject constructor(
                         request.monitor.isMonitorOfStandardType() &&
                         Monitor.MonitorType.valueOf(request.monitor.monitorType.uppercase(Locale.ROOT)) ==
                         Monitor.MonitorType.DOC_LEVEL_MONITOR &&
-                        request.monitor.owner == "alerting"
+                        request.monitor.deleteQueryIndexInEveryRun == false
                     ) {
                         indexDocLevelMonitorQueries(request.monitor, indexResponse.id, metadata, request.refreshPolicy)
                     }
@@ -711,7 +711,7 @@ class TransportIndexMonitorAction @Inject constructor(
                                 .execute(it)
                         }
                     }
-                    if (currentMonitor.owner == "alerting") {
+                    if (currentMonitor.deleteQueryIndexInEveryRun == false) {
                         indexDocLevelMonitorQueries(
                             request.monitor,
                             currentMonitor.id,

diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/util/DocLevelMonitorQueries.kt b/alerting/src/main/kotlin/org/opensearch/alerting/util/DocLevelMonitorQueries.kt
@@ -445,6 +445,7 @@ class DocLevelMonitorQueries(private val client: Client, private val clusterServ
                     )
                 )
             indexRequests.add(indexRequest)
+            log.debug("query $query added for execution of monitor $monitorId on index $sourceIndex")
         }
         log.debug("bulk inserting percolate [${queries.size}] queries")
         if (indexRequests.isNotEmpty()) {

diff --git a/...-monitor-plugin/src/main/java/org/opensearch/alerting/SampleRemoteMonitorRestHandler.java b/...-monitor-plugin/src/main/java/org/opensearch/alerting/SampleRemoteMonitorRestHandler.java
@@ -94,6 +94,7 @@ protected RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient
                                 "id", null)), trigger1Serialized)),
                 Map.of(),
                 new DataSources(),
+                true,
                 "sample-remote-monitor-plugin"
         );
         IndexMonitorRequest indexMonitorRequest1 = new IndexMonitorRequest(
@@ -154,6 +155,7 @@ public void onFailure(Exception e) {
                     List.of(),
                     Map.of(),
                     new DataSources(),
+                    true,
                     "sample-remote-monitor-plugin"
             );
             IndexMonitorRequest indexMonitorRequest2 = new IndexMonitorRequest(
@@ -237,6 +239,7 @@ public void onFailure(Exception e) {
                                     "id", null)), trigger1Serialized)),
                     Map.of(),
                     new DataSources(),
+                    true,
                     "sample-remote-monitor-plugin"
             );
             IndexMonitorRequest indexDocLevelMonitorRequest = new IndexMonitorRequest(