Fix get alerts api to also return chained alerts with default params.…

… and skip audit state alerts (#1020) * fix get alerts api with default params to return chained alerts and skip audit alerts Signed-off-by: Surya Sashank Nistala <snistala@amazon.com> * fix alert state query filter for get workflow alerts to avoid audit alerts and fetch only chained alerts Signed-off-by: Surya Sashank Nistala <snistala@amazon.com> * fix typo Signed-off-by: Surya Sashank Nistala <snistala@amazon.com> --------- Signed-off-by: Surya Sashank Nistala <snistala@amazon.com>
opensearch-project · Jul 13, 2023 · 0add91f · 0add91f
1 parent 3ba65f2
commit 0add91f
Show file tree

Hide file tree

Showing 5 changed files with 68 additions and 15 deletions.
diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/MonitorMetadataService.kt b/alerting/src/main/kotlin/org/opensearch/alerting/MonitorMetadataService.kt
@@ -120,7 +120,7 @@ object MonitorMetadataService :
         monitor: Monitor,
         createWithRunContext: Boolean = true,
         skipIndex: Boolean = false,
-        workflowMetadataId: String? = null,
+        workflowMetadataId: String? = null
     ): Pair<MonitorMetadata, Boolean> {
         try {
             val created = true
@@ -154,7 +154,7 @@ object MonitorMetadataService :
                     XContentType.JSON
                 )
                 XContentParserUtils.ensureExpectedToken(XContentParser.Token.START_OBJECT, xcp.nextToken(), xcp)
-                MonitorMetadata.parse(xcp)
+                MonitorMetadata.parse(xcp, getResponse.id, getResponse.seqNo, getResponse.primaryTerm)
             } else {
                 null
             }

diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportGetAlertsAction.kt b/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportGetAlertsAction.kt
@@ -58,7 +58,7 @@ class TransportGetAlertsAction @Inject constructor(
     clusterService: ClusterService,
     actionFilters: ActionFilters,
     val settings: Settings,
-    val xContentRegistry: NamedXContentRegistry
+    val xContentRegistry: NamedXContentRegistry,
 ) : HandledTransportAction<ActionRequest, GetAlertsResponse>(
     AlertingActions.GET_ALERTS_ACTION_NAME,
     transportService,
@@ -77,7 +77,7 @@ class TransportGetAlertsAction @Inject constructor(
     override fun doExecute(
         task: Task,
         request: ActionRequest,
-        actionListener: ActionListener<GetAlertsResponse>
+        actionListener: ActionListener<GetAlertsResponse>,
     ) {
         val getAlertsRequest = request as? GetAlertsRequest
             ?: recreateObject(request) { GetAlertsRequest(it) }
@@ -97,7 +97,15 @@ class TransportGetAlertsAction @Inject constructor(
             queryBuilder.filter(QueryBuilders.termQuery("severity", getAlertsRequest.severityLevel))
         }
 
-        if (getAlertsRequest.alertState != "ALL") {
+        if (getAlertsRequest.alertState == "ALL") {
+            // alerting dashboards expects chained alerts and individually executed monitors' alerts to be returned from this api
+            // when invoked with state=ALL. They require that audit alerts are NOT returned in this page
+            // and only be shown in "associated alerts" field under get workflow_alerts API.
+            // But if the API is called with query_params: state=AUDIT,monitor_id=<123>,workflow_id=<abc>, this api
+            // will return audit alerts generated by delegate monitor <123> in workflow <abc>
+            QueryBuilders.boolQuery()
+                .filter(QueryBuilders.boolQuery().mustNot(QueryBuilders.termsQuery(Alert.STATE_FIELD, Alert.State.AUDIT.name)))
+        } else {
             queryBuilder.filter(QueryBuilders.termQuery("state", getAlertsRequest.alertState))
         }
 
@@ -107,16 +115,23 @@ class TransportGetAlertsAction @Inject constructor(
 
         if (getAlertsRequest.monitorId != null) {
             queryBuilder.filter(QueryBuilders.termQuery("monitor_id", getAlertsRequest.monitorId))
+            if (getAlertsRequest.workflowIds.isNullOrEmpty()) {
+                val noWorkflowIdQuery = QueryBuilders.boolQuery()
+                    .should(QueryBuilders.boolQuery().mustNot(QueryBuilders.existsQuery(Alert.WORKFLOW_ID_FIELD)))
+                    .should(QueryBuilders.termsQuery(Alert.WORKFLOW_ID_FIELD, ""))
+                queryBuilder.must(noWorkflowIdQuery)
+            }
         } else if (getAlertsRequest.monitorIds.isNullOrEmpty() == false) {
             queryBuilder.filter(QueryBuilders.termsQuery("monitor_id", getAlertsRequest.monitorIds))
+            if (getAlertsRequest.workflowIds.isNullOrEmpty()) {
+                val noWorkflowIdQuery = QueryBuilders.boolQuery()
+                    .should(QueryBuilders.boolQuery().mustNot(QueryBuilders.existsQuery(Alert.WORKFLOW_ID_FIELD)))
+                    .should(QueryBuilders.termsQuery(Alert.WORKFLOW_ID_FIELD, ""))
+                queryBuilder.must(noWorkflowIdQuery)
+            }
         }
         if (getAlertsRequest.workflowIds.isNullOrEmpty() == false) {
             queryBuilder.must(QueryBuilders.termsQuery("workflow_id", getAlertsRequest.workflowIds))
-        } else {
-            val noWorklfowIdQuery = QueryBuilders.boolQuery()
-                .should(QueryBuilders.boolQuery().mustNot(QueryBuilders.existsQuery(Alert.WORKFLOW_ID_FIELD)))
-                .should(QueryBuilders.termsQuery(Alert.WORKFLOW_ID_FIELD, ""))
-            queryBuilder.must(noWorklfowIdQuery)
         }
         if (!tableProp.searchString.isNullOrBlank()) {
             queryBuilder
@@ -196,7 +211,7 @@ class TransportGetAlertsAction @Inject constructor(
         alertIndex: String,
         searchSourceBuilder: SearchSourceBuilder,
         actionListener: ActionListener<GetAlertsResponse>,
-        user: User?
+        user: User?,
     ) {
         // user is null when: 1/ security is disabled. 2/when user is super-admin.
         if (user == null) {

diff --git a/...ing/src/main/kotlin/org/opensearch/alerting/transport/TransportGetWorkflowAlertsAction.kt b/...ing/src/main/kotlin/org/opensearch/alerting/transport/TransportGetWorkflowAlertsAction.kt
@@ -98,8 +98,11 @@ class TransportGetWorkflowAlertsAction @Inject constructor(
             queryBuilder.filter(QueryBuilders.termQuery("severity", getWorkflowAlertsRequest.severityLevel))
         }
 
-        if (getWorkflowAlertsRequest.alertState != "ALL") {
-            queryBuilder.filter(QueryBuilders.termQuery("state", getWorkflowAlertsRequest.alertState))
+        if (getWorkflowAlertsRequest.alertState == "ALL") {
+            QueryBuilders.boolQuery()
+                .filter(QueryBuilders.boolQuery().mustNot(QueryBuilders.termsQuery(Alert.STATE_FIELD, Alert.State.AUDIT.name)))
+        } else {
+            queryBuilder.filter(QueryBuilders.termQuery(Alert.STATE_FIELD, getWorkflowAlertsRequest.alertState))
         }
 
         if (getWorkflowAlertsRequest.alertIds.isNullOrEmpty() == false) {
@@ -148,7 +151,7 @@ class TransportGetWorkflowAlertsAction @Inject constructor(
     }
 
     fun resolveAlertsIndexName(getAlertsRequest: GetWorkflowAlertsRequest): String {
-        return if (getAlertsRequest.alertIndex.isNullOrEmpty()) AlertIndices.ALL_ALERT_INDEX_PATTERN
+        return if (getAlertsRequest.alertIndex.isNullOrEmpty()) AlertIndices.ALERT_INDEX
         else getAlertsRequest.alertIndex!!
     }
 

diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/workflow/CompositeWorkflowRunner.kt b/alerting/src/main/kotlin/org/opensearch/alerting/workflow/CompositeWorkflowRunner.kt
@@ -310,7 +310,7 @@ object CompositeWorkflowRunner : WorkflowRunner() {
                 .should(boolQuery().mustNot(existsQuery(Alert.ERROR_MESSAGE_FIELD)))
                 .should(termsQuery(Alert.ERROR_MESSAGE_FIELD, ""))
             queryBuilder.must(noErrorQuery)
-            searchRequest.source().query(queryBuilder)
+            searchRequest.source().query(queryBuilder).size(9999)
             val searchResponse: SearchResponse = monitorCtx.client!!.suspendUntil { monitorCtx.client!!.search(searchRequest, it) }
             val alerts = searchResponse.hits.map { hit ->
                 val xcp = XContentHelper.createParser(

diff --git a/alerting/src/test/kotlin/org/opensearch/alerting/MonitorDataSourcesIT.kt b/alerting/src/test/kotlin/org/opensearch/alerting/MonitorDataSourcesIT.kt
@@ -3942,6 +3942,23 @@ class MonitorDataSourcesIT : AlertingSingleNodeTestCase() {
         )
         var chainedAlerts = res.alerts
         Assert.assertTrue(chainedAlerts.size == 1)
+
+        // verify get alerts api with defaults set in query params returns only chained alerts and not audit alerts
+        val table = Table("asc", "id", null, 1, 0, "")
+        val getAlertsDefaultParamsResponse = client().execute(
+            AlertingActions.GET_ALERTS_ACTION_TYPE,
+            GetAlertsRequest(
+                table = table,
+                severityLevel = "ALL",
+                alertState = "ALL",
+                monitorId = null,
+                alertIndex = null,
+                monitorIds = null,
+                workflowIds = null,
+                alertIds = null
+            )
+        ).get()
+        Assert.assertEquals(getAlertsDefaultParamsResponse.alerts.size, 1)
         Assert.assertTrue(res.associatedAlerts.isEmpty())
         verifyAcknowledgeChainedAlerts(chainedAlerts, workflowId, 1)
         Assert.assertTrue(chainedAlerts[0].executionId == executeWorkflowResponse.workflowRunResult.executionId)
@@ -3985,6 +4002,20 @@ class MonitorDataSourcesIT : AlertingSingleNodeTestCase() {
         notTriggerResult = triggerResults[notTrigger.id]
         Assert.assertFalse(notTriggerResult!!.triggered)
         Assert.assertTrue(andTriggerResult!!.triggered)
+        val getAuditAlertsForMonitor1 = client().execute(
+            AlertingActions.GET_ALERTS_ACTION_TYPE,
+            GetAlertsRequest(
+                table = table,
+                severityLevel = "ALL",
+                alertState = "AUDIT",
+                monitorId = monitorResponse.id,
+                alertIndex = null,
+                monitorIds = null,
+                workflowIds = listOf(workflowId),
+                alertIds = null
+            )
+        ).get()
+        Assert.assertEquals(getAuditAlertsForMonitor1.alerts.size, 1)
         res = getWorkflowAlerts(workflowId)
         chainedAlerts = res.alerts
         Assert.assertTrue(chainedAlerts.size == 1)
@@ -4022,6 +4053,10 @@ class MonitorDataSourcesIT : AlertingSingleNodeTestCase() {
         assertAuditStateAlerts(monitorResponse2.id, alerts1)
         assertFindings(monitorResponse2.id, customFindingsIndex2, 1, 1, listOf("2"))
         verifyAcknowledgeChainedAlerts(chainedAlerts, workflowId, 1)
+        // test redundant executions of workflow dont query old data again to verify metadata updation works fine
+        val redundantExec = executeWorkflow(workflow)
+        Assert.assertFalse(redundantExec?.workflowRunResult!!.triggerResults[andTrigger.id]!!.triggered)
+        Assert.assertTrue(redundantExec.workflowRunResult.triggerResults[notTrigger.id]!!.triggered)
     }
 
     private fun getDelegateMonitorMetadataId(