"adding batch findings request and action"

This reverts commit ad4d0c2.

Author: toepkerd-zz

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

@@ -27,6 +27,7 @@ import org.opensearch.commons.alerting.action.IndexMonitorRequest
 import org.opensearch.commons.alerting.action.IndexMonitorResponse
 import org.opensearch.commons.alerting.action.IndexWorkflowRequest
 import org.opensearch.commons.alerting.action.IndexWorkflowResponse
+import org.opensearch.commons.alerting.action.PublishBatchFindingsRequest
 import org.opensearch.commons.alerting.action.PublishFindingsRequest
 import org.opensearch.commons.alerting.action.SearchMonitorRequest
 import org.opensearch.commons.alerting.action.SubscribeFindingsResponse
@@ -268,6 +269,24 @@ object AlertingPluginInterface {
         )
     }
 
+    fun publishBatchFindings(
+        client: NodeClient,
+        request: PublishBatchFindingsRequest,
+        listener: ActionListener<SubscribeFindingsResponse>
+    ) {
+        client.execute(
+            AlertingActions.SUBSCRIBE_BATCH_FINDINGS_ACTION_TYPE,
+            request,
+            wrapActionListener(listener) { response ->
+                recreateObject(response) {
+                    SubscribeFindingsResponse(
+                        it
+                    )
+                }
+            }
+        )
+    }
+
     /**
      * Acknowledge Chained Alerts interface.
      * @param client Node client for making transport action

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

@@ -19,6 +19,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 SUBSCRIBE_BATCH_FINDINGS_ACTION_NAME = "cluster:admin/opensearch/alerting/findings/batch/subscribe"
     const val GET_MONITOR_ACTION_NAME = "cluster:admin/opendistro/alerting/monitor/get"
     const val SEARCH_MONITORS_ACTION_NAME = "cluster:admin/opendistro/alerting/monitor/search"
     const val INDEX_COMMENT_ACTION_NAME = "cluster:admin/opensearch/alerting/comments/write"
@@ -65,6 +66,10 @@ object AlertingActions {
     val SUBSCRIBE_FINDINGS_ACTION_TYPE =
         ActionType(SUBSCRIBE_FINDINGS_ACTION_NAME, ::SubscribeFindingsResponse)
 
+    @JvmField
+    val SUBSCRIBE_BATCH_FINDINGS_ACTION_TYPE =
+        ActionType(SUBSCRIBE_BATCH_FINDINGS_ACTION_NAME, ::SubscribeFindingsResponse)
+
     @JvmField
     val ACKNOWLEDGE_CHAINED_ALERTS_ACTION_TYPE =
         ActionType(ACKNOWLEDGE_CHAINED_ALERTS_ACTION_NAME, ::AcknowledgeAlertResponse)

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

@@ -0,0 +1,39 @@
+package org.opensearch.commons.alerting.action
+
+import org.opensearch.action.ActionRequest
+import org.opensearch.action.ActionRequestValidationException
+import org.opensearch.commons.alerting.model.Finding
+import org.opensearch.core.common.io.stream.StreamInput
+import org.opensearch.core.common.io.stream.StreamOutput
+import java.io.IOException
+import java.util.Collections
+
+class PublishBatchFindingsRequest : ActionRequest {
+
+    val monitorId: String
+
+    val findings: List<Finding>
+
+    constructor(
+        monitorId: String,
+        findings: List<Finding>
+    ) : super() {
+        this.monitorId = monitorId
+        this.findings = findings
+    }
+
+    @Throws(IOException::class)
+    constructor(sin: StreamInput) : this(
+        monitorId = sin.readString(),
+        findings = Collections.unmodifiableList(sin.readList(::Finding))
+    )
+
+    override fun validate(): ActionRequestValidationException? {
+        return null
+    }
+
+    override fun writeTo(out: StreamOutput) {
+        out.writeString(monitorId)
+        out.writeCollection(findings)
+    }
+}

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

@@ -6,26 +6,25 @@ import org.opensearch.commons.alerting.model.Finding
 import org.opensearch.core.common.io.stream.StreamInput
 import org.opensearch.core.common.io.stream.StreamOutput
 import java.io.IOException
-import java.util.Collections
 
 class PublishFindingsRequest : ActionRequest {
 
     val monitorId: String
 
-    val findings: List<Finding>
+    val finding: Finding
 
     constructor(
         monitorId: String,
-        findings: List<Finding>
+        finding: Finding
     ) : super() {
         this.monitorId = monitorId
-        this.findings = findings
+        this.finding = finding
     }
 
     @Throws(IOException::class)
     constructor(sin: StreamInput) : this(
         monitorId = sin.readString(),
-        findings = Collections.unmodifiableList(sin.readList(::Finding))
+        finding = Finding.readFrom(sin)
     )
 
     override fun validate(): ActionRequestValidationException? {
@@ -34,6 +33,6 @@ class PublishFindingsRequest : ActionRequest {
 
     override fun writeTo(out: StreamOutput) {
         out.writeString(monitorId)
-        out.writeCollection(findings)
+        finding.writeTo(out)
     }
 }

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

@@ -0,0 +1,29 @@
+package org.opensearch.commons.alerting.action
+
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertNotNull
+import org.junit.jupiter.api.Test
+import org.opensearch.common.io.stream.BytesStreamOutput
+import org.opensearch.commons.alerting.randomFinding
+import org.opensearch.core.common.io.stream.StreamInput
+
+class PublishBatchFindingsRequestTests {
+
+    @Test
+    fun `test publish batch findings request`() {
+        val findings = listOf(randomFinding(), randomFinding())
+        val monitorId = "mid"
+        val req = PublishBatchFindingsRequest(monitorId, findings)
+        assertNotNull(req)
+        assertEquals(monitorId, req.monitorId)
+        assertEquals(findings, req.findings)
+
+        val out = BytesStreamOutput()
+        req.writeTo(out)
+        val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes)
+        val newReq = PublishBatchFindingsRequest(sin)
+        assertEquals(monitorId, newReq.monitorId)
+        assertEquals(findings.size, newReq.findings.size)
+        assert(newReq.findings.zip(findings).all { (f1, f2) -> f1.id == f2.id })
+    }
+}

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

@@ -11,19 +11,18 @@ class PublishFindingsRequestTests {
 
     @Test
     fun `test delete monitor request`() {
-        val findings = listOf(randomFinding())
+        val finding = randomFinding()
         val monitorId = "mid"
-        val req = PublishFindingsRequest(monitorId, findings)
+        val req = PublishFindingsRequest(monitorId, finding)
         assertNotNull(req)
         assertEquals(monitorId, req.monitorId)
-        assertEquals(findings, req.findings)
+        assertEquals(finding, req.finding)
 
         val out = BytesStreamOutput()
         req.writeTo(out)
         val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes)
         val newReq = PublishFindingsRequest(sin)
         assertEquals(monitorId, newReq.monitorId)
-        assertEquals(findings.size, newReq.findings.size)
-        assertEquals(findings[0].id, newReq.findings[0].id)
+        assertEquals(finding.id, newReq.finding.id)
     }
 }