Added workflow execution logic

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

@@ -8,6 +8,7 @@ package org.opensearch.alerting
 import org.opensearch.action.ActionRequest
 import org.opensearch.action.ActionResponse
 import org.opensearch.alerting.action.ExecuteMonitorAction
+import org.opensearch.alerting.action.ExecuteWorkflowAction
 import org.opensearch.alerting.action.GetDestinationsAction
 import org.opensearch.alerting.action.GetEmailAccountAction
 import org.opensearch.alerting.action.GetEmailGroupAction
@@ -27,6 +28,7 @@ import org.opensearch.alerting.core.settings.ScheduledJobSettings
 import org.opensearch.alerting.resthandler.RestAcknowledgeAlertAction
 import org.opensearch.alerting.resthandler.RestDeleteMonitorAction
 import org.opensearch.alerting.resthandler.RestExecuteMonitorAction
+import org.opensearch.alerting.resthandler.RestExecuteWorkflowAction
 import org.opensearch.alerting.resthandler.RestGetAlertsAction
 import org.opensearch.alerting.resthandler.RestGetDestinationsAction
 import org.opensearch.alerting.resthandler.RestGetEmailAccountAction
@@ -46,6 +48,7 @@ import org.opensearch.alerting.transport.TransportAcknowledgeAlertAction
 import org.opensearch.alerting.transport.TransportDeleteMonitorAction
 import org.opensearch.alerting.transport.TransportDeleteWorkflowAction
 import org.opensearch.alerting.transport.TransportExecuteMonitorAction
+import org.opensearch.alerting.transport.TransportExecuteWorkflowAction
 import org.opensearch.alerting.transport.TransportGetAlertsAction
 import org.opensearch.alerting.transport.TransportGetDestinationsAction
 import org.opensearch.alerting.transport.TransportGetEmailAccountAction
@@ -60,6 +63,7 @@ import org.opensearch.alerting.transport.TransportSearchEmailGroupAction
 import org.opensearch.alerting.transport.TransportSearchMonitorAction
 import org.opensearch.alerting.util.DocLevelMonitorQueries
 import org.opensearch.alerting.util.destinationmigration.DestinationMigrationCoordinator
+import org.opensearch.alerting.workflow.WorkflowRunnerService
 import org.opensearch.client.Client
 import org.opensearch.cluster.metadata.IndexNameExpressionResolver
 import org.opensearch.cluster.node.DiscoveryNodes
@@ -146,6 +150,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
     }
 
     lateinit var runner: MonitorRunnerService
+    lateinit var workflowRunner: WorkflowRunnerService
     lateinit var scheduler: JobScheduler
     lateinit var sweeper: JobSweeper
     lateinit var scheduledJobIndices: ScheduledJobIndices
@@ -170,6 +175,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
             RestIndexMonitorAction(),
             RestSearchMonitorAction(settings, clusterService),
             RestExecuteMonitorAction(),
+            RestExecuteWorkflowAction(),
             RestAcknowledgeAlertAction(),
             RestScheduledJobStatsHandler("_alerting"),
             RestSearchEmailAccountAction(),
@@ -200,7 +206,8 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
             ActionPlugin.ActionHandler(AlertingActions.GET_FINDINGS_ACTION_TYPE, TransportGetFindingsSearchAction::class.java),
             ActionPlugin.ActionHandler(AlertingActions.INDEX_WORKFLOW_ACTION_TYPE, TransportIndexWorkflowAction::class.java),
             ActionPlugin.ActionHandler(AlertingActions.GET_WORKFLOW_ACTION_TYPE, TransportGetWorkflowAction::class.java),
-            ActionPlugin.ActionHandler(AlertingActions.DELETE_WORKFLOW_ACTION_TYPE, TransportDeleteWorkflowAction::class.java)
+            ActionPlugin.ActionHandler(AlertingActions.DELETE_WORKFLOW_ACTION_TYPE, TransportDeleteWorkflowAction::class.java),
+            ActionPlugin.ActionHandler(ExecuteWorkflowAction.INSTANCE, TransportExecuteWorkflowAction::class.java)
         )
     }
 
@@ -248,6 +255,22 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
             .registerDocLevelMonitorQueries(DocLevelMonitorQueries(client, clusterService))
             .registerConsumers()
             .registerDestinationSettings()
+        workflowRunner = WorkflowRunnerService
+            .registerClusterService(clusterService)
+            .registerClient(client)
+            .registerNamedXContentRegistry(xContentRegistry)
+            .registerScriptService(scriptService)
+            .registerIndexNameExpressionResolver(indexNameExpressionResolver)
+            .registerSettings(settings)
+            .registerThreadPool(threadPool)
+            .registerAlertIndices(alertIndices)
+            .registerInputService(InputService(client, scriptService, namedWriteableRegistry, xContentRegistry))
+            .registerTriggerService(TriggerService(scriptService))
+            .registerAlertService(AlertService(client, xContentRegistry, alertIndices))
+            .registerDocLevelMonitorQueries(DocLevelMonitorQueries(client, clusterService))
+            .registerWorkflowService(WorkflowService(client, xContentRegistry))
+            .registerConsumers()
+            .registerDestinationSettings()
         scheduledJobIndices = ScheduledJobIndices(client.admin(), clusterService)
         docLevelMonitorQueries = DocLevelMonitorQueries(client, clusterService)
         scheduler = JobScheduler(threadPool, runner)
@@ -263,6 +286,13 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
             settings
         )
 
+        WorkflowMetadataService.initialize(
+            client,
+            clusterService,
+            xContentRegistry,
+            settings
+        )
+
         return listOf(sweeper, scheduler, runner, scheduledJobIndices, docLevelMonitorQueries, destinationMigrationCoordinator)
     }
 

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

@@ -25,6 +25,7 @@ import org.opensearch.alerting.util.defaultToPerExecutionAction
 import org.opensearch.alerting.util.getActionExecutionPolicy
 import org.opensearch.alerting.util.getBucketKeysHash
 import org.opensearch.alerting.util.getCombinedTriggerRunResult
+import org.opensearch.alerting.workflow.WorkflowRunContext
 import org.opensearch.common.xcontent.LoggingDeprecationHandler
 import org.opensearch.common.xcontent.XContentType
 import org.opensearch.commons.alerting.model.Alert
@@ -59,7 +60,8 @@ object BucketLevelMonitorRunner : MonitorRunner() {
         monitorCtx: MonitorRunnerExecutionContext,
         periodStart: Instant,
         periodEnd: Instant,
-        dryrun: Boolean
+        dryrun: Boolean,
+        workflowRunContext: WorkflowRunContext?
     ): MonitorRunResult<BucketLevelTriggerRunResult> {
         val roles = MonitorRunnerService.getRolesForMonitor(monitor)
         logger.debug("Running monitor: ${monitor.name} with roles: $roles Thread: ${Thread.currentThread().name}")
@@ -118,7 +120,8 @@ object BucketLevelMonitorRunner : MonitorRunner() {
                     monitor,
                     periodStart,
                     periodEnd,
-                    monitorResult.inputResults
+                    monitorResult.inputResults,
+                    workflowRunContext
                 )
                 if (firstIteration) {
                     firstPageOfInputResults = inputResults
@@ -154,7 +157,8 @@ object BucketLevelMonitorRunner : MonitorRunner() {
                             monitorCtx,
                             periodStart,
                             periodEnd,
-                            !dryrun && monitor.id != Monitor.NO_ID
+                            !dryrun && monitor.id != Monitor.NO_ID,
+                            workflowRunContext
                         )
                     } else {
                         emptyList()
@@ -350,7 +354,8 @@ object BucketLevelMonitorRunner : MonitorRunner() {
         monitorCtx: MonitorRunnerExecutionContext,
         periodStart: Instant,
         periodEnd: Instant,
-        shouldCreateFinding: Boolean
+        shouldCreateFinding: Boolean,
+        workflowRunContext: WorkflowRunContext? = null
     ): List<String> {
         monitor.inputs.forEach { input ->
             if (input is SearchInput) {
@@ -361,14 +366,14 @@ object BucketLevelMonitorRunner : MonitorRunner() {
                 for (aggFactory in (query.aggregations() as AggregatorFactories.Builder).aggregatorFactories) {
                     when (aggFactory) {
                         is CompositeAggregationBuilder -> {
-                            var grouByFields = 0 // if number of fields used to group by > 1 we won't calculate findings
+                            var groupByFields = 0 // if number of fields used to group by > 1 we won't calculate findings
                             val sources = aggFactory.sources()
                             for (source in sources) {
-                                if (grouByFields > 0) {
+                                if (groupByFields > 0) {
                                     logger.error("grouByFields > 0. not generating findings for bucket level monitor ${monitor.id}")
                                     return listOf()
                                 }
-                                grouByFields++
+                                groupByFields++
                                 fieldName = source.field()
                             }
                         }
@@ -409,7 +414,7 @@ object BucketLevelMonitorRunner : MonitorRunner() {
                             sr.source().query(queryBuilder)
                         }
                     val searchResponse: SearchResponse = monitorCtx.client!!.suspendUntil { monitorCtx.client!!.search(sr, it) }
-                    return createFindingPerIndex(searchResponse, monitor, monitorCtx, shouldCreateFinding)
+                    return createFindingPerIndex(searchResponse, monitor, monitorCtx, shouldCreateFinding, workflowRunContext?.executionId)
                 } else {
                     logger.error("Couldn't resolve groupBy field. Not generating bucket level monitor findings for monitor %${monitor.id}")
                 }
@@ -422,7 +427,8 @@ object BucketLevelMonitorRunner : MonitorRunner() {
         searchResponse: SearchResponse,
         monitor: Monitor,
         monitorCtx: MonitorRunnerExecutionContext,
-        shouldCreateFinding: Boolean
+        shouldCreateFinding: Boolean,
+        workflowExecutionId: String? = null
     ): List<String> {
         val docIdsByIndexName: MutableMap<String, MutableList<String>> = mutableMapOf()
         for (hit in searchResponse.hits.hits) {
@@ -441,7 +447,8 @@ object BucketLevelMonitorRunner : MonitorRunner() {
                     monitorName = monitor.name,
                     index = it.key,
                     timestamp = Instant.now(),
-                    docLevelQueries = listOf()
+                    docLevelQueries = listOf(),
+                    executionId = workflowExecutionId
                 )
 
                 val findingStr = finding.toXContent(XContentBuilder.builder(XContentType.JSON.xContent()), ToXContent.EMPTY_PARAMS).string()

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

@@ -24,6 +24,7 @@ import org.opensearch.alerting.util.AlertingException
 import org.opensearch.alerting.util.IndexUtils
 import org.opensearch.alerting.util.defaultToPerExecutionAction
 import org.opensearch.alerting.util.getActionExecutionPolicy
+import org.opensearch.alerting.workflow.WorkflowRunContext
 import org.opensearch.client.Client
 import org.opensearch.cluster.metadata.IndexMetadata
 import org.opensearch.cluster.routing.ShardRouting
@@ -62,7 +63,8 @@ object DocumentLevelMonitorRunner : MonitorRunner() {
         monitorCtx: MonitorRunnerExecutionContext,
         periodStart: Instant,
         periodEnd: Instant,
-        dryrun: Boolean
+        dryrun: Boolean,
+        workflowRunContext: WorkflowRunContext?
     ): MonitorRunResult<DocumentLevelTriggerRunResult> {
         logger.debug("Document-level-monitor is running ...")
         val isTempMonitor = dryrun || monitor.id == Monitor.NO_ID
@@ -88,7 +90,8 @@ object DocumentLevelMonitorRunner : MonitorRunner() {
         var (monitorMetadata, _) = MonitorMetadataService.getOrCreateMetadata(
             monitor = monitor,
             createWithRunContext = false,
-            skipIndex = isTempMonitor
+            skipIndex = isTempMonitor,
+            workflowRunContext?.workflowMetadataId
         )
 
         val docLevelMonitorInput = monitor.inputs[0] as DocLevelMonitorInput
@@ -127,6 +130,9 @@ object DocumentLevelMonitorRunner : MonitorRunner() {
                 }
             }
 
+            // Map of document ids per index when monitor is workflow delegate and has chained findings
+            val matchingDocIdsPerIndex = workflowRunContext?.matchingDocIdsPerIndex
+
             indices.forEach { indexName ->
                 // Prepare lastRunContext for each index
                 val indexLastRunContext = lastRunContext.getOrPut(indexName) {
@@ -161,7 +167,13 @@ object DocumentLevelMonitorRunner : MonitorRunner() {
                 // Prepare DocumentExecutionContext for each index
                 val docExecutionContext = DocumentExecutionContext(queries, indexLastRunContext, indexUpdatedRunContext)
 
-                val matchingDocs = getMatchingDocs(monitor, monitorCtx, docExecutionContext, indexName)
+                val matchingDocs = getMatchingDocs(
+                    monitor,
+                    monitorCtx,
+                    docExecutionContext,
+                    indexName,
+                    matchingDocIdsPerIndex?.get(indexName)
+                )
 
                 if (matchingDocs.isNotEmpty()) {
                     val matchedQueriesForDocs = getMatchedQueries(
@@ -214,7 +226,8 @@ object DocumentLevelMonitorRunner : MonitorRunner() {
                 idQueryMap,
                 docsToQueries,
                 queryToDocIds,
-                dryrun
+                dryrun,
+                workflowRunContext?.executionId
             )
         }
 
@@ -238,7 +251,8 @@ object DocumentLevelMonitorRunner : MonitorRunner() {
         idQueryMap: Map<String, DocLevelQuery>,
         docsToQueries: Map<String, List<String>>,
         queryToDocIds: Map<DocLevelQuery, Set<String>>,
-        dryrun: Boolean
+        dryrun: Boolean,
+        workflowExecutionId: String? = null
     ): DocumentLevelTriggerRunResult {
         val triggerCtx = DocumentLevelTriggerExecutionContext(monitor, trigger)
         val triggerResult = monitorCtx.triggerService!!.runDocLevelTrigger(monitor, trigger, queryToDocIds)
@@ -249,7 +263,14 @@ object DocumentLevelMonitorRunner : MonitorRunner() {
         // TODO: Implement throttling for findings
         docsToQueries.forEach {
             val triggeredQueries = it.value.map { queryId -> idQueryMap[queryId]!! }
-            val findingId = createFindings(monitor, monitorCtx, triggeredQueries, it.key, !dryrun && monitor.id != Monitor.NO_ID)
+            val findingId = createFindings(
+                monitor,
+                monitorCtx,
+                triggeredQueries,
+                it.key,
+                !dryrun && monitor.id != Monitor.NO_ID,
+                workflowExecutionId
+            )
             findings.add(findingId)
 
             if (triggerResult.triggeredDocs.contains(it.key)) {
@@ -329,7 +350,8 @@ object DocumentLevelMonitorRunner : MonitorRunner() {
         monitorCtx: MonitorRunnerExecutionContext,
         docLevelQueries: List<DocLevelQuery>,
         matchingDocId: String,
-        shouldCreateFinding: Boolean
+        shouldCreateFinding: Boolean,
+        workflowExecutionId: String? = null,
     ): String {
         // Before the "|" is the doc id and after the "|" is the index
         val docIndex = matchingDocId.split("|")
@@ -341,7 +363,8 @@ object DocumentLevelMonitorRunner : MonitorRunner() {
             monitorName = monitor.name,
             index = docIndex[1],
             docLevelQueries = docLevelQueries,
-            timestamp = Instant.now()
+            timestamp = Instant.now(),
+            executionId = workflowExecutionId
         )
 
         val findingStr = finding.toXContent(XContentBuilder.builder(XContentType.JSON.xContent()), ToXContent.EMPTY_PARAMS).string()
@@ -458,7 +481,8 @@ object DocumentLevelMonitorRunner : MonitorRunner() {
         monitor: Monitor,
         monitorCtx: MonitorRunnerExecutionContext,
         docExecutionCtx: DocumentExecutionContext,
-        index: String
+        index: String,
+        docIds: List<String>? = null
     ): List<Pair<String, BytesReference>> {
         val count: Int = docExecutionCtx.updatedLastRunContext["shards_count"] as Int
         val matchingDocs = mutableListOf<Pair<String, BytesReference>>()
@@ -474,7 +498,8 @@ object DocumentLevelMonitorRunner : MonitorRunner() {
                     shard,
                     prevSeqNo,
                     maxSeqNo,
-                    null
+                    null,
+                    docIds
                 )
 
                 if (hits.hits.isNotEmpty()) {
@@ -493,7 +518,8 @@ object DocumentLevelMonitorRunner : MonitorRunner() {
         shard: String,
         prevSeqNo: Long?,
         maxSeqNo: Long,
-        query: String?
+        query: String?,
+        docIds: List<String>? = null
     ): SearchHits {
         if (prevSeqNo?.equals(maxSeqNo) == true && maxSeqNo != 0L) {
             return SearchHits.empty()
@@ -505,6 +531,10 @@ object DocumentLevelMonitorRunner : MonitorRunner() {
             boolQueryBuilder.must(QueryBuilders.queryStringQuery(query))
         }
 
+        if (!docIds.isNullOrEmpty()) {
+            boolQueryBuilder.filter(QueryBuilders.termsQuery("_id", docIds))
+        }
+
         val request: SearchRequest = SearchRequest()
             .indices(index)
             .preference("_shards:$shard")