change serde failure handling to expect failure only in monitor object for Doc Level monitor fanout request

Signed-off-by: Surya Sashank Nistala <[email protected]>

Author: eirsep

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

@@ -40,18 +40,13 @@ class DocLevelMonitorFanOutRequest : ActionRequest, ToXContentObject {
     val workflowRunContext: WorkflowRunContext?
     val hasSerializationFailed: Boolean
 
-    init {
-        serializationFailedFlag = false
-    }
     companion object {
         // flag flipped to true whenever a safeRead*() method fails to serialize a field correctly
-        private var serializationFailedFlag: Boolean = false
         val log = LogManager.getLogger(DocLevelMonitorFanOutRequest::class.java)
         private fun safeReadMonitor(sin: StreamInput): Monitor =
             try {
                 Monitor.readFrom(sin)!!
             } catch (e: Exception) {
-                serializationFailedFlag = true
                 log.error("Error parsing monitor in Doc level monitor fanout request", e)
                 Monitor(
                     "failed_serde", NO_VERSION, "failed_serde", true,
@@ -65,7 +60,6 @@ class DocLevelMonitorFanOutRequest : ActionRequest, ToXContentObject {
             try {
                 sin.readBoolean()
             } catch (e: Exception) {
-                serializationFailedFlag = true
                 log.error("Error parsing boolean in Doc level monitor fanout request", e)
                 false
             }
@@ -74,7 +68,6 @@ class DocLevelMonitorFanOutRequest : ActionRequest, ToXContentObject {
             try {
                 MonitorMetadata.readFrom(sin)
             } catch (e: Exception) {
-                serializationFailedFlag = true
                 log.error("Error parsing monitor in Doc level monitor fanout request", e)
                 MonitorMetadata(
                     "failed_serde",
@@ -91,7 +84,6 @@ class DocLevelMonitorFanOutRequest : ActionRequest, ToXContentObject {
             try {
                 sin.readString()
             } catch (e: Exception) {
-                serializationFailedFlag = true
                 log.error("Error parsing string in Doc level monitor fanout request", e)
                 ""
             }
@@ -100,7 +92,6 @@ class DocLevelMonitorFanOutRequest : ActionRequest, ToXContentObject {
             try {
                 sin.readList(::ShardId)
             } catch (e: Exception) {
-                serializationFailedFlag = true
                 log.error("Error parsing shardId list in Doc level monitor fanout request", e)
                 emptyList()
             }
@@ -109,7 +100,6 @@ class DocLevelMonitorFanOutRequest : ActionRequest, ToXContentObject {
             try {
                 sin.readStringList()
             } catch (e: Exception) {
-                serializationFailedFlag = true
                 log.error("Error parsing string list in Doc level monitor fanout request", e)
                 emptyList()
             }
@@ -118,7 +108,6 @@ class DocLevelMonitorFanOutRequest : ActionRequest, ToXContentObject {
             try {
                 if (sin.readBoolean()) WorkflowRunContext(sin) else null
             } catch (e: Exception) {
-                serializationFailedFlag = true
                 log.error("Error parsing workflow context in Doc level monitor fanout request", e)
                 null
             }
@@ -128,7 +117,6 @@ class DocLevelMonitorFanOutRequest : ActionRequest, ToXContentObject {
             return try {
                 indexExecutionContext = IndexExecutionContext(sin)
                 while (sin.read() != 0) {
-                    serializationFailedFlag = true
                     // read and discard bytes until stream is entirely consumed
                     try {
                         sin.readByte()
@@ -139,7 +127,6 @@ class DocLevelMonitorFanOutRequest : ActionRequest, ToXContentObject {
             } catch (e: EOFException) {
                 indexExecutionContext
             } catch (e: Exception) {
-                serializationFailedFlag = true
                 log.error("Error parsing index execution context in Doc level monitor fanout request", e)
                 while (sin.read() != 0) {
                     try { // read and throw bytes until stream is entirely consumed
@@ -152,28 +139,6 @@ class DocLevelMonitorFanOutRequest : ActionRequest, ToXContentObject {
         }
     }
 
-    private constructor(
-        monitor: Monitor,
-        dryRun: Boolean,
-        monitorMetadata: MonitorMetadata,
-        executionId: String,
-        indexExecutionContext: IndexExecutionContext?,
-        shardIds: List<ShardId>,
-        concreteIndicesSeenSoFar: List<String>,
-        workflowRunContext: WorkflowRunContext?,
-        hasSerializationFailed: Boolean
-    ) : super() {
-        this.monitor = monitor
-        this.dryRun = dryRun
-        this.monitorMetadata = monitorMetadata
-        this.executionId = executionId
-        this.indexExecutionContext = indexExecutionContext
-        this.shardIds = shardIds
-        this.concreteIndicesSeenSoFar = concreteIndicesSeenSoFar
-        this.workflowRunContext = workflowRunContext
-        this.hasSerializationFailed = hasSerializationFailed ?: false
-    }
-
     constructor(
         monitor: Monitor,
         dryRun: Boolean,
@@ -196,16 +161,63 @@ class DocLevelMonitorFanOutRequest : ActionRequest, ToXContentObject {
     }
 
     @Throws(IOException::class)
-    constructor(sin: StreamInput) : this(
-        monitor = safeReadMonitor(sin),
-        dryRun = safeReadBoolean(sin),
-        monitorMetadata = safeReadMonitorMetadata(sin),
-        executionId = safeReadString(sin),
-        shardIds = safeReadShardIds(sin),
-        concreteIndicesSeenSoFar = safeReadStringList(sin),
-        workflowRunContext = safeReadWorkflowRunContext(sin),
-        indexExecutionContext = safeReadIndexExecutionContext(sin),
-        hasSerializationFailed = serializationFailedFlag
+    constructor(sin: StreamInput) : super() {
+        var monitorSerializationSucceded = true
+        var parsedMonitor = getDummyMonitor()
+        try {
+            parsedMonitor = Monitor(sin)
+        } catch (e: Exception) {
+            log.error("Error parsing monitor in Doc level monitor fanout request", e)
+            monitorSerializationSucceded = false
+            log.info("Force consuming stream in Doc level monitor fanout request")
+            while (sin.read() != 0) {
+                // read and discard bytes until stream is entirely consumed
+                try {
+                    sin.readByte()
+                } catch (_: EOFException) {
+                }
+            }
+        }
+        if (monitorSerializationSucceded) {
+            this.monitor = parsedMonitor
+            this.dryRun = sin.readBoolean()
+            this.monitorMetadata = MonitorMetadata.readFrom(sin)
+            this.executionId = sin.readString()
+            this.shardIds = sin.readList(::ShardId)
+            this.concreteIndicesSeenSoFar = sin.readStringList()
+            this.workflowRunContext = if (sin.readBoolean()) {
+                WorkflowRunContext(sin)
+            } else {
+                null
+            }
+            indexExecutionContext = IndexExecutionContext(sin)
+            this.hasSerializationFailed = false == monitorSerializationSucceded
+        } else {
+            this.monitor = parsedMonitor
+            this.dryRun = false
+            this.monitorMetadata = MonitorMetadata(
+                "failed_serde",
+                SequenceNumbers.UNASSIGNED_SEQ_NO,
+                SequenceNumbers.UNASSIGNED_PRIMARY_TERM,
+                "failed_serde",
+                emptyList(),
+                emptyMap(),
+                mutableMapOf()
+            )
+            this.executionId = ""
+            this.shardIds = emptyList()
+            this.concreteIndicesSeenSoFar = emptyList()
+            this.workflowRunContext = null
+            this.indexExecutionContext = null
+            this.hasSerializationFailed = false == monitorSerializationSucceded
+        }
+    }
+
+    private fun getDummyMonitor() = Monitor(
+        "failed_serde", NO_VERSION, "failed_serde", true,
+        IntervalSchedule(1, ChronoUnit.MINUTES), Instant.now(), Instant.now(), "",
+        null, NO_SCHEMA_VERSION, emptyList(), emptyList(), emptyMap(),
+        DataSources(), false, false, "failed"
     )
 
     @Throws(IOException::class)

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

@@ -7,7 +7,6 @@ package org.opensearch.commons.alerting.action
 
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertFalse
-import org.junit.Assert.assertNull
 import org.junit.Assert.assertTrue
 import org.junit.jupiter.api.Test
 import org.opensearch.common.io.stream.BytesStreamOutput
@@ -164,65 +163,6 @@ class DocLevelMonitorFanOutRequestTests {
         val docQuery = DocLevelQuery(query = "test_field:\"us-west-2\"", fields = listOf(), name = "3")
         val docLevelInput = DocLevelMonitorInput("description", listOf("test-index"), listOf(docQuery))
 
-        val trigger = randomDocumentLevelTrigger(condition = Script("return true"))
-        val monitor = randomDocumentLevelMonitor(
-            inputs = listOf(docLevelInput),
-            triggers = listOf(trigger),
-            enabled = true,
-            schedule = IntervalSchedule(1, ChronoUnit.MINUTES)
-        )
-        val monitorMetadata = MonitorMetadata(
-            "test",
-            SequenceNumbers.UNASSIGNED_SEQ_NO,
-            SequenceNumbers.UNASSIGNED_PRIMARY_TERM,
-            Monitor.NO_ID,
-            listOf(ActionExecutionTime("", Instant.now())),
-            mutableMapOf("index" to mutableMapOf("1" to "1")),
-            mutableMapOf("test-index" to ".opensearch-sap-test_windows-queries-000001")
-        )
-        val indexExecutionContext = IndexExecutionContext(
-            listOf(docQuery),
-            mutableMapOf("index" to mutableMapOf("1" to "1")),
-            mutableMapOf("index" to mutableMapOf("1" to "1")),
-            "test-index",
-            "test-index",
-            listOf("test-index"),
-            listOf("test-index"),
-            listOf("test-field"),
-            listOf("1", "2")
-        )
-        val workflowRunContext = WorkflowRunContext(
-            Workflow.NO_ID,
-            Workflow.NO_ID,
-            Monitor.NO_ID,
-            mutableMapOf("index" to listOf("1")),
-            true,
-            listOf("finding1")
-        )
-        val docLevelMonitorFanOutRequest = DocLevelMonitorFanOutRequest(
-            monitor,
-            false,
-            monitorMetadata,
-            UUID.randomUUID().toString(),
-            indexExecutionContext,
-            listOf(ShardId("test-index", UUID.randomUUID().toString(), 0)),
-            listOf("test-index"),
-            workflowRunContext
-        )
-        val out = BytesStreamOutput()
-        monitor.writeTo(out)
-        val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes)
-        val newDocLevelMonitorFanOutRequest = DocLevelMonitorFanOutRequest(sin)
-        assertNull(newDocLevelMonitorFanOutRequest.indexExecutionContext)
-        assertTrue(newDocLevelMonitorFanOutRequest.hasSerializationFailed)
-        assertEquals(sin.read(), -1)
-    }
-
-    @Test
-    fun `test doc level monitor fan out request as stream when there are additional bytes left to handle`() {
-        val docQuery = DocLevelQuery(query = "test_field:\"us-west-2\"", fields = listOf(), name = "3")
-        val docLevelInput = DocLevelMonitorInput("description", listOf("test-index"), listOf(docQuery))
-
         val trigger = randomDocumentLevelTrigger(condition = Script("return true"))
         val monitor = randomDocumentLevelMonitor(
             inputs = listOf(docLevelInput),
@@ -268,16 +208,10 @@ class DocLevelMonitorFanOutRequestTests {
             workflowRunContext
         )
         val out = BytesStreamOutput()
+        out.writeString(UUID.randomUUID().toString())
         docLevelMonitorFanOutRequest.writeTo(out)
-        out.writeByte(Byte.MIN_VALUE)
         val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes)
         val newDocLevelMonitorFanOutRequest = DocLevelMonitorFanOutRequest(sin)
-        assertEquals(docLevelMonitorFanOutRequest.monitor, newDocLevelMonitorFanOutRequest.monitor)
-        assertEquals(docLevelMonitorFanOutRequest.executionId, newDocLevelMonitorFanOutRequest.executionId)
-        assertEquals(docLevelMonitorFanOutRequest.monitorMetadata, newDocLevelMonitorFanOutRequest.monitorMetadata)
-        assertEquals(docLevelMonitorFanOutRequest.indexExecutionContext, newDocLevelMonitorFanOutRequest.indexExecutionContext)
-        assertEquals(docLevelMonitorFanOutRequest.shardIds, newDocLevelMonitorFanOutRequest.shardIds)
-        assertEquals(docLevelMonitorFanOutRequest.workflowRunContext, newDocLevelMonitorFanOutRequest.workflowRunContext)
         assertEquals(sin.read(), 0)
         assertTrue(newDocLevelMonitorFanOutRequest.hasSerializationFailed)
     }