PPL Alerting: Adding PPL Related Models (#940) (#955)

* PPL Alerting: Adding PPL Related Models



* post load test optimization



* adding constructor for bwc



* extra tests and cleanup



* removing mention of SQL in names, generalizing query and queryResults fields in Alert, add extra serde tests



* removing experimental flag from PPLTrigger



* random PPL Monitor for tests now always includes at least 1 trigger



* improving serde tests and added comment about base ppl query



* small cleanup



* minor comments adjustment



* removing notifications ID validation check



---------

Signed-off-by: Dennis Toepker <toepkerd@amazon.com>
Co-authored-by: Dennis Toepker <toepkerd@amazon.com>
Co-authored-by: Peter Zhu <zhujiaxi@amazon.com>

Author: 3 people

src/main/kotlin/org/opensearch/commons/alerting/model/Alert.kt

@@ -1,7 +1,9 @@
 package org.opensearch.commons.alerting.model
 
+import org.opensearch.Version
 import org.opensearch.common.lucene.uid.Versions
 import org.opensearch.commons.alerting.alerts.AlertError
+import org.opensearch.commons.alerting.model.Monitor.Companion.suppressWarning
 import org.opensearch.commons.alerting.util.IndexUtils.Companion.NO_SCHEMA_VERSION
 import org.opensearch.commons.alerting.util.instant
 import org.opensearch.commons.alerting.util.optionalTimeField
@@ -43,7 +45,12 @@ data class Alert(
     val aggregationResultBucket: AggregationResultBucket? = null,
     val executionId: String? = null,
     val associatedAlertIds: List<String>,
-    val clusters: List<String>? = null
+    val clusters: List<String>? = null,
+    // these fields are specifically used for PPL Monitors
+    // for now, Query-Level Monitors can support including
+    // query results in Alerts by using these fields
+    val query: String? = null,
+    val queryResults: List<Map<String, Any?>> = listOf()
 ) : Writeable, ToXContent {
 
     init {
@@ -54,6 +61,64 @@ data class Alert(
         }
     }
 
+    constructor(
+        id: String = NO_ID,
+        version: Long = NO_VERSION,
+        schemaVersion: Int = NO_SCHEMA_VERSION,
+        monitorId: String,
+        workflowId: String,
+        workflowName: String,
+        monitorName: String,
+        monitorVersion: Long,
+        monitorUser: User?,
+        triggerId: String,
+        triggerName: String,
+        findingIds: List<String>,
+        relatedDocIds: List<String>,
+        state: State,
+        startTime: Instant,
+        endTime: Instant? = null,
+        lastNotificationTime: Instant? = null,
+        acknowledgedTime: Instant? = null,
+        errorMessage: String? = null,
+        errorHistory: List<AlertError>,
+        severity: String,
+        actionExecutionResults: List<ActionExecutionResult>,
+        aggregationResultBucket: AggregationResultBucket? = null,
+        executionId: String? = null,
+        associatedAlertIds: List<String>,
+        clusters: List<String>? = null
+    ) : this (
+        id = id,
+        version = version,
+        schemaVersion = schemaVersion,
+        monitorId = monitorId,
+        workflowId = workflowId,
+        workflowName = workflowName,
+        monitorName = monitorName,
+        monitorVersion = monitorVersion,
+        monitorUser = monitorUser,
+        triggerId = triggerId,
+        triggerName = triggerName,
+        findingIds = findingIds,
+        relatedDocIds = relatedDocIds,
+        state = state,
+        startTime = startTime,
+        endTime = endTime,
+        lastNotificationTime = lastNotificationTime,
+        acknowledgedTime = acknowledgedTime,
+        errorMessage = errorMessage,
+        errorHistory = errorHistory,
+        severity = severity,
+        actionExecutionResults = actionExecutionResults,
+        aggregationResultBucket = aggregationResultBucket,
+        executionId = executionId,
+        associatedAlertIds = associatedAlertIds,
+        clusters = clusters,
+        query = null,
+        queryResults = listOf()
+    )
+
     constructor(
         startTime: Instant,
         lastNotificationTime: Instant?,
@@ -87,12 +152,16 @@ data class Alert(
         workflowId = workflow.id,
         workflowName = workflow.name,
         associatedAlertIds = associatedAlertIds,
-        clusters = clusters
+        clusters = clusters,
+        query = null,
+        queryResults = listOf()
     )
 
+    // constructor for Alerts from QueryLevelMonitorRunner
+    // this monitor runner runs Query-level, Cluster Metrics, and PPL Monitors
     constructor(
         monitor: Monitor,
-        trigger: QueryLevelTrigger,
+        trigger: Trigger,
         startTime: Instant,
         lastNotificationTime: Instant?,
         state: State = State.ACTIVE,
@@ -102,7 +171,9 @@ data class Alert(
         schemaVersion: Int = NO_SCHEMA_VERSION,
         executionId: String? = null,
         workflowId: String? = null,
-        clusters: List<String>? = null
+        clusters: List<String>? = null,
+        query: String? = null,
+        queryResults: List<Map<String, Any?>> = listOf()
     ) : this(
         monitorId = monitor.id,
         monitorName = monitor.name,
@@ -125,7 +196,9 @@ data class Alert(
         workflowId = workflowId ?: "",
         workflowName = "",
         associatedAlertIds = emptyList(),
-        clusters = clusters
+        clusters = clusters,
+        query = query,
+        queryResults = queryResults
     )
 
     constructor(
@@ -164,7 +237,9 @@ data class Alert(
         workflowId = workflowId ?: "",
         workflowName = "",
         associatedAlertIds = emptyList(),
-        clusters = clusters
+        clusters = clusters,
+        query = null,
+        queryResults = listOf()
     )
 
     constructor(
@@ -204,7 +279,9 @@ data class Alert(
         workflowId = workflowId ?: "",
         workflowName = "",
         associatedAlertIds = emptyList(),
-        clusters = clusters
+        clusters = clusters,
+        query = null,
+        queryResults = listOf()
     )
 
     constructor(
@@ -246,7 +323,9 @@ data class Alert(
         workflowId = workflowId ?: "",
         workflowName = "",
         associatedAlertIds = emptyList(),
-        clusters = clusters
+        clusters = clusters,
+        query = null,
+        queryResults = listOf()
     )
 
     constructor(
@@ -285,7 +364,9 @@ data class Alert(
         workflowId = workflowId ?: "",
         executionId = executionId,
         associatedAlertIds = emptyList(),
-        clusters = clusters
+        clusters = clusters,
+        query = null,
+        queryResults = listOf()
     )
 
     enum class State {
@@ -329,7 +410,17 @@ data class Alert(
         aggregationResultBucket = if (sin.readBoolean()) AggregationResultBucket(sin) else null,
         executionId = sin.readOptionalString(),
         associatedAlertIds = sin.readStringList(),
-        clusters = sin.readOptionalStringList()
+        clusters = sin.readOptionalStringList(),
+        query = if (sin.version.onOrAfter(Version.V_3_5_0)) {
+            sin.readOptionalString()
+        } else {
+            null
+        },
+        queryResults = if (sin.version.onOrAfter(Version.V_3_5_0)) {
+            sin.readList { input -> suppressWarning(input.readMap()) }
+        } else {
+            listOf()
+        }
     )
 
     fun isAcknowledged(): Boolean = (state == State.ACKNOWLEDGED)
@@ -368,6 +459,12 @@ data class Alert(
         out.writeOptionalString(executionId)
         out.writeStringCollection(associatedAlertIds)
         out.writeOptionalStringArray(clusters?.toTypedArray())
+        if (out.version.onOrAfter(Version.V_3_5_0)) {
+            out.writeOptionalString(query)
+            out.writeCollection(queryResults) { output, map ->
+                output.writeMap(map)
+            }
+        }
     }
 
     companion object {
@@ -399,6 +496,8 @@ data class Alert(
         const val BUCKET_KEYS = AggregationResultBucket.BUCKET_KEYS
         const val PARENTS_BUCKET_PATH = AggregationResultBucket.PARENTS_BUCKET_PATH
         const val CLUSTERS_FIELD = "clusters"
+        const val QUERY_FIELD = "query"
+        const val QUERY_RESULTS_FIELD = "query_results"
         const val NO_ID = ""
         const val NO_VERSION = Versions.NOT_FOUND
 
@@ -430,6 +529,8 @@ data class Alert(
             var aggAlertBucket: AggregationResultBucket? = null
             val associatedAlertIds = mutableListOf<String>()
             val clusters = mutableListOf<String>()
+            var query: String? = null
+            var queryResults: List<Map<String, Any?>> = listOf()
             ensureExpectedToken(XContentParser.Token.START_OBJECT, xcp.currentToken(), xcp)
             while (xcp.nextToken() != XContentParser.Token.END_OBJECT) {
                 val fieldName = xcp.currentName()
@@ -505,6 +606,19 @@ data class Alert(
                             clusters.add(xcp.text())
                         }
                     }
+                    QUERY_FIELD -> {
+                        if (xcp.currentToken() != XContentParser.Token.VALUE_NULL) {
+                            query = xcp.text()
+                        }
+                    }
+                    QUERY_RESULTS_FIELD -> {
+                        ensureExpectedToken(XContentParser.Token.START_ARRAY, xcp.currentToken(), xcp)
+                        val resultsList = mutableListOf<Map<String, Any?>>()
+                        while (xcp.nextToken() != XContentParser.Token.END_ARRAY) {
+                            resultsList.add(xcp.map())
+                        }
+                        queryResults = resultsList
+                    }
                 }
             }
 
@@ -534,7 +648,9 @@ data class Alert(
                 workflowId = workflowId,
                 workflowName = workflowName,
                 associatedAlertIds = associatedAlertIds,
-                clusters = if (clusters.size > 0) clusters else null
+                clusters = if (clusters.size > 0) clusters else null,
+                query = query,
+                queryResults = queryResults
             )
         }
 
@@ -587,6 +703,11 @@ data class Alert(
 
         if (!clusters.isNullOrEmpty()) builder.field(CLUSTERS_FIELD, clusters.toTypedArray())
 
+        if (!query.isNullOrEmpty()) builder.field(QUERY_FIELD, query)
+        if (queryResults.isNotEmpty()) {
+            builder.field(QUERY_RESULTS_FIELD, queryResults.toTypedArray())
+        }
+
         builder.endObject()
         return builder
     }
@@ -611,7 +732,9 @@ data class Alert(
             PARENTS_BUCKET_PATH to aggregationResultBucket?.parentBucketPath,
             FINDING_IDS to findingIds.joinToString(","),
             RELATED_DOC_IDS to relatedDocIds.joinToString(","),
-            CLUSTERS_FIELD to clusters?.joinToString(",")
+            CLUSTERS_FIELD to clusters?.joinToString(","),
+            QUERY_FIELD to query,
+            QUERY_RESULTS_FIELD to queryResults
         )
     }
 }

src/main/kotlin/org/opensearch/commons/alerting/model/BucketLevelTrigger.kt

@@ -63,7 +63,7 @@ data class BucketLevelTrigger(
         out.writeCollection(actions)
     }
 
-    fun asTemplateArg(): Map<String, Any> {
+    override fun asTemplateArg(): Map<String, Any> {
         return mapOf(
             ID_FIELD to id,
             NAME_FIELD to name,

src/main/kotlin/org/opensearch/commons/alerting/model/ChainedAlertTrigger.kt

@@ -55,7 +55,7 @@ data class ChainedAlertTrigger(
     }
 
     /** Returns a representation of the trigger suitable for passing into painless and mustache scripts. */
-    fun asTemplateArg(): Map<String, Any> {
+    override fun asTemplateArg(): Map<String, Any> {
         return mapOf(
             ID_FIELD to id,
             NAME_FIELD to name,

src/main/kotlin/org/opensearch/commons/alerting/model/DocumentLevelTrigger.kt

@@ -55,7 +55,7 @@ data class DocumentLevelTrigger(
     }
 
     /** Returns a representation of the trigger suitable for passing into painless and mustache scripts. */
-    fun asTemplateArg(): Map<String, Any> {
+    override fun asTemplateArg(): Map<String, Any> {
         return mapOf(
             ID_FIELD to id,
             NAME_FIELD to name,

src/main/kotlin/org/opensearch/commons/alerting/model/Input.kt

@@ -2,6 +2,7 @@ package org.opensearch.commons.alerting.model
 
 import org.opensearch.commons.alerting.model.ClusterMetricsInput.Companion.URI_FIELD
 import org.opensearch.commons.alerting.model.DocLevelMonitorInput.Companion.DOC_LEVEL_INPUT_FIELD
+import org.opensearch.commons.alerting.model.PPLInput.Companion.PPL_INPUT_FIELD
 import org.opensearch.commons.alerting.model.SearchInput.Companion.SEARCH_FIELD
 import org.opensearch.commons.alerting.model.remote.monitors.RemoteDocLevelMonitorInput
 import org.opensearch.commons.alerting.model.remote.monitors.RemoteDocLevelMonitorInput.Companion.REMOTE_DOC_LEVEL_MONITOR_INPUT_FIELD
@@ -20,7 +21,8 @@ interface Input : BaseModel {
         CLUSTER_METRICS_INPUT(URI_FIELD),
         SEARCH_INPUT(SEARCH_FIELD),
         REMOTE_MONITOR_INPUT(REMOTE_MONITOR_INPUT_FIELD),
-        REMOTE_DOC_LEVEL_MONITOR_INPUT(REMOTE_DOC_LEVEL_MONITOR_INPUT_FIELD);
+        REMOTE_DOC_LEVEL_MONITOR_INPUT(REMOTE_DOC_LEVEL_MONITOR_INPUT_FIELD),
+        PPL_INPUT(PPL_INPUT_FIELD);
 
         override fun toString(): String {
             return value
@@ -42,8 +44,10 @@ interface Input : BaseModel {
                 DocLevelMonitorInput.parse(xcp)
             } else if (xcp.currentName() == Type.REMOTE_MONITOR_INPUT.value) {
                 RemoteMonitorInput.parse(xcp)
-            } else {
+            } else if (xcp.currentName() == Type.REMOTE_DOC_LEVEL_MONITOR_INPUT.value) {
                 RemoteDocLevelMonitorInput.parse(xcp)
+            } else {
+                PPLInput.parseInner(xcp)
             }
             XContentParserUtils.ensureExpectedToken(XContentParser.Token.END_OBJECT, xcp.nextToken(), xcp)
             return input
@@ -58,6 +62,7 @@ interface Input : BaseModel {
                 Type.SEARCH_INPUT -> SearchInput(sin)
                 Type.REMOTE_MONITOR_INPUT -> RemoteMonitorInput(sin)
                 Type.REMOTE_DOC_LEVEL_MONITOR_INPUT -> RemoteDocLevelMonitorInput(sin)
+                Type.PPL_INPUT -> PPLInput(sin)
                 // This shouldn't be reachable but ensuring exhaustiveness as Kotlin warns
                 // enum can be null in Java
                 else -> throw IllegalStateException("Unexpected input [$type] when reading Trigger")