add workflowId field in alert (opensearch-project#463)

Signed-off-by: Surya Sashank Nistala <[email protected]>
AWSHurneyt · Jun 22, 2023 · 110bf01 · 110bf01
1 parent f1c0f1d
commit 110bf01
Show file tree

Hide file tree

Showing 4 changed files with 33 additions and 15 deletions.
diff --git a/src/main/kotlin/org/opensearch/commons/alerting/model/Alert.kt b/src/main/kotlin/org/opensearch/commons/alerting/model/Alert.kt
@@ -22,6 +22,7 @@ data class Alert(
     val version: Long = NO_VERSION,
     val schemaVersion: Int = NO_SCHEMA_VERSION,
     val monitorId: String,
+    val workflowId: String,
     val monitorName: String,
     val monitorVersion: Long,
     val monitorUser: User?,
@@ -48,6 +49,7 @@ data class Alert(
         }
     }
 
+    // constructor for chained alerts.
     constructor(
         startTime: Instant,
         lastNotificationTime: Instant?,
@@ -63,7 +65,7 @@ data class Alert(
         lastNotificationTime = lastNotificationTime, errorMessage = errorMessage, errorHistory = emptyList(),
         severity = chainedAlertTrigger.severity, actionExecutionResults = emptyList(), schemaVersion = schemaVersion,
         aggregationResultBucket = null, findingIds = emptyList(), relatedDocIds = emptyList(),
-        executionId = executionId
+        executionId = executionId, workflowId = workflow.id
     )
 
     constructor(
@@ -76,14 +78,15 @@ data class Alert(
         errorHistory: List<AlertError> = mutableListOf(),
         actionExecutionResults: List<ActionExecutionResult> = mutableListOf(),
         schemaVersion: Int = NO_SCHEMA_VERSION,
-        executionId: String? = null
+        executionId: String? = null,
+        workflow: Workflow? = null,
     ) : this(
         monitorId = monitor.id, monitorName = monitor.name, monitorVersion = monitor.version, monitorUser = monitor.user,
         triggerId = trigger.id, triggerName = trigger.name, state = state, startTime = startTime,
         lastNotificationTime = lastNotificationTime, errorMessage = errorMessage, errorHistory = errorHistory,
         severity = trigger.severity, actionExecutionResults = actionExecutionResults, schemaVersion = schemaVersion,
         aggregationResultBucket = null, findingIds = emptyList(), relatedDocIds = emptyList(),
-        executionId = executionId
+        executionId = executionId, workflowId = workflow?.id ?: ""
     )
 
     constructor(
@@ -97,14 +100,15 @@ data class Alert(
         actionExecutionResults: List<ActionExecutionResult> = mutableListOf(),
         schemaVersion: Int = NO_SCHEMA_VERSION,
         findingIds: List<String> = emptyList(),
-        executionId: String? = null
+        executionId: String? = null,
+        workflow: Workflow? = null,
     ) : this(
         monitorId = monitor.id, monitorName = monitor.name, monitorVersion = monitor.version, monitorUser = monitor.user,
         triggerId = trigger.id, triggerName = trigger.name, state = state, startTime = startTime,
         lastNotificationTime = lastNotificationTime, errorMessage = errorMessage, errorHistory = errorHistory,
         severity = trigger.severity, actionExecutionResults = actionExecutionResults, schemaVersion = schemaVersion,
         aggregationResultBucket = null, findingIds = findingIds, relatedDocIds = emptyList(),
-        executionId = executionId
+        executionId = executionId, workflowId = workflow?.id ?: ""
     )
 
     constructor(
@@ -119,14 +123,15 @@ data class Alert(
         schemaVersion: Int = NO_SCHEMA_VERSION,
         aggregationResultBucket: AggregationResultBucket,
         findingIds: List<String> = emptyList(),
-        executionId: String? = null
+        executionId: String? = null,
+        workflow: Workflow? = null,
     ) : this(
         monitorId = monitor.id, monitorName = monitor.name, monitorVersion = monitor.version, monitorUser = monitor.user,
         triggerId = trigger.id, triggerName = trigger.name, state = state, startTime = startTime,
         lastNotificationTime = lastNotificationTime, errorMessage = errorMessage, errorHistory = errorHistory,
         severity = trigger.severity, actionExecutionResults = actionExecutionResults, schemaVersion = schemaVersion,
         aggregationResultBucket = aggregationResultBucket, findingIds = findingIds, relatedDocIds = emptyList(),
-        executionId = executionId
+        executionId = executionId, workflowId = workflow?.id ?: ""
     )
 
     constructor(
@@ -142,14 +147,15 @@ data class Alert(
         errorHistory: List<AlertError> = mutableListOf(),
         actionExecutionResults: List<ActionExecutionResult> = mutableListOf(),
         schemaVersion: Int = NO_SCHEMA_VERSION,
-        executionId: String? = null
+        executionId: String? = null,
+        workflow: Workflow? = null,
     ) : this(
         id = id, monitorId = monitor.id, monitorName = monitor.name, monitorVersion = monitor.version, monitorUser = monitor.user,
         triggerId = trigger.id, triggerName = trigger.name, state = state, startTime = startTime,
         lastNotificationTime = lastNotificationTime, errorMessage = errorMessage, errorHistory = errorHistory,
         severity = trigger.severity, actionExecutionResults = actionExecutionResults, schemaVersion = schemaVersion,
         aggregationResultBucket = null, findingIds = findingIds, relatedDocIds = relatedDocIds,
-        executionId = executionId
+        executionId = executionId, workflowId = workflow?.id ?: ""
     )
 
     constructor(
@@ -161,13 +167,15 @@ data class Alert(
         state: State = State.ERROR,
         errorMessage: String,
         errorHistory: List<AlertError> = mutableListOf(),
-        schemaVersion: Int = NO_SCHEMA_VERSION
+        schemaVersion: Int = NO_SCHEMA_VERSION,
+        workflow: Workflow? = null,
     ) : this(
         id = id, monitorId = monitor.id, monitorName = monitor.name, monitorVersion = monitor.version, monitorUser = monitor.user,
         triggerId = trigger.id, triggerName = trigger.name, state = state, startTime = startTime,
         lastNotificationTime = lastNotificationTime, errorMessage = errorMessage, errorHistory = errorHistory,
         severity = trigger.severity, actionExecutionResults = listOf(), schemaVersion = schemaVersion,
-        aggregationResultBucket = null, findingIds = listOf(), relatedDocIds = listOf()
+        aggregationResultBucket = null, findingIds = listOf(), relatedDocIds = listOf(),
+        workflowId = workflow?.id ?: ""
     )
 
     enum class State {
@@ -185,6 +193,7 @@ data class Alert(
         version = sin.readLong(),
         schemaVersion = sin.readInt(),
         monitorId = sin.readString(),
+        workflowId = sin.readString(),
         monitorName = sin.readString(),
         monitorVersion = sin.readLong(),
         monitorUser = if (sin.readBoolean()) {
@@ -215,6 +224,7 @@ data class Alert(
         out.writeLong(version)
         out.writeInt(schemaVersion)
         out.writeString(monitorId)
+        out.writeString(workflowId)
         out.writeString(monitorName)
         out.writeLong(monitorVersion)
         out.writeBoolean(monitorUser != null)
@@ -247,6 +257,7 @@ data class Alert(
         const val SCHEMA_VERSION_FIELD = "schema_version"
         const val ALERT_VERSION_FIELD = "version"
         const val MONITOR_ID_FIELD = "monitor_id"
+        const val WORKFLOW_ID_FIELD = "workflow_id"
         const val MONITOR_VERSION_FIELD = "monitor_version"
         const val MONITOR_NAME_FIELD = "monitor_name"
         const val MONITOR_USER_FIELD = "monitor_user"
@@ -274,6 +285,7 @@ data class Alert(
         fun parse(xcp: XContentParser, id: String = NO_ID, version: Long = NO_VERSION): Alert {
 
             lateinit var monitorId: String
+            var workflowId = ""
             var schemaVersion = NO_SCHEMA_VERSION
             lateinit var monitorName: String
             var monitorVersion: Long = Versions.NOT_FOUND
@@ -300,6 +312,7 @@ data class Alert(
 
                 when (fieldName) {
                     MONITOR_ID_FIELD -> monitorId = xcp.text()
+                    WORKFLOW_ID_FIELD -> workflowId = xcp.text()
                     SCHEMA_VERSION_FIELD -> schemaVersion = xcp.intValue()
                     MONITOR_NAME_FIELD -> monitorName = xcp.text()
                     MONITOR_VERSION_FIELD -> monitorVersion = xcp.longValue()
@@ -360,7 +373,7 @@ data class Alert(
                 lastNotificationTime = lastNotificationTime, acknowledgedTime = acknowledgedTime,
                 errorMessage = errorMessage, errorHistory = errorHistory, severity = severity,
                 actionExecutionResults = actionExecutionResults, aggregationResultBucket = aggAlertBucket, findingIds = findingIds,
-                relatedDocIds = relatedDocIds, executionId = executionId
+                relatedDocIds = relatedDocIds, executionId = executionId, workflowId = workflowId
             )
         }
 
@@ -383,6 +396,7 @@ data class Alert(
             .field(ALERT_ID_FIELD, id)
             .field(ALERT_VERSION_FIELD, version)
             .field(MONITOR_ID_FIELD, monitorId)
+            .field(WORKFLOW_ID_FIELD, workflowId)
             .field(SCHEMA_VERSION_FIELD, schemaVersion)
             .field(MONITOR_VERSION_FIELD, monitorVersion)
             .field(MONITOR_NAME_FIELD, monitorName)

diff --git a/src/test/kotlin/org/opensearch/commons/alerting/AlertTests.kt b/src/test/kotlin/org/opensearch/commons/alerting/AlertTests.kt
@@ -77,5 +77,6 @@ class AlertTests {
         val alert = randomChainedAlert(workflow = workflow, trigger = trigger)
         assertEquals(alert.monitorId, "")
         assertEquals(alert.id, "")
+        assertEquals(workflow.id, alert.workflowId)
     }
 }
diff --git a/src/test/kotlin/org/opensearch/commons/alerting/action/AcknowledgeAlertResponseTests.kt b/src/test/kotlin/org/opensearch/commons/alerting/action/AcknowledgeAlertResponseTests.kt
@@ -18,15 +18,15 @@ class AcknowledgeAlertResponseTests {
 
         val acknowledged = mutableListOf(
             Alert(
-                "1234", 0L, 1, "monitor-1234", "test-monitor", 0L, randomUser(),
+                "1234", 0L, 1, "monitor-1234", "", "test-monitor", 0L, randomUser(),
                 "trigger-14", "test-trigger", ArrayList(), ArrayList(), Alert.State.ACKNOWLEDGED,
                 Instant.now(), Instant.now(), Instant.now(), Instant.now(), null, ArrayList(),
                 "sev-2", ArrayList(), null
             )
         )
         val failed = mutableListOf(
             Alert(
-                "1234", 0L, 1, "monitor-1234", "test-monitor", 0L, randomUser(),
+                "1234", 0L, 1, "monitor-1234", "", "test-monitor", 0L, randomUser(),
                 "trigger-14", "test-trigger", ArrayList(), ArrayList(), Alert.State.ERROR, Instant.now(), Instant.now(),
                 Instant.now(), Instant.now(), null, mutableListOf(AlertError(Instant.now(), "Error msg")),
                 "sev-2", mutableListOf(ActionExecutionResult("7890", null, 0)), null

diff --git a/src/test/kotlin/org/opensearch/commons/alerting/action/GetAlertsResponseTests.kt b/src/test/kotlin/org/opensearch/commons/alerting/action/GetAlertsResponseTests.kt
@@ -36,6 +36,7 @@ class GetAlertsResponseTests {
             0L,
             0,
             "monitorId",
+            "workflowId",
             "monitorName",
             0L,
             randomUser(),
@@ -64,6 +65,7 @@ class GetAlertsResponseTests {
         assertEquals(1, newReq.alerts.size)
         assertEquals(alert, newReq.alerts[0])
         assertEquals(1, newReq.totalAlerts)
+        assertEquals(newReq.alerts[0].workflowId, "workflowId")
     }
 
     @Test
@@ -75,6 +77,7 @@ class GetAlertsResponseTests {
             0L,
             0,
             "monitorId",
+            "workflowId",
             "monitorName",
             0L,
             null,
@@ -96,7 +99,7 @@ class GetAlertsResponseTests {
         val req = GetAlertsResponse(listOf(alert), 1)
         var actualXContentString = req.toXContent(builder(), ToXContent.EMPTY_PARAMS).string()
         val expectedXContentString = "{\"alerts\":[{\"id\":\"id\",\"version\":0,\"monitor_id\":\"monitorId\"," +
-            "\"schema_version\":0,\"monitor_version\":0,\"monitor_name\":\"monitorName\"," +
+            "\"workflow_id\":\"workflowId\",\"schema_version\":0,\"monitor_version\":0,\"monitor_name\":\"monitorName\"," +
             "\"execution_id\":null,\"trigger_id\":\"triggerId\",\"trigger_name\":\"triggerName\"," +
             "\"finding_ids\":[],\"related_doc_ids\":[],\"state\":\"ACKNOWLEDGED\",\"error_message\":null,\"alert_history\":[]," +
             "\"severity\":\"severity\",\"action_execution_results\":[],\"start_time\":" + now.toEpochMilli() +