CRUD APIs integration Tests and validation"conflict resolved"

alerting/src/main/kotlin/org/opensearch/alerting/model/Monitor.kt

@@ -67,9 +67,11 @@ data class Monitor(
             // Verify Trigger type based on Monitor type
             when (monitorType) {
                 MonitorType.QUERY_LEVEL_MONITOR ->
-                    require(trigger is QueryLevelTrigger) { "Incompatible trigger [$trigger.id] for monitor type [$monitorType]" }
+                    require(trigger is QueryLevelTrigger) { "Incompatible trigger [${trigger.id}] for monitor type [$monitorType]" }
                 MonitorType.BUCKET_LEVEL_MONITOR ->
-                    require(trigger is BucketLevelTrigger) { "Incompatible trigger [$trigger.id] for monitor type [$monitorType]" }
+                    require(trigger is BucketLevelTrigger) { "Incompatible trigger [${trigger.id}] for monitor type [$monitorType]" }
+                MonitorType.DOC_LEVEL_MONITOR ->
+                    require(trigger is DocumentLevelTrigger) { "Incompatible trigger [${trigger.id}] for monitor type [$monitorType]" }
             }
         }
         if (enabled) {
@@ -185,6 +187,7 @@ data class Monitor(
         out.writeVInt(triggers.size)
         triggers.forEach {
             if (it is QueryLevelTrigger) out.writeEnum(Trigger.Type.QUERY_LEVEL_TRIGGER)
+            else if (it is DocumentLevelTrigger) out.writeEnum(Trigger.Type.DOCUMENT_LEVEL_TRIGGER)
             else out.writeEnum(Trigger.Type.BUCKET_LEVEL_TRIGGER)
             it.writeTo(out)
         }

alerting/src/main/kotlin/org/opensearch/alerting/resthandler/RestIndexMonitorAction.kt

@@ -10,7 +10,10 @@ import org.opensearch.alerting.AlertingPlugin
 import org.opensearch.alerting.action.IndexMonitorAction
 import org.opensearch.alerting.action.IndexMonitorRequest
 import org.opensearch.alerting.action.IndexMonitorResponse
+import org.opensearch.alerting.model.BucketLevelTrigger
+import org.opensearch.alerting.model.DocumentLevelTrigger
 import org.opensearch.alerting.model.Monitor
+import org.opensearch.alerting.model.QueryLevelTrigger
 import org.opensearch.alerting.util.IF_PRIMARY_TERM
 import org.opensearch.alerting.util.IF_SEQ_NO
 import org.opensearch.alerting.util.REFRESH
@@ -79,6 +82,31 @@ class RestIndexMonitorAction : BaseRestHandler() {
         val xcp = request.contentParser()
         ensureExpectedToken(Token.START_OBJECT, xcp.nextToken(), xcp)
         val monitor = Monitor.parse(xcp, id).copy(lastUpdateTime = Instant.now())
+        val monitorType = monitor.monitorType
+        val triggers = monitor.triggers
+        when (monitorType) {
+            Monitor.MonitorType.QUERY_LEVEL_MONITOR -> {
+                triggers.forEach {
+                    if (it !is QueryLevelTrigger) {
+                        throw IllegalArgumentException("Illegal trigger type, ${it.javaClass.name}, for query level monitor")
+                    }
+                }
+            }
+            Monitor.MonitorType.BUCKET_LEVEL_MONITOR -> {
+                triggers.forEach {
+                    if (it !is BucketLevelTrigger) {
+                        throw IllegalArgumentException("Illegal trigger type, ${it.javaClass.name}, for bucket level monitor")
+                    }
+                }
+            }
+            Monitor.MonitorType.DOC_LEVEL_MONITOR -> {
+                triggers.forEach {
+                    if (it !is DocumentLevelTrigger) {
+                        throw IllegalArgumentException("Illegal trigger type, ${it.javaClass.name}, for document level monitor")
+                    }
+                }
+            }
+        }
         val seqNo = request.paramAsLong(IF_SEQ_NO, SequenceNumbers.UNASSIGNED_SEQ_NO)
         val primaryTerm = request.paramAsLong(IF_PRIMARY_TERM, SequenceNumbers.UNASSIGNED_PRIMARY_TERM)
         val refreshPolicy = if (request.hasParam(REFRESH)) {

alerting/src/test/kotlin/org/opensearch/alerting/AlertingRestTestCase.kt

@@ -23,6 +23,7 @@ import org.opensearch.alerting.core.settings.ScheduledJobSettings
 import org.opensearch.alerting.elasticapi.string
 import org.opensearch.alerting.model.Alert
 import org.opensearch.alerting.model.BucketLevelTrigger
+import org.opensearch.alerting.model.DocumentLevelTrigger
 import org.opensearch.alerting.model.Monitor
 import org.opensearch.alerting.model.QueryLevelTrigger
 import org.opensearch.alerting.model.destination.Destination
@@ -75,7 +76,8 @@ abstract class AlertingRestTestCase : ODFERestTestCase() {
                 Monitor.XCONTENT_REGISTRY,
                 SearchInput.XCONTENT_REGISTRY,
                 QueryLevelTrigger.XCONTENT_REGISTRY,
-                BucketLevelTrigger.XCONTENT_REGISTRY
+                BucketLevelTrigger.XCONTENT_REGISTRY,
+                DocumentLevelTrigger.XCONTENT_REGISTRY
             ) + SearchModule(Settings.EMPTY, false, emptyList()).namedXContents
         )
     }
@@ -400,6 +402,15 @@ abstract class AlertingRestTestCase : ODFERestTestCase() {
         return getMonitor(monitorId = monitorId)
     }
 
+    protected fun createRandomDocumentMonitor(refresh: Boolean = false, withMetadata: Boolean = false): Monitor {
+        val monitor = randomDocumentLevelMonitor(withMetadata = withMetadata)
+        val monitorId = createMonitor(monitor, refresh).id
+        if (withMetadata) {
+            return getMonitor(monitorId = monitorId, header = BasicHeader(HttpHeaders.USER_AGENT, "OpenSearch-Dashboards"))
+        }
+        return getMonitor(monitorId = monitorId)
+    }
+
     @Suppress("UNCHECKED_CAST")
     protected fun updateMonitor(monitor: Monitor, refresh: Boolean = false): Monitor {
         val response = client().makeRequest(

alerting/src/test/kotlin/org/opensearch/alerting/TestHelpers.kt

@@ -572,7 +572,10 @@ fun parser(xc: String): XContentParser {
 fun xContentRegistry(): NamedXContentRegistry {
     return NamedXContentRegistry(
         listOf(
-            SearchInput.XCONTENT_REGISTRY, QueryLevelTrigger.XCONTENT_REGISTRY, BucketLevelTrigger.XCONTENT_REGISTRY
+            SearchInput.XCONTENT_REGISTRY,
+            QueryLevelTrigger.XCONTENT_REGISTRY,
+            BucketLevelTrigger.XCONTENT_REGISTRY,
+            DocumentLevelTrigger.XCONTENT_REGISTRY
         ) + SearchModule(Settings.EMPTY, false, emptyList()).namedXContents
     )
 }
@@ -585,3 +588,21 @@ fun assertUserNull(map: Map<String, Any?>) {
 fun assertUserNull(monitor: Monitor) {
     assertNull("User is not null", monitor.user)
 }
+
+fun randomDocumentLevelMonitor(
+    name: String = OpenSearchRestTestCase.randomAlphaOfLength(10),
+    user: User = randomUser(),
+    inputs: List<Input> = listOf(SearchInput(emptyList(), SearchSourceBuilder().query(QueryBuilders.matchAllQuery()))),
+    schedule: Schedule = IntervalSchedule(interval = 5, unit = ChronoUnit.MINUTES),
+    enabled: Boolean = randomBoolean(),
+    triggers: List<Trigger> = (1..randomInt(10)).map { randomDocLevelTrigger() },
+    enabledTime: Instant? = if (enabled) Instant.now().truncatedTo(ChronoUnit.MILLIS) else null,
+    lastUpdateTime: Instant = Instant.now().truncatedTo(ChronoUnit.MILLIS),
+    withMetadata: Boolean = false
+): Monitor {
+    return Monitor(
+        name = name, monitorType = Monitor.MonitorType.DOC_LEVEL_MONITOR, enabled = enabled, inputs = inputs,
+        schedule = schedule, triggers = triggers, enabledTime = enabledTime, lastUpdateTime = lastUpdateTime, user = user,
+        lastRunContext = mapOf(), uiMetadata = if (withMetadata) mapOf("foo" to "bar") else mapOf()
+    )
+}

alerting/src/test/kotlin/org/opensearch/alerting/resthandler/MonitorRestApiIT.kt

@@ -21,6 +21,7 @@ import org.opensearch.alerting.core.model.SearchInput
 import org.opensearch.alerting.core.settings.ScheduledJobSettings
 import org.opensearch.alerting.makeRequest
 import org.opensearch.alerting.model.Alert
+import org.opensearch.alerting.model.DocumentLevelTrigger
 import org.opensearch.alerting.model.Monitor
 import org.opensearch.alerting.model.QueryLevelTrigger
 import org.opensearch.alerting.model.destination.Chime
@@ -30,6 +31,8 @@ import org.opensearch.alerting.randomAction
 import org.opensearch.alerting.randomAlert
 import org.opensearch.alerting.randomAnomalyDetector
 import org.opensearch.alerting.randomAnomalyDetectorWithUser
+import org.opensearch.alerting.randomBucketLevelTrigger
+import org.opensearch.alerting.randomDocumentLevelMonitor
 import org.opensearch.alerting.randomQueryLevelMonitor
 import org.opensearch.alerting.randomQueryLevelTrigger
 import org.opensearch.alerting.randomThrottle
@@ -1102,4 +1105,95 @@ class MonitorRestApiIT : AlertingRestTestCase() {
 
         return false
     }
+
+    @Throws(Exception::class)
+    fun `test creating a document monitor`() {
+        val monitor = randomDocumentLevelMonitor()
+
+        val createResponse = client().makeRequest("POST", ALERTING_BASE_URI, emptyMap(), monitor.toHttpEntity())
+
+        assertEquals("Create monitor failed", RestStatus.CREATED, createResponse.restStatus())
+        val responseBody = createResponse.asMap()
+        val createdId = responseBody["_id"] as String
+        val createdVersion = responseBody["_version"] as Int
+        assertNotEquals("response is missing Id", Monitor.NO_ID, createdId)
+        assertTrue("incorrect version", createdVersion > 0)
+        val actualLocation = createResponse.getHeader("Location")
+        assertEquals("Incorrect Location header", "$ALERTING_BASE_URI/$createdId", actualLocation)
+    }
+
+    @Throws(Exception::class)
+    fun `test getting a document level monitor`() {
+        val monitor = createRandomDocumentMonitor()
+
+        val storedMonitor = getMonitor(monitor.id)
+
+        assertEquals("Indexed and retrieved monitor differ", monitor, storedMonitor)
+    }
+
+    @Throws(Exception::class)
+    fun `test updating conditions for a doc-level monitor`() {
+        val monitor = createRandomDocumentMonitor()
+        val updatedTriggers = listOf(
+            DocumentLevelTrigger(
+                name = "foo",
+                severity = "1",
+                condition = Script("return true"),
+                actions = emptyList()
+            )
+        )
+        val updateResponse = client().makeRequest(
+            "PUT", monitor.relativeUrl(),
+            emptyMap(), monitor.copy(triggers = updatedTriggers).toHttpEntity()
+        )
+
+        assertEquals("Update monitor failed", RestStatus.OK, updateResponse.restStatus())
+        val responseBody = updateResponse.asMap()
+        assertEquals("Updated monitor id doesn't match", monitor.id, responseBody["_id"] as String)
+        assertEquals("Version not incremented", (monitor.version + 1).toInt(), responseBody["_version"] as Int)
+
+        val updatedMonitor = getMonitor(monitor.id)
+        assertEquals("Monitor trigger not updated", updatedTriggers, updatedMonitor.triggers)
+    }
+
+    @Throws(Exception::class)
+    fun `test deleting a document level monitor`() {
+        val monitor = createRandomDocumentMonitor()
+
+        val deleteResponse = client().makeRequest("DELETE", monitor.relativeUrl())
+        assertEquals("Delete failed", RestStatus.OK, deleteResponse.restStatus())
+
+        val getResponse = client().makeRequest("HEAD", monitor.relativeUrl())
+        assertEquals("Deleted monitor still exists", RestStatus.NOT_FOUND, getResponse.restStatus())
+    }
+
+    fun `test creating a document monitor with error trigger`() {
+        val trigger = randomQueryLevelTrigger()
+        try {
+            val monitor = randomDocumentLevelMonitor(triggers = listOf(trigger))
+            client().makeRequest("POST", ALERTING_BASE_URI, emptyMap(), monitor.toHttpEntity())
+            fail("Monitor with illegal trigger should be rejected.")
+        } catch (e: IllegalArgumentException) {
+            assertEquals(
+                "a document monitor with error trigger",
+                "Incompatible trigger [${trigger.id}] for monitor type [${Monitor.MonitorType.DOC_LEVEL_MONITOR}]",
+                e.message
+            )
+        }
+    }
+
+    fun `test creating a query monitor with error trigger`() {
+        val trigger = randomBucketLevelTrigger()
+        try {
+            val monitor = randomQueryLevelMonitor(triggers = listOf(trigger))
+            client().makeRequest("POST", ALERTING_BASE_URI, emptyMap(), monitor.toHttpEntity())
+            fail("Monitor with illegal trigger should be rejected.")
+        } catch (e: IllegalArgumentException) {
+            assertEquals(
+                "a query monitor with error trigger",
+                "Incompatible trigger [${trigger.id}] for monitor type [${Monitor.MonitorType.QUERY_LEVEL_MONITOR}]",
+                e.message
+            )
+        }
+    }
 }