[Backport 3.3] PPL Alerting: Create and Update Monitor V2 (#1961)

alerting/build.gradle

@@ -151,6 +151,8 @@ dependencies {
     // Needed for integ tests
     zipArchive group: 'org.opensearch.plugin', name:'opensearch-notifications-core', version: "${opensearch_build}"
     zipArchive group: 'org.opensearch.plugin', name:'notifications', version: "${opensearch_build}"
+    zipArchive group: 'org.opensearch.plugin', name:'opensearch-job-scheduler', version: "${opensearch_build}"
+    zipArchive group: 'org.opensearch.plugin', name:'opensearch-sql-plugin', version: "${opensearch_build}"
 
     // Needed for security tests
     if (securityEnabled) {
@@ -168,7 +170,10 @@ dependencies {
     implementation "org.jetbrains.kotlin:kotlin-stdlib-common:${kotlin_version}"
     implementation "org.jetbrains:annotations:13.0"
 
+    // SQL/PPL plugin dependencies are included in alerting-core
     api project(":alerting-core")
+    implementation 'org.json:json:20240303'
+
     implementation "com.github.seancfoley:ipaddress:5.4.1"
     implementation project(path: ":alerting-spi", configuration: 'shadow')
 
@@ -246,6 +251,28 @@ testClusters.integTest {
         }
     }))
 
+    plugin(provider({
+        new RegularFile() {
+            @Override
+            File getAsFile() {
+                return configurations.zipArchive.asFileTree.matching {
+                    include '**/opensearch-job-scheduler*'
+                }.singleFile
+            }
+        }
+    }))
+
+    plugin(provider({
+        new RegularFile() {
+            @Override
+            File getAsFile() {
+                return configurations.zipArchive.asFileTree.matching {
+                    include '**/opensearch-sql-plugin*'
+                }.singleFile
+            }
+        }
+    }))
+
     if (securityEnabled) {
         plugin(provider({
             new RegularFile() {

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

@@ -14,6 +14,7 @@ import org.opensearch.alerting.action.GetEmailGroupAction
 import org.opensearch.alerting.action.GetRemoteIndexesAction
 import org.opensearch.alerting.action.SearchEmailAccountAction
 import org.opensearch.alerting.action.SearchEmailGroupAction
+import org.opensearch.alerting.actionv2.IndexMonitorV2Action
 import org.opensearch.alerting.alerts.AlertIndices
 import org.opensearch.alerting.alerts.AlertIndices.Companion.ALL_ALERT_INDEX_PATTERN
 import org.opensearch.alerting.comments.CommentsIndices
@@ -27,6 +28,7 @@ import org.opensearch.alerting.core.resthandler.RestScheduledJobStatsHandler
 import org.opensearch.alerting.core.schedule.JobScheduler
 import org.opensearch.alerting.core.settings.LegacyOpenDistroScheduledJobSettings
 import org.opensearch.alerting.core.settings.ScheduledJobSettings
+import org.opensearch.alerting.modelv2.MonitorV2
 import org.opensearch.alerting.remote.monitors.RemoteMonitorRegistry
 import org.opensearch.alerting.resthandler.RestAcknowledgeAlertAction
 import org.opensearch.alerting.resthandler.RestAcknowledgeChainedAlertAction
@@ -51,6 +53,7 @@ import org.opensearch.alerting.resthandler.RestSearchAlertingCommentAction
 import org.opensearch.alerting.resthandler.RestSearchEmailAccountAction
 import org.opensearch.alerting.resthandler.RestSearchEmailGroupAction
 import org.opensearch.alerting.resthandler.RestSearchMonitorAction
+import org.opensearch.alerting.resthandlerv2.RestIndexMonitorV2Action
 import org.opensearch.alerting.script.TriggerScript
 import org.opensearch.alerting.service.DeleteMonitorService
 import org.opensearch.alerting.settings.AlertingSettings
@@ -83,6 +86,7 @@ import org.opensearch.alerting.transport.TransportSearchAlertingCommentAction
 import org.opensearch.alerting.transport.TransportSearchEmailAccountAction
 import org.opensearch.alerting.transport.TransportSearchEmailGroupAction
 import org.opensearch.alerting.transport.TransportSearchMonitorAction
+import org.opensearch.alerting.transportv2.TransportIndexMonitorV2Action
 import org.opensearch.alerting.util.DocLevelMonitorQueries
 import org.opensearch.alerting.util.destinationmigration.DestinationMigrationCoordinator
 import org.opensearch.cluster.metadata.IndexNameExpressionResolver
@@ -157,6 +161,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
         @JvmField val OPEN_SEARCH_DASHBOARDS_USER_AGENT = "OpenSearch-Dashboards"
         @JvmField val UI_METADATA_EXCLUDE = arrayOf("monitor.${Monitor.UI_METADATA_FIELD}")
         @JvmField val MONITOR_BASE_URI = "/_plugins/_alerting/monitors"
+        @JvmField val MONITOR_V2_BASE_URI = "/_plugins/_alerting/v2/monitors"
         @JvmField val WORKFLOW_BASE_URI = "/_plugins/_alerting/workflows"
         @JvmField val REMOTE_BASE_URI = "/_plugins/_alerting/remote"
         @JvmField val DESTINATION_BASE_URI = "/_plugins/_alerting/destinations"
@@ -169,7 +174,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
         @JvmField val FINDING_BASE_URI = "/_plugins/_alerting/findings"
         @JvmField val COMMENTS_BASE_URI = "/_plugins/_alerting/comments"
 
-        @JvmField val ALERTING_JOB_TYPES = listOf("monitor", "workflow")
+        @JvmField val ALERTING_JOB_TYPES = listOf("monitor", "workflow", "monitor_v2")
     }
 
     lateinit var runner: MonitorRunnerService
@@ -194,6 +199,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
         nodesInCluster: Supplier<DiscoveryNodes>
     ): List<RestHandler> {
         return listOf(
+            // Alerting V1
             RestGetMonitorAction(),
             RestDeleteMonitorAction(),
             RestIndexMonitorAction(),
@@ -218,11 +224,15 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
             RestIndexAlertingCommentAction(),
             RestSearchAlertingCommentAction(),
             RestDeleteAlertingCommentAction(),
+
+            // Alerting V2
+            RestIndexMonitorV2Action(),
         )
     }
 
     override fun getActions(): List<ActionPlugin.ActionHandler<out ActionRequest, out ActionResponse>> {
         return listOf(
+            // Alerting V1
             ActionPlugin.ActionHandler(ScheduledJobsStatsAction.INSTANCE, ScheduledJobsStatsTransportAction::class.java),
             ActionPlugin.ActionHandler(AlertingActions.INDEX_MONITOR_ACTION_TYPE, TransportIndexMonitorAction::class.java),
             ActionPlugin.ActionHandler(AlertingActions.GET_MONITOR_ACTION_TYPE, TransportGetMonitorAction::class.java),
@@ -249,13 +259,17 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
             ActionPlugin.ActionHandler(AlertingActions.DELETE_COMMENT_ACTION_TYPE, TransportDeleteAlertingCommentAction::class.java),
             ActionPlugin.ActionHandler(ExecuteWorkflowAction.INSTANCE, TransportExecuteWorkflowAction::class.java),
             ActionPlugin.ActionHandler(GetRemoteIndexesAction.INSTANCE, TransportGetRemoteIndexesAction::class.java),
-            ActionPlugin.ActionHandler(DocLevelMonitorFanOutAction.INSTANCE, TransportDocLevelMonitorFanOutAction::class.java)
+            ActionPlugin.ActionHandler(DocLevelMonitorFanOutAction.INSTANCE, TransportDocLevelMonitorFanOutAction::class.java),
+
+            // Alerting V2
+            ActionPlugin.ActionHandler(IndexMonitorV2Action.INSTANCE, TransportIndexMonitorV2Action::class.java),
         )
     }
 
     override fun getNamedXContent(): List<NamedXContentRegistry.Entry> {
         return listOf(
             Monitor.XCONTENT_REGISTRY,
+            MonitorV2.XCONTENT_REGISTRY,
             SearchInput.XCONTENT_REGISTRY,
             DocLevelMonitorInput.XCONTENT_REGISTRY,
             QueryLevelTrigger.XCONTENT_REGISTRY,
@@ -431,7 +445,22 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
             AlertingSettings.COMMENTS_HISTORY_RETENTION_PERIOD,
             AlertingSettings.COMMENTS_MAX_CONTENT_SIZE,
             AlertingSettings.MAX_COMMENTS_PER_ALERT,
-            AlertingSettings.MAX_COMMENTS_PER_NOTIFICATION
+            AlertingSettings.MAX_COMMENTS_PER_NOTIFICATION,
+            AlertingSettings.ALERT_V2_HISTORY_ENABLED,
+            AlertingSettings.ALERT_V2_HISTORY_ROLLOVER_PERIOD,
+            AlertingSettings.ALERT_V2_HISTORY_INDEX_MAX_AGE,
+            AlertingSettings.ALERT_V2_HISTORY_MAX_DOCS,
+            AlertingSettings.ALERT_V2_HISTORY_RETENTION_PERIOD,
+            AlertingSettings.ALERTING_V2_MAX_MONITORS,
+            AlertingSettings.ALERTING_V2_MAX_THROTTLE_DURATION,
+            AlertingSettings.ALERTING_V2_MAX_EXPIRE_DURATION,
+            AlertingSettings.ALERTING_V2_MAX_LOOK_BACK_WINDOW,
+            AlertingSettings.ALERTING_V2_MAX_QUERY_LENGTH,
+            AlertingSettings.ALERTING_V2_QUERY_RESULTS_MAX_DATAROWS,
+            AlertingSettings.ALERT_V2_QUERY_RESULTS_MAX_SIZE,
+            AlertingSettings.ALERT_V2_PER_RESULT_TRIGGER_MAX_ALERTS,
+            AlertingSettings.NOTIFICATION_SUBJECT_SOURCE_MAX_LENGTH,
+            AlertingSettings.NOTIFICATION_MESSAGE_SOURCE_MAX_LENGTH
         )
     }
 
@@ -449,7 +478,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
         return listOf(
             SystemIndexDescriptor(ALL_ALERT_INDEX_PATTERN, "Alerting Plugin system index pattern"),
             SystemIndexDescriptor(SCHEDULED_JOBS_INDEX, "Alerting Plugin Configuration index"),
-            SystemIndexDescriptor(ALL_COMMENTS_INDEX_PATTERN, "Alerting Comments system index pattern")
+            SystemIndexDescriptor(ALL_COMMENTS_INDEX_PATTERN, "Alerting Comments system index pattern"),
         )
     }
 

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

@@ -0,0 +1,87 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.alerting
+
+import org.apache.lucene.search.TotalHits
+import org.apache.lucene.search.TotalHits.Relation
+import org.opensearch.action.search.SearchResponse
+import org.opensearch.action.search.ShardSearchFailure
+import org.opensearch.alerting.modelv2.MonitorV2
+import org.opensearch.commons.alerting.model.Monitor
+import org.opensearch.commons.alerting.model.ScheduledJob
+import org.opensearch.commons.alerting.model.Workflow
+import org.opensearch.index.IndexNotFoundException
+import org.opensearch.search.SearchHits
+import org.opensearch.search.aggregations.InternalAggregations
+import org.opensearch.search.internal.InternalSearchResponse
+import org.opensearch.search.profile.SearchProfileShardResults
+import org.opensearch.search.suggest.Suggest
+import org.opensearch.transport.RemoteTransportException
+import java.util.Collections
+
+object AlertingV2Utils {
+    // Validates that the given scheduled job is a Monitor
+    // returns the exception to pass into actionListener.onFailure if not.
+    fun validateMonitorV1(scheduledJob: ScheduledJob): Exception? {
+        if (scheduledJob is MonitorV2) {
+            return IllegalStateException("The ID given corresponds to a V2 Monitor, but a V1 Monitor was expected")
+        } else if (scheduledJob !is Monitor && scheduledJob !is Workflow) {
+            return IllegalStateException("The ID given corresponds to a scheduled job of unknown type: ${scheduledJob.javaClass.name}")
+        }
+        return null
+    }
+
+    // Validates that the given scheduled job is a MonitorV2
+    // returns the exception to pass into actionListener.onFailure if not.
+    fun validateMonitorV2(scheduledJob: ScheduledJob): Exception? {
+        if (scheduledJob is Monitor || scheduledJob is Workflow) {
+            return IllegalStateException("The ID given corresponds to a V1 Monitor, but a V2 Monitor was expected")
+        } else if (scheduledJob !is MonitorV2) {
+            return IllegalStateException("The ID given corresponds to a scheduled job of unknown type: ${scheduledJob.javaClass.name}")
+        }
+        return null
+    }
+
+    // Checks if the exception is caused by an IndexNotFoundException (directly or nested).
+    // Used in Get and Search monitor functionalities to determine whether a "no results"
+    // response should be returned
+    fun isIndexNotFoundException(e: Exception): Boolean {
+        if (e is IndexNotFoundException) {
+            return true
+        }
+        if (e is RemoteTransportException) {
+            val cause = e.cause
+            if (cause is IndexNotFoundException) {
+                return true
+            }
+        }
+        return false
+    }
+
+    // Used in Get and Search monitor functionalities to return a "no results" response
+    fun getEmptySearchResponse(): SearchResponse {
+        val internalSearchResponse = InternalSearchResponse(
+            SearchHits(emptyArray(), TotalHits(0L, Relation.EQUAL_TO), 0.0f),
+            InternalAggregations.from(Collections.emptyList()),
+            Suggest(Collections.emptyList()),
+            SearchProfileShardResults(Collections.emptyMap()),
+            false,
+            false,
+            0
+        )
+
+        return SearchResponse(
+            internalSearchResponse,
+            "",
+            0,
+            0,
+            0,
+            0,
+            ShardSearchFailure.EMPTY_ARRAY,
+            SearchResponse.Clusters.EMPTY
+        )
+    }
+}