PPL Alerting: Execute Monitor and Monitor Stats

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

@@ -15,6 +15,7 @@ import org.opensearch.alerting.action.GetRemoteIndexesAction
 import org.opensearch.alerting.action.SearchEmailAccountAction
 import org.opensearch.alerting.action.SearchEmailGroupAction
 import org.opensearch.alerting.actionv2.DeleteMonitorV2Action
+import org.opensearch.alerting.actionv2.ExecuteMonitorV2Action
 import org.opensearch.alerting.actionv2.GetAlertsV2Action
 import org.opensearch.alerting.actionv2.GetMonitorV2Action
 import org.opensearch.alerting.actionv2.IndexMonitorV2Action
@@ -32,6 +33,7 @@ import org.opensearch.alerting.core.action.node.ScheduledJobsStatsAction
 import org.opensearch.alerting.core.action.node.ScheduledJobsStatsTransportAction
 import org.opensearch.alerting.core.lock.LockService
 import org.opensearch.alerting.core.resthandler.RestScheduledJobStatsHandler
+import org.opensearch.alerting.core.resthandler.RestScheduledJobStatsV2Handler
 import org.opensearch.alerting.core.schedule.JobScheduler
 import org.opensearch.alerting.core.settings.AlertingV2Settings
 import org.opensearch.alerting.core.settings.LegacyOpenDistroScheduledJobSettings
@@ -62,6 +64,7 @@ import org.opensearch.alerting.resthandler.RestSearchEmailAccountAction
 import org.opensearch.alerting.resthandler.RestSearchEmailGroupAction
 import org.opensearch.alerting.resthandler.RestSearchMonitorAction
 import org.opensearch.alerting.resthandlerv2.RestDeleteMonitorV2Action
+import org.opensearch.alerting.resthandlerv2.RestExecuteMonitorV2Action
 import org.opensearch.alerting.resthandlerv2.RestGetAlertsV2Action
 import org.opensearch.alerting.resthandlerv2.RestGetMonitorV2Action
 import org.opensearch.alerting.resthandlerv2.RestIndexMonitorV2Action
@@ -99,6 +102,7 @@ import org.opensearch.alerting.transport.TransportSearchEmailAccountAction
 import org.opensearch.alerting.transport.TransportSearchEmailGroupAction
 import org.opensearch.alerting.transport.TransportSearchMonitorAction
 import org.opensearch.alerting.transportv2.TransportDeleteMonitorV2Action
+import org.opensearch.alerting.transportv2.TransportExecuteMonitorV2Action
 import org.opensearch.alerting.transportv2.TransportGetAlertsV2Action
 import org.opensearch.alerting.transportv2.TransportGetMonitorV2Action
 import org.opensearch.alerting.transportv2.TransportIndexMonitorV2Action
@@ -245,10 +249,12 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
 
             // Alerting V2
             RestIndexMonitorV2Action(),
+            RestExecuteMonitorV2Action(),
             RestDeleteMonitorV2Action(),
             RestGetMonitorV2Action(),
             RestSearchMonitorV2Action(settings, clusterService),
             RestGetAlertsV2Action(),
+            RestScheduledJobStatsV2Handler()
         )
     }
 
@@ -288,6 +294,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
             ActionPlugin.ActionHandler(GetMonitorV2Action.INSTANCE, TransportGetMonitorV2Action::class.java),
             ActionPlugin.ActionHandler(SearchMonitorV2Action.INSTANCE, TransportSearchMonitorV2Action::class.java),
             ActionPlugin.ActionHandler(DeleteMonitorV2Action.INSTANCE, TransportDeleteMonitorV2Action::class.java),
+            ActionPlugin.ActionHandler(ExecuteMonitorV2Action.INSTANCE, TransportExecuteMonitorV2Action::class.java),
             ActionPlugin.ActionHandler(GetAlertsV2Action.INSTANCE, TransportGetAlertsV2Action::class.java)
         )
     }

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

@@ -31,8 +31,8 @@ object AlertingV2Utils {
         if (scheduledJob is MonitorV2) {
             return IllegalStateException(
                 "The ID given corresponds to an Alerting V2 Monitor, but a V1 Monitor was expected. " +
-                    "If you wish to operate on a V1 Monitor (e.g. Per Query, Per Document, etc), please use " +
-                    "the Alerting V1 APIs with endpoint prefix: $MONITOR_BASE_URI."
+                    "If you wish to operate on a V2 Monitor (e.g. PPL Monitor), please use " +
+                    "the Alerting V2 APIs with endpoint prefix: $MONITOR_V2_BASE_URI."
             )
         } else if (scheduledJob !is Monitor && scheduledJob !is Workflow) {
             return IllegalStateException(
@@ -49,8 +49,8 @@ object AlertingV2Utils {
         if (scheduledJob is Monitor || scheduledJob is Workflow) {
             return IllegalStateException(
                 "The ID given corresponds to an Alerting V1 Monitor, but a V2 Monitor was expected. " +
-                    "If you wish to operate on a V2 Monitor (e.g. PPL Monitor), please use " +
-                    "the Alerting V2 APIs with endpoint prefix: $MONITOR_V2_BASE_URI."
+                    "If you wish to operate on a V1 Monitor (e.g. Per Query, Per Document, etc), please use " +
+                    "the Alerting V1 APIs with endpoint prefix: $MONITOR_BASE_URI."
             )
         } else if (scheduledJob !is MonitorV2) {
             return IllegalStateException(

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

@@ -21,6 +21,9 @@ import org.opensearch.alerting.action.ExecuteMonitorResponse
 import org.opensearch.alerting.action.ExecuteWorkflowAction
 import org.opensearch.alerting.action.ExecuteWorkflowRequest
 import org.opensearch.alerting.action.ExecuteWorkflowResponse
+import org.opensearch.alerting.actionv2.ExecuteMonitorV2Action
+import org.opensearch.alerting.actionv2.ExecuteMonitorV2Request
+import org.opensearch.alerting.actionv2.ExecuteMonitorV2Response
 import org.opensearch.alerting.alerts.AlertIndices
 import org.opensearch.alerting.alerts.AlertMover.Companion.moveAlerts
 import org.opensearch.alerting.alertsv2.AlertV2Indices
@@ -31,11 +34,15 @@ import org.opensearch.alerting.core.lock.LockModel
 import org.opensearch.alerting.core.lock.LockService
 import org.opensearch.alerting.model.destination.DestinationContextFactory
 import org.opensearch.alerting.modelv2.MonitorV2
+import org.opensearch.alerting.modelv2.MonitorV2RunResult
+import org.opensearch.alerting.modelv2.PPLSQLMonitor
+import org.opensearch.alerting.modelv2.PPLSQLMonitor.Companion.PPL_SQL_MONITOR_TYPE
 import org.opensearch.alerting.opensearchapi.retry
 import org.opensearch.alerting.opensearchapi.suspendUntil
 import org.opensearch.alerting.remote.monitors.RemoteDocumentLevelMonitorRunner
 import org.opensearch.alerting.remote.monitors.RemoteMonitorRegistry
 import org.opensearch.alerting.script.TriggerExecutionContext
+import org.opensearch.alerting.script.TriggerV2ExecutionContext
 import org.opensearch.alerting.settings.AlertingSettings
 import org.opensearch.alerting.settings.AlertingSettings.Companion.ALERT_BACKOFF_COUNT
 import org.opensearch.alerting.settings.AlertingSettings.Companion.ALERT_BACKOFF_MILLIS
@@ -437,6 +444,44 @@ object MonitorRunnerService : JobRunner, CoroutineScope, AbstractLifecycleCompon
                     }
                 }
             }
+            is MonitorV2 -> {
+                if (job !is PPLSQLMonitor) {
+                    throw IllegalStateException("Unexpected invalid MonitorV2 type: ${job.javaClass.name}")
+                }
+
+                launch {
+                    var monitorLock: LockModel? = null
+                    try {
+                        monitorLock = monitorCtx.client!!.suspendUntil<Client, LockModel?> {
+                            monitorCtx.lockService!!.acquireLock(job, it)
+                        } ?: return@launch
+                        logger.debug("lock ${monitorLock!!.lockId} acquired")
+                        logger.debug(
+                            "PERF_DEBUG: executing $PPL_SQL_MONITOR_TYPE ${job.id} on node " +
+                                monitorCtx.clusterService!!.state().nodes().localNode.id
+                        )
+                        val executeMonitorV2Request = ExecuteMonitorV2Request(
+                            false,
+                            false,
+                            job.id, // only need to pass in MonitorV2 ID
+                            null, // no need to pass in MonitorV2 object itself
+                            TimeValue(periodEnd.toEpochMilli())
+                        )
+                        monitorCtx.client!!.suspendUntil<Client, ExecuteMonitorV2Response> {
+                            monitorCtx.client!!.execute(
+                                ExecuteMonitorV2Action.INSTANCE,
+                                executeMonitorV2Request,
+                                it
+                            )
+                        }
+                    } catch (e: Exception) {
+                        logger.error("MonitorV2 run failed for monitor with id ${job.id}", e)
+                    } finally {
+                        monitorCtx.client!!.suspendUntil { monitorCtx.lockService!!.release(monitorLock, it) }
+                        logger.debug("lock ${monitorLock?.lockId} released")
+                    }
+                }
+            }
             else -> {
                 throw IllegalArgumentException("Invalid job type")
             }
@@ -555,6 +600,44 @@ object MonitorRunnerService : JobRunner, CoroutineScope, AbstractLifecycleCompon
         }
     }
 
+    // after the above JobRunner interface override runJob calls ExecuteMonitorV2 API,
+    // the ExecuteMonitorV2 transport action calls this function to call the PPLSQLMonitorRunner,
+    // where the core PPL/SQL Monitor execution logic resides
+    suspend fun runJobV2(
+        monitorV2: MonitorV2,
+        periodEnd: Instant,
+        dryrun: Boolean,
+        manual: Boolean,
+        transportService: TransportService,
+    ): MonitorV2RunResult<*> {
+        updateAlertingConfigIndexSchema()
+
+        val executionId = "${monitorV2.id}_${LocalDateTime.now(ZoneOffset.UTC)}_${UUID.randomUUID()}"
+        val monitorV2Type = when (monitorV2) {
+            is PPLSQLMonitor -> PPL_SQL_MONITOR_TYPE
+            else -> throw IllegalStateException("Unexpected MonitorV2 type: ${monitorV2.javaClass.name}")
+        }
+
+        logger.info(
+            "Executing scheduled monitor v2 - id: ${monitorV2.id}, type: $monitorV2Type, " +
+                "periodEnd: $periodEnd, dryrun: $dryrun, manual: $manual, executionId: $executionId"
+        )
+
+        // for now, always call PPLSQLMonitorRunner since only PPL Monitors are initially supported
+        // to introduce new MonitorV2 type, create its MonitorRunner, and if/else branch
+        // to the corresponding MonitorRunners based on type. For now, default to PPLSQLMonitorRunner
+        val runResult = PPLSQLMonitorRunner.runMonitorV2(
+            monitorV2,
+            monitorCtx,
+            periodEnd,
+            dryrun,
+            manual,
+            executionId = executionId,
+            transportService = transportService,
+        )
+        return runResult
+    }
+
     // TODO: See if we can move below methods (or few of these) to a common utils
     internal fun getRolesForMonitor(monitor: Monitor): List<String> {
         /*
@@ -599,6 +682,12 @@ object MonitorRunnerService : JobRunner, CoroutineScope, AbstractLifecycleCompon
             .execute()
     }
 
+    internal fun compileTemplateV2(template: Script, ctx: TriggerV2ExecutionContext): String {
+        return monitorCtx.scriptService!!.compile(template, TemplateScript.CONTEXT)
+            .newInstance(template.params + mapOf("ctx" to ctx.asTemplateArg()))
+            .execute()
+    }
+
     private fun updateAlertingConfigIndexSchema() {
         if (!IndexUtils.scheduledJobIndexUpdated && monitorCtx.clusterService != null && monitorCtx.client != null) {
             IndexUtils.updateIndexMapping(

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

@@ -0,0 +1,139 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.alerting
+
+import org.opensearch.OpenSearchSecurityException
+import org.opensearch.alerting.action.GetDestinationsAction
+import org.opensearch.alerting.action.GetDestinationsRequest
+import org.opensearch.alerting.action.GetDestinationsResponse
+import org.opensearch.alerting.model.destination.Destination
+import org.opensearch.alerting.modelv2.MonitorV2
+import org.opensearch.alerting.modelv2.MonitorV2RunResult
+import org.opensearch.alerting.opensearchapi.suspendUntil
+import org.opensearch.alerting.util.destinationmigration.NotificationActionConfigs
+import org.opensearch.alerting.util.destinationmigration.NotificationApiUtils.Companion.getNotificationConfigInfo
+import org.opensearch.alerting.util.destinationmigration.getTitle
+import org.opensearch.alerting.util.destinationmigration.publishLegacyNotification
+import org.opensearch.alerting.util.destinationmigration.sendNotification
+import org.opensearch.alerting.util.isAllowed
+import org.opensearch.alerting.util.isTestAction
+import org.opensearch.commons.alerting.model.Table
+import org.opensearch.commons.alerting.model.action.Action
+import org.opensearch.commons.notifications.model.NotificationConfigInfo
+import org.opensearch.transport.TransportService
+import org.opensearch.transport.client.node.NodeClient
+import java.time.Instant
+
+abstract class MonitorV2Runner {
+    abstract suspend fun runMonitorV2(
+        monitorV2: MonitorV2,
+        monitorCtx: MonitorRunnerExecutionContext, // MonitorV2 reads from same context as Monitor does
+        periodEnd: Instant,
+        dryRun: Boolean,
+        manual: Boolean,
+        executionId: String,
+        transportService: TransportService
+    ): MonitorV2RunResult<*>
+
+    protected suspend fun getConfigAndSendNotification(
+        action: Action,
+        monitorCtx: MonitorRunnerExecutionContext,
+        subject: String?,
+        message: String
+    ): String {
+        val config = getConfigForNotificationAction(action, monitorCtx)
+        if (config.destination == null && config.channel == null) {
+            throw IllegalStateException("Unable to find a Notification Channel or Destination config with id [${action.destinationId}]")
+        }
+
+        // Adding a check on TEST_ACTION Destination type here to avoid supporting it as a LegacyBaseMessage type
+        // just for Alerting integration tests
+        if (config.destination?.isTestAction() == true) {
+            return "test action"
+        }
+
+        if (config.destination?.isAllowed(monitorCtx.allowList) == false) {
+            throw IllegalStateException(
+                "Monitor contains a Destination type that is not allowed: ${config.destination.type}"
+            )
+        }
+
+        var actionResponseContent = ""
+        actionResponseContent = config.channel
+            ?.sendNotification(
+                monitorCtx.client!!,
+                config.channel.getTitle(subject),
+                message
+            ) ?: actionResponseContent
+
+        actionResponseContent = config.destination
+            ?.buildLegacyBaseMessage(subject, message, monitorCtx.destinationContextFactory!!.getDestinationContext(config.destination))
+            ?.publishLegacyNotification(monitorCtx.client!!)
+            ?: actionResponseContent
+
+        return actionResponseContent
+    }
+
+    /**
+     * The "destination" ID referenced in a Monitor Action could either be a Notification config or a Destination config
+     * depending on whether the background migration process has already migrated it from a Destination to a Notification config.
+     *
+     * To cover both of these cases, the Notification config will take precedence and if it is not found, the Destination will be retrieved.
+     */
+    private suspend fun getConfigForNotificationAction(
+        action: Action,
+        monitorCtx: MonitorRunnerExecutionContext
+    ): NotificationActionConfigs {
+        var destination: Destination? = null
+        var notificationPermissionException: Exception? = null
+
+        var channel: NotificationConfigInfo? = null
+        try {
+            channel = getNotificationConfigInfo(monitorCtx.client as NodeClient, action.destinationId)
+        } catch (e: OpenSearchSecurityException) {
+            notificationPermissionException = e
+        }
+
+        // If the channel was not found, try to retrieve the Destination
+        if (channel == null) {
+            destination = try {
+                val table = Table(
+                    "asc",
+                    "destination.name.keyword",
+                    null,
+                    1,
+                    0,
+                    null
+                )
+                val getDestinationsRequest = GetDestinationsRequest(
+                    action.destinationId,
+                    0L,
+                    null,
+                    table,
+                    "ALL"
+                )
+
+                val getDestinationsResponse: GetDestinationsResponse = monitorCtx.client!!.suspendUntil {
+                    monitorCtx.client!!.execute(GetDestinationsAction.INSTANCE, getDestinationsRequest, it)
+                }
+                getDestinationsResponse.destinations.firstOrNull()
+            } catch (e: IllegalStateException) {
+                // Catching the exception thrown when the Destination was not found so the NotificationActionConfigs object can be returned
+                null
+            } catch (e: OpenSearchSecurityException) {
+                if (notificationPermissionException != null)
+                    throw notificationPermissionException
+                else
+                    throw e
+            }
+
+            if (destination == null && notificationPermissionException != null)
+                throw notificationPermissionException
+        }
+
+        return NotificationActionConfigs(destination, channel)
+    }
+}