optimize execution of workflow consisting of bucket-level followed by doc-level monitors

alerting/src/main/kotlin/org/opensearch/alerting/DocumentLevelMonitorRunner.kt

@@ -138,7 +138,8 @@ class DocumentLevelMonitorRunner : MonitorRunner() {
             }
 
             // Map of document ids per index when monitor is workflow delegate and has chained findings
-            val matchingDocIdsPerIndex = workflowRunContext?.matchingDocIdsPerIndex
+            val matchingDocIdsPerIndex = workflowRunContext?.matchingDocIdsPerIndex?.first
+            val findingIdsForMatchingDocIds = workflowRunContext?.matchingDocIdsPerIndex?.second
 
             val concreteIndicesSeenSoFar = mutableListOf<String>()
             val updatedIndexNames = mutableListOf<String>()
@@ -226,6 +227,7 @@ class DocumentLevelMonitorRunner : MonitorRunner() {
                         concreteIndices,
                         conflictingFields.toList(),
                         matchingDocIdsPerIndex?.get(concreteIndexName),
+                        findingIdsForMatchingDocIds
                     )
 
                     val shards = mutableSetOf<String>()

alerting/src/main/kotlin/org/opensearch/alerting/InputService.kt

@@ -90,7 +90,7 @@ class InputService(
                             periodStart = periodStart,
                             periodEnd = periodEnd,
                             prevResult = prevResult,
-                            matchingDocIdsPerIndex = matchingDocIdsPerIndex,
+                            matchingDocIdsPerIndex = matchingDocIdsPerIndex?.first,
                             returnSampleDocs = false
                         )
                         val searchResponse: SearchResponse = client.suspendUntil { client.search(searchRequest, it) }

alerting/src/main/kotlin/org/opensearch/alerting/WorkflowService.kt

@@ -41,15 +41,16 @@ class WorkflowService(
      * @param chainedMonitors Monitors that have previously executed
      * @param workflowExecutionId Execution id of the current workflow
      */
-    suspend fun getFindingDocIdsByExecutionId(chainedMonitors: List<Monitor>, workflowExecutionId: String): Map<String, List<String>> {
+    suspend fun getFindingDocIdsByExecutionId(chainedMonitors: List<Monitor>, workflowExecutionId: String):
+        Pair<Map<String, List<String>>, List<String>> {
         if (chainedMonitors.isEmpty())
-            return emptyMap()
+            return Pair(emptyMap(), listOf())
         val dataSources = chainedMonitors[0].dataSources
         try {
             val existsResponse: IndicesExistsResponse = client.admin().indices().suspendUntil {
                 exists(IndicesExistsRequest(dataSources.findingsIndex).local(true), it)
             }
-            if (existsResponse.isExists == false) return emptyMap()
+            if (existsResponse.isExists == false) return Pair(emptyMap(), listOf())
             // Search findings index to match id of monitors and workflow execution id
             val bqb = QueryBuilders.boolQuery()
                 .filter(
@@ -83,7 +84,7 @@ class WorkflowService(
             for (finding in findings) {
                 indexToRelatedDocIdsMap.getOrPut(finding.index) { mutableListOf() }.addAll(finding.relatedDocIds)
             }
-            return indexToRelatedDocIdsMap
+            return Pair(indexToRelatedDocIdsMap, findings.map { it.id })
         } catch (t: Exception) {
             log.error("Error getting finding doc ids: ${t.message}", t)
             throw AlertingException.wrap(t)

alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportDocLevelMonitorFanOutAction.kt

@@ -297,7 +297,11 @@ class TransportDocLevelMonitorFanOutAction
                     createFindings(monitor, docsToQueries, idQueryMap, true)
                 }
             } else {
-                if (monitor.ignoreFindingsAndAlerts == null || monitor.ignoreFindingsAndAlerts == false) {
+                /**
+                 * if should_persist_findings_and_alerts flag is not set, doc-level trigger generates alerts else doc-level trigger
+                 * generates a single alert with multiple findings.
+                 */
+                if (monitor.shouldPersistFindingsAndAlerts == null || monitor.shouldPersistFindingsAndAlerts == false) {
                     monitor.triggers.forEach {
                         triggerResults[it.id] = runForEachDocTrigger(
                             monitorResult,
@@ -312,9 +316,9 @@ class TransportDocLevelMonitorFanOutAction
                             workflowRunContext = workflowRunContext
                         )
                     }
-                } else if (monitor.ignoreFindingsAndAlerts == true) {
+                } else if (monitor.shouldPersistFindingsAndAlerts == true) {
                     monitor.triggers.forEach {
-                        triggerResults[it.id] = runForEachDocTriggerIgnoringFindingsAndAlerts(
+                        triggerResults[it.id] = runForEachDocTriggerWithoutPersistFindingsAndAlerts(
                             monitorResult,
                             it as DocumentLevelTrigger,
                             monitor,
@@ -363,7 +367,10 @@ class TransportDocLevelMonitorFanOutAction
         }
     }
 
-    private suspend fun runForEachDocTriggerIgnoringFindingsAndAlerts(
+    /**
+     * run doc-level triggers ignoring findings and alerts and generating a single alert.
+     */
+    private suspend fun runForEachDocTriggerWithoutPersistFindingsAndAlerts(
         monitorResult: MonitorRunResult<DocumentLevelTriggerRunResult>,
         trigger: DocumentLevelTrigger,
         monitor: Monitor,
@@ -374,9 +381,14 @@ class TransportDocLevelMonitorFanOutAction
     ): DocumentLevelTriggerRunResult {
         val triggerResult = triggerService.runDocLevelTrigger(monitor, trigger, queryToDocIds)
         if (triggerResult.triggeredDocs.isNotEmpty()) {
+            val findingIds = if (workflowRunContext?.matchingDocIdsPerIndex?.second != null) {
+                workflowRunContext.matchingDocIdsPerIndex.second
+            } else {
+                listOf()
+            }
             val triggerCtx = DocumentLevelTriggerExecutionContext(monitor, trigger)
             val alert = alertService.composeDocLevelAlert(
-                listOf(),
+                findingIds,
                 triggerResult.triggeredDocs,
                 triggerCtx,
                 monitorResult.alertError() ?: triggerResult.alertError(),
@@ -570,7 +582,7 @@ class TransportDocLevelMonitorFanOutAction
                     .string()
             log.debug("Findings: $findingStr")
 
-            if (shouldCreateFinding and (monitor.ignoreFindingsAndAlerts == null || monitor.ignoreFindingsAndAlerts == false)) {
+            if (shouldCreateFinding and (monitor.shouldPersistFindingsAndAlerts == null || monitor.shouldPersistFindingsAndAlerts == false)) {
                 indexRequests += IndexRequest(monitor.dataSources.findingsIndex)
                     .source(findingStr, XContentType.JSON)
                     .id(finding.id)
@@ -582,7 +594,7 @@ class TransportDocLevelMonitorFanOutAction
             bulkIndexFindings(monitor, indexRequests)
         }
 
-        if (monitor.ignoreFindingsAndAlerts == null || monitor.ignoreFindingsAndAlerts == false) {
+        if (monitor.shouldPersistFindingsAndAlerts == null || monitor.shouldPersistFindingsAndAlerts == false) {
             try {
                 findings.forEach { finding ->
                     publishFinding(monitor, finding)
@@ -945,11 +957,11 @@ class TransportDocLevelMonitorFanOutAction
         val boolQueryBuilder = BoolQueryBuilder()
         boolQueryBuilder.filter(QueryBuilders.rangeQuery("_seq_no").gt(prevSeqNo).lte(maxSeqNo))
 
-        if (monitor.ignoreFindingsAndAlerts == null || monitor.ignoreFindingsAndAlerts == false) {
+        if (monitor.shouldPersistFindingsAndAlerts == null || monitor.shouldPersistFindingsAndAlerts == false) {
             if (!docIds.isNullOrEmpty()) {
                 boolQueryBuilder.filter(QueryBuilders.termsQuery("_id", docIds))
             }
-        } else if (monitor.ignoreFindingsAndAlerts == true) {
+        } else if (monitor.shouldPersistFindingsAndAlerts == true) {
             val docIdsParam = mutableListOf<String>()
             if (docIds != null) {
                 docIdsParam.addAll(docIds)

alerting/src/main/kotlin/org/opensearch/alerting/workflow/CompositeWorkflowRunner.kt

@@ -95,7 +95,7 @@ object CompositeWorkflowRunner : WorkflowRunner() {
         var lastErrorDelegateRun: Exception? = null
 
         for (delegate in delegates) {
-            var indexToDocIds = mapOf<String, List<String>>()
+            var indexToDocIdsWithFindings: Pair<Map<String, List<String>>, List<String>>? = Pair(mapOf(), listOf())
             var delegateMonitor: Monitor
             delegateMonitor = monitorsById[delegate.monitorId]
                 ?: throw AlertingException.wrap(
@@ -118,7 +118,7 @@ object CompositeWorkflowRunner : WorkflowRunner() {
                 }
 
                 try {
-                    indexToDocIds = monitorCtx.workflowService!!.getFindingDocIdsByExecutionId(chainedMonitors, executionId)
+                    indexToDocIdsWithFindings = monitorCtx.workflowService!!.getFindingDocIdsByExecutionId(chainedMonitors, executionId)
                 } catch (e: Exception) {
                     logger.error("Failed to execute workflow due to failure in chained findings.  Error: ${e.message}", e)
                     return WorkflowRunResult(
@@ -131,7 +131,7 @@ object CompositeWorkflowRunner : WorkflowRunner() {
                 workflowId = workflowMetadata.workflowId,
                 workflowMetadataId = workflowMetadata.id,
                 chainedMonitorId = delegate.chainedMonitorFindings?.monitorId,
-                matchingDocIdsPerIndex = indexToDocIds,
+                matchingDocIdsPerIndex = indexToDocIdsWithFindings!!,
                 auditDelegateMonitorAlerts = if (workflow.auditDelegateMonitorAlerts == null) true
                 else workflow.auditDelegateMonitorAlerts!!
             )

sample-remote-monitor-plugin/src/main/java/org/opensearch/alerting/SampleRemoteMonitorRestHandler.java

@@ -95,6 +95,7 @@ protected RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient
                 Map.of(),
                 new DataSources(),
                 false,
+                false,
                 "sample-remote-monitor-plugin"
         );
         IndexMonitorRequest indexMonitorRequest1 = new IndexMonitorRequest(
@@ -156,6 +157,7 @@ public void onFailure(Exception e) {
                     Map.of(),
                     new DataSources(),
                     false,
+                    false,
                     "sample-remote-monitor-plugin"
             );
             IndexMonitorRequest indexMonitorRequest2 = new IndexMonitorRequest(
@@ -240,6 +242,7 @@ public void onFailure(Exception e) {
                     Map.of(),
                     new DataSources(),
                     false,
+                    false,
                     "sample-remote-monitor-plugin"
             );
             IndexMonitorRequest indexDocLevelMonitorRequest = new IndexMonitorRequest(