Move get monitor and search monitor action / request / responses to common-utils

src/main/kotlin/org/opensearch/commons/alerting/AlertingPluginInterface.kt

@@ -17,6 +17,8 @@
 import org.opensearch.commons.alerting.action.GetAlertsResponse
 import org.opensearch.commons.alerting.action.GetFindingsRequest
 import org.opensearch.commons.alerting.action.GetFindingsResponse
+import org.opensearch.commons.alerting.action.GetMonitorRequest
+import org.opensearch.commons.alerting.action.GetMonitorResponse
 import org.opensearch.commons.alerting.action.GetWorkflowAlertsRequest
 import org.opensearch.commons.alerting.action.GetWorkflowAlertsResponse
 import org.opensearch.commons.alerting.action.GetWorkflowRequest
@@ -288,6 +290,30 @@
         )
     }
 
+    /**
+     * Get Monitor interface.
+     * @param client Node client for making transport action
+     * @param request The request object
+     * @param listener The listener for getting response
+     */
+    fun getMonitor(
+        client: NodeClient,
+        request: GetMonitorRequest,
+        listener: ActionListener<GetMonitorResponse>
+    ) {
+        client.execute(
+            AlertingActions.GET_MONITOR_ACTION_TYPE,
+            request,
+            wrapActionListener(listener) { response ->
+                recreateObject(response) {
+                    GetMonitorResponse(
+                        it
+                    )
+                }
+            }
+        )
+    }
+
     @Suppress("UNCHECKED_CAST")
     private fun <Response : BaseResponse> wrapActionListener(
         listener: ActionListener<Response>,

src/main/kotlin/org/opensearch/commons/alerting/action/AlertingActions.kt

@@ -18,6 +18,7 @@ object AlertingActions {
     const val ACKNOWLEDGE_ALERTS_ACTION_NAME = "cluster:admin/opendistro/alerting/alerts/ack"
     const val ACKNOWLEDGE_CHAINED_ALERTS_ACTION_NAME = "cluster:admin/opendistro/alerting/chained_alerts/ack"
     const val SUBSCRIBE_FINDINGS_ACTION_NAME = "cluster:admin/opensearch/alerting/findings/subscribe"
+    const val GET_MONITOR_ACTION_NAME = "cluster:admin/opendistro/alerting/monitor/get"
 
     @JvmField
     val INDEX_MONITOR_ACTION_TYPE =
@@ -60,4 +61,8 @@ object AlertingActions {
     @JvmField
     val ACKNOWLEDGE_CHAINED_ALERTS_ACTION_TYPE =
         ActionType(ACKNOWLEDGE_CHAINED_ALERTS_ACTION_NAME, ::AcknowledgeAlertResponse)
+
+    @JvmField
+    val GET_MONITOR_ACTION_TYPE =
+        ActionType(GET_MONITOR_ACTION_NAME, ::GetMonitorResponse)
 }

src/main/kotlin/org/opensearch/commons/alerting/action/GetMonitorRequest.kt

@@ -0,0 +1,56 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.commons.alerting.action
+
+import org.opensearch.action.ActionRequest
+import org.opensearch.action.ActionRequestValidationException
+import org.opensearch.core.common.io.stream.StreamInput
+import org.opensearch.core.common.io.stream.StreamOutput
+import org.opensearch.rest.RestRequest
+import org.opensearch.search.fetch.subphase.FetchSourceContext
+import java.io.IOException
+
+class GetMonitorRequest : ActionRequest {
+    val monitorId: String
+    val version: Long
+    val method: RestRequest.Method
+    val srcContext: FetchSourceContext?
+
+    constructor(
+        monitorId: String,
+        version: Long,
+        method: RestRequest.Method,
+        srcContext: FetchSourceContext?
+    ) : super() {
+        this.monitorId = monitorId
+        this.version = version
+        this.method = method
+        this.srcContext = srcContext
+    }
+
+    @Throws(IOException::class)
+    constructor(sin: StreamInput) : this(
+        sin.readString(), // monitorId
+        sin.readLong(), // version
+        sin.readEnum(RestRequest.Method::class.java), // method
+        if (sin.readBoolean()) {
+            FetchSourceContext(sin) // srcContext
+        } else null
+    )
+
+    override fun validate(): ActionRequestValidationException? {
+        return null
+    }
+
+    @Throws(IOException::class)
+    override fun writeTo(out: StreamOutput) {
+        out.writeString(monitorId)
+        out.writeLong(version)
+        out.writeEnum(method)
+        out.writeBoolean(srcContext != null)
+        srcContext?.writeTo(out)
+    }
+}

src/main/kotlin/org/opensearch/commons/alerting/action/GetMonitorResponse.kt

@@ -0,0 +1,126 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.commons.alerting.action
+
+import org.opensearch.commons.alerting.model.Monitor
+import org.opensearch.commons.alerting.util.IndexUtils.Companion._ID
+import org.opensearch.commons.alerting.util.IndexUtils.Companion._PRIMARY_TERM
+import org.opensearch.commons.alerting.util.IndexUtils.Companion._SEQ_NO
+import org.opensearch.commons.alerting.util.IndexUtils.Companion._VERSION
+import org.opensearch.commons.notifications.action.BaseResponse
+import org.opensearch.core.common.io.stream.StreamInput
+import org.opensearch.core.common.io.stream.StreamOutput
+import org.opensearch.core.xcontent.ToXContent
+import org.opensearch.core.xcontent.ToXContentFragment
+import org.opensearch.core.xcontent.XContentBuilder
+import java.io.IOException
+
+class GetMonitorResponse : BaseResponse {
+    var id: String
+    var version: Long
+    var seqNo: Long
+    var primaryTerm: Long
+    var monitor: Monitor?
+    var associatedWorkflows: List<AssociatedWorkflow>?
+
+    constructor(
+        id: String,
+        version: Long,
+        seqNo: Long,
+        primaryTerm: Long,
+        monitor: Monitor?,
+        associatedCompositeMonitors: List<AssociatedWorkflow>?,
+    ) : super() {
+        this.id = id
+        this.version = version
+        this.seqNo = seqNo
+        this.primaryTerm = primaryTerm
+        this.monitor = monitor
+        this.associatedWorkflows = associatedCompositeMonitors ?: emptyList()
+    }
+
+    @Throws(IOException::class)
+    constructor(sin: StreamInput) : this(
+        id = sin.readString(), // id
+        version = sin.readLong(), // version
+        seqNo = sin.readLong(), // seqNo
+        primaryTerm = sin.readLong(), // primaryTerm
+        monitor = if (sin.readBoolean()) {
+            Monitor.readFrom(sin) // monitor
+        } else null,
+        associatedCompositeMonitors = sin.readList((AssociatedWorkflow)::readFrom),
+    )
+
+    @Throws(IOException::class)
+    override fun writeTo(out: StreamOutput) {
+        out.writeString(id)
+        out.writeLong(version)
+        out.writeLong(seqNo)
+        out.writeLong(primaryTerm)
+        if (monitor != null) {
+            out.writeBoolean(true)
+            monitor?.writeTo(out)
+        } else {
+            out.writeBoolean(false)
+        }
+        associatedWorkflows?.forEach {
+            it.writeTo(out)
+        }
+    }
+
+    @Throws(IOException::class)
+    override fun toXContent(builder: XContentBuilder, params: ToXContent.Params): XContentBuilder {
+        builder.startObject()
+            .field(_ID, id)
+            .field(_VERSION, version)
+            .field(_SEQ_NO, seqNo)
+            .field(_PRIMARY_TERM, primaryTerm)
+        if (monitor != null) {
+            builder.field("monitor", monitor)
+        }
+        if (associatedWorkflows != null) {
+            builder.field("associated_workflows", associatedWorkflows!!.toTypedArray())
+        }
+        return builder.endObject()
+    }
+
+    class AssociatedWorkflow : ToXContentFragment {
+        val id: String
+        val name: String
+
+        constructor(id: String, name: String) {
+            this.id = id
+            this.name = name
+        }
+
+        override fun toXContent(builder: XContentBuilder, params: ToXContent.Params?): XContentBuilder {
+            builder.startObject()
+                .field("id", id)
+                .field("name", name)
+                .endObject()
+            return builder
+        }
+
+        fun writeTo(out: StreamOutput) {
+            out.writeString(id)
+            out.writeString(name)
+        }
+
+        @Throws(IOException::class)
+        constructor(sin: StreamInput) : this(
+            sin.readString(),
+            sin.readString()
+        )
+
+        companion object {
+            @JvmStatic
+            @Throws(IOException::class)
+            fun readFrom(sin: StreamInput): AssociatedWorkflow {
+                return AssociatedWorkflow(sin)
+            }
+        }
+    }
+}

src/test/kotlin/org/opensearch/commons/alerting/AlertingPluginInterfaceTests.kt

@@ -23,6 +23,8 @@ import org.opensearch.commons.alerting.action.GetAlertsRequest
 import org.opensearch.commons.alerting.action.GetAlertsResponse
 import org.opensearch.commons.alerting.action.GetFindingsRequest
 import org.opensearch.commons.alerting.action.GetFindingsResponse
+import org.opensearch.commons.alerting.action.GetMonitorRequest
+import org.opensearch.commons.alerting.action.GetMonitorResponse
 import org.opensearch.commons.alerting.action.GetWorkflowAlertsRequest
 import org.opensearch.commons.alerting.action.GetWorkflowAlertsResponse
 import org.opensearch.commons.alerting.action.GetWorkflowRequest
@@ -269,4 +271,18 @@ internal class AlertingPluginInterfaceTests {
         AlertingPluginInterface.acknowledgeChainedAlerts(client, request, listener)
         Mockito.verify(listener, Mockito.times(1)).onResponse(ArgumentMatchers.eq(response))
     }
+
+    @Test
+    fun getMonitor() {
+        val request = mock(GetMonitorRequest::class.java)
+        val response = GetMonitorResponse("test-id", 1, 1, 1, null, null)
+        val listener: ActionListener<GetMonitorResponse> =
+            mock(ActionListener::class.java) as ActionListener<GetMonitorResponse>
+        Mockito.doAnswer {
+            (it.getArgument(2) as ActionListener<GetMonitorResponse>)
+                .onResponse(response)
+        }.whenever(client).execute(Mockito.any(ActionType::class.java), Mockito.any(), Mockito.any())
+        AlertingPluginInterface.getMonitor(client, request, listener)
+        Mockito.verify(listener, Mockito.times(1)).onResponse(ArgumentMatchers.eq(response))
+    }
 }

src/test/kotlin/org/opensearch/commons/alerting/action/GetMonitorRequestTests.kt

@@ -0,0 +1,60 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.commons.alerting.action
+
+import org.opensearch.common.io.stream.BytesStreamOutput
+import org.opensearch.core.common.io.stream.StreamInput
+import org.opensearch.rest.RestRequest
+import org.opensearch.search.fetch.subphase.FetchSourceContext
+import org.opensearch.test.OpenSearchTestCase
+
+class GetMonitorRequestTests : OpenSearchTestCase() {
+
+    fun `test get monitor request`() {
+
+        val req = GetMonitorRequest("1234", 1L, RestRequest.Method.GET, FetchSourceContext.FETCH_SOURCE)
+        assertNotNull(req)
+
+        val out = BytesStreamOutput()
+        req.writeTo(out)
+        val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes)
+        val newReq = GetMonitorRequest(sin)
+        assertEquals("1234", newReq.monitorId)
+        assertEquals(1L, newReq.version)
+        assertEquals(RestRequest.Method.GET, newReq.method)
+        assertEquals(FetchSourceContext.FETCH_SOURCE, newReq.srcContext)
+    }
+
+    fun `test get monitor request without src context`() {
+
+        val req = GetMonitorRequest("1234", 1L, RestRequest.Method.GET, null)
+        assertNotNull(req)
+
+        val out = BytesStreamOutput()
+        req.writeTo(out)
+        val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes)
+        val newReq = GetMonitorRequest(sin)
+        assertEquals("1234", newReq.monitorId)
+        assertEquals(1L, newReq.version)
+        assertEquals(RestRequest.Method.GET, newReq.method)
+        assertEquals(null, newReq.srcContext)
+    }
+
+    fun `test head monitor request`() {
+
+        val req = GetMonitorRequest("1234", 2L, RestRequest.Method.HEAD, FetchSourceContext.FETCH_SOURCE)
+        assertNotNull(req)
+
+        val out = BytesStreamOutput()
+        req.writeTo(out)
+        val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes)
+        val newReq = GetMonitorRequest(sin)
+        assertEquals("1234", newReq.monitorId)
+        assertEquals(2L, newReq.version)
+        assertEquals(RestRequest.Method.HEAD, newReq.method)
+        assertEquals(FetchSourceContext.FETCH_SOURCE, newReq.srcContext)
+    }
+}

src/test/kotlin/org/opensearch/commons/alerting/action/GetMonitorResponseTests.kt

@@ -0,0 +1,63 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.commons.alerting.action
+
+import org.opensearch.common.io.stream.BytesStreamOutput
+import org.opensearch.commons.alerting.model.CronSchedule
+import org.opensearch.commons.alerting.model.Monitor
+import org.opensearch.commons.alerting.randomUser
+import org.opensearch.core.common.io.stream.StreamInput
+import org.opensearch.test.OpenSearchTestCase
+import java.time.Instant
+import java.time.ZoneId
+
+class GetMonitorResponseTests : OpenSearchTestCase() {
+
+    fun `test get monitor response`() {
+        val req = GetMonitorResponse("1234", 1L, 2L, 0L, null, null)
+        assertNotNull(req)
+
+        val out = BytesStreamOutput()
+        req.writeTo(out)
+        val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes)
+        val newReq = GetMonitorResponse(sin)
+        assertEquals("1234", newReq.id)
+        assertEquals(1L, newReq.version)
+        assertEquals(null, newReq.monitor)
+    }
+
+    fun `test get monitor response with monitor`() {
+        val cronExpression = "31 * * * *" // Run at minute 31.
+        val testInstance = Instant.ofEpochSecond(1538164858L)
+
+        val cronSchedule = CronSchedule(cronExpression, ZoneId.of("Asia/Kolkata"), testInstance)
+        val monitor = Monitor(
+            id = "123",
+            version = 0L,
+            name = "test-monitor",
+            enabled = true,
+            schedule = cronSchedule,
+            lastUpdateTime = Instant.now(),
+            enabledTime = Instant.now(),
+            monitorType = Monitor.MonitorType.QUERY_LEVEL_MONITOR,
+            user = randomUser(),
+            schemaVersion = 0,
+            inputs = mutableListOf(),
+            triggers = mutableListOf(),
+            uiMetadata = mutableMapOf()
+        )
+        val req = GetMonitorResponse("1234", 1L, 2L, 0L, monitor, null)
+        assertNotNull(req)
+
+        val out = BytesStreamOutput()
+        req.writeTo(out)
+        val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes)
+        val newReq = GetMonitorResponse(sin)
+        assertEquals("1234", newReq.id)
+        assertEquals(1L, newReq.version)
+        assertNotNull(newReq.monitor)
+    }
+}