initial alert v2 history implementation

Author: toepkerd-zz

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

@@ -135,6 +135,8 @@ import org.opensearch.threadpool.ThreadPool
 import org.opensearch.transport.client.Client
 import org.opensearch.watcher.ResourceWatcherService
 import java.util.function.Supplier
+import org.opensearch.alerting.alertsv2.AlertV2Indices
+import org.opensearch.alerting.alertsv2.AlertV2Mover
 
 /**
  * Entry point of the OpenDistro for Elasticsearch alerting plugin
@@ -178,9 +180,10 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
     lateinit var docLevelMonitorQueries: DocLevelMonitorQueries
     lateinit var threadPool: ThreadPool
     lateinit var alertIndices: AlertIndices
+    lateinit var alertV2Indices: AlertV2Indices
     lateinit var clusterService: ClusterService
     lateinit var destinationMigrationCoordinator: DestinationMigrationCoordinator
-    lateinit var alertV2Expirer: AlertV2Expirer
+    lateinit var alertV2Mover: AlertV2Mover
     var monitorTypeToMonitorRunners: MutableMap<String, RemoteMonitorRegistry> = mutableMapOf()
 
     override fun getRestHandlers(
@@ -314,6 +317,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
             .registerSettings(settings)
             .registerThreadPool(threadPool)
             .registerAlertIndices(alertIndices)
+            .registerAlertV2Indices(alertV2Indices)
             .registerInputService(
                 InputService(
                     client,
@@ -340,7 +344,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
         scheduler = JobScheduler(threadPool, runner)
         sweeper = JobSweeper(environment.settings(), client, clusterService, threadPool, xContentRegistry, scheduler, ALERTING_JOB_TYPES)
         destinationMigrationCoordinator = DestinationMigrationCoordinator(client, clusterService, threadPool, scheduledJobIndices)
-        alertV2Expirer = AlertV2Expirer(client, threadPool, clusterService)
+        alertV2Mover = AlertV2Mover(environment.settings(), client, threadPool, clusterService)
         this.threadPool = threadPool
         this.clusterService = clusterService
 
@@ -368,7 +372,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
             commentsIndices,
             docLevelMonitorQueries,
             destinationMigrationCoordinator,
-            alertV2Expirer,
+            alertV2Mover,
             lockService,
             alertService,
             triggerService

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

@@ -7,6 +7,7 @@ package org.opensearch.alerting
 
 import org.opensearch.action.bulk.BackoffPolicy
 import org.opensearch.alerting.alerts.AlertIndices
+import org.opensearch.alerting.alertsv2.AlertV2Indices
 import org.opensearch.alerting.core.lock.LockService
 import org.opensearch.alerting.model.destination.DestinationContextFactory
 import org.opensearch.alerting.remote.monitors.RemoteMonitorRegistry
@@ -35,6 +36,7 @@ data class MonitorRunnerExecutionContext(
     var settings: Settings? = null,
     var threadPool: ThreadPool? = null,
     var alertIndices: AlertIndices? = null,
+    var alertV2Indices: AlertV2Indices? = null,
     var inputService: InputService? = null,
     var triggerService: TriggerService? = null,
     var alertService: AlertService? = null,

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

@@ -94,6 +94,7 @@ import java.time.Instant
 import java.time.LocalDateTime
 import java.time.ZoneOffset
 import java.util.UUID
+import org.opensearch.alerting.alertsv2.AlertV2Indices
 import kotlin.coroutines.CoroutineContext
 
 object MonitorRunnerService : JobRunner, CoroutineScope, AbstractLifecycleComponent() {
@@ -145,6 +146,11 @@ object MonitorRunnerService : JobRunner, CoroutineScope, AbstractLifecycleCompon
         return this
     }
 
+    fun registerAlertV2Indices(alertV2Indices: AlertV2Indices): MonitorRunnerService {
+        this.monitorCtx.alertV2Indices = alertV2Indices
+        return this
+    }
+
     fun registerInputService(inputService: InputService): MonitorRunnerService {
         this.monitorCtx.inputService = inputService
         return this

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

@@ -44,6 +44,7 @@ import java.time.Instant
 import java.time.ZoneOffset.UTC
 import java.time.format.DateTimeFormatter
 import java.time.temporal.ChronoUnit
+import org.opensearch.alerting.alertsv2.AlertV2Indices
 
 object PPLMonitorRunner : MonitorV2Runner {
     private val logger = LogManager.getLogger(javaClass)
@@ -86,11 +87,10 @@ object PPLMonitorRunner : MonitorV2Runner {
         // use threadpool time for cross node consistency
         val timeOfCurrentExecution = Instant.ofEpochMilli(MonitorRunnerService.monitorCtx.threadPool!!.absoluteTimeInMillis())
 
-        // TODO: should alerting v1 and v2 alerts index be separate?
+        // TODO: put alertV2s in their own index
         try {
-            // TODO: write generated V2 alerts to existing alerts v1 index for now, revisit this decision
-            monitorCtx.alertIndices!!.createOrUpdateAlertIndex()
-            monitorCtx.alertIndices!!.createOrUpdateInitialAlertHistoryIndex()
+            monitorCtx.alertV2Indices!!.createOrUpdateAlertV2Index()
+            monitorCtx.alertV2Indices!!.createOrUpdateInitialAlertV2HistoryIndex()
         } catch (e: Exception) {
             val id = if (pplMonitor.id.trim().isEmpty()) "_na_" else pplMonitor.id
             logger.error("Error loading alerts for monitorV2: $id", e)
@@ -121,21 +121,6 @@ object PPLMonitorRunner : MonitorV2Runner {
                 }
                 logger.info("suppression check passed, executing trigger ${pplTrigger.name} from monitor ${pplMonitor.name}")
 
-//                internal fun isActionActionable(action: Action, alert: Alert?): Boolean {
-//                    if (alert != null && alert.state == Alert.State.AUDIT)
-//                        return false
-//                    if (alert == null || action.throttle == null) {
-//                        return true
-//                    }
-//                    if (action.throttleEnabled) {
-//                        val result = alert.actionExecutionResults.firstOrNull { r -> r.actionId == action.id }
-//                        val lastExecutionTime: Instant? = result?.lastExecutionTime
-//                        val throttledTimeBound = currentTime().minus(action.throttle!!.value.toLong(), action.throttle!!.unit)
-//                        return (lastExecutionTime == null || lastExecutionTime.isBefore(throttledTimeBound))
-//                    }
-//                    return true
-//                }
-
                 // if trigger uses custom condition, append the custom condition to query, otherwise simply proceed
                 val queryToExecute = if (pplTrigger.conditionType == ConditionType.NUMBER_OF_RESULTS) { // number of results trigger
                     timeFilteredQuery
@@ -147,15 +132,15 @@ object PPLMonitorRunner : MonitorV2Runner {
                 val queryResponseJson = executePplQuery(queryToExecute, nodeClient)
                 logger.info("query execution results for trigger ${pplTrigger.name}: $queryResponseJson")
 
-                // retrieve only the relevant query response rows.
+                // retrieve deep copies of only the relevant query response rows.
                 // for num_results triggers, that's the entire response
                 // for custom triggers, that's only rows that evaluated to true
                 val relevantQueryResultRows = if (pplTrigger.conditionType == ConditionType.NUMBER_OF_RESULTS) {
                     // number of results trigger
                     getQueryResponseWithoutSize(queryResponseJson)
                 } else {
                     // custom condition trigger
-                    evaluateCustomConditionTrigger(queryResponseJson, pplTrigger)
+                    collectCustomConditionResults(queryResponseJson, pplTrigger)
                 }
 
                 // retrieve the number of results
@@ -194,10 +179,6 @@ object PPLMonitorRunner : MonitorV2Runner {
                         timeOfCurrentExecution
                     )
 
-                    // collect the generated alerts to be written to alerts index
-                    // if the trigger is on result_set mode
-//                    generatedAlerts.addAll(thisTriggersGeneratedAlerts)
-
                     // update the trigger's last execution time for future suppression checks
                     pplTrigger.lastTriggeredTime = timeOfCurrentExecution
 
@@ -354,7 +335,7 @@ object PPLMonitorRunner : MonitorV2Runner {
         return queryResponseDeepCopy
     }
 
-    private fun evaluateCustomConditionTrigger(customConditionQueryResponse: JSONObject, pplTrigger: PPLTrigger): JSONObject {
+    private fun collectCustomConditionResults(customConditionQueryResponse: JSONObject, pplTrigger: PPLTrigger): JSONObject {
         // a PPL query with custom condition returning 0 results should imply a valid but not useful query.
         // do not trigger alert, but warn that query likely is not functioning as user intended
         if (customConditionQueryResponse.getLong("total") == 0L) {
@@ -445,7 +426,10 @@ object PPLMonitorRunner : MonitorV2Runner {
             alertV2s.add(alertV2)
         }
 
-        return alertV2s.toList() // return as immutable list
+        // TODO: this is a magic number right now, make it a setting
+        val alertsLimit = 10
+
+        return alertV2s.take(alertsLimit).toList() // return as immutable list
     }
 
     private fun generateErrorAlert(
@@ -488,7 +472,7 @@ object PPLMonitorRunner : MonitorV2Runner {
 
         var requestsToRetry = alerts.flatMap { alert ->
             listOf<DocWriteRequest<*>>(
-                IndexRequest(AlertIndices.ALERT_INDEX)
+                IndexRequest(AlertV2Indices.ALERT_V2_INDEX)
                     .routing(pplMonitor.id) // set routing ID to PPL Monitor ID
                     .source(alert.toXContent(XContentFactory.jsonBuilder(), ToXContent.EMPTY_PARAMS))
                     .id(if (alert.id != Alert.NO_ID) alert.id else null)