add version checks in serde flows for Fan out request changes since 2.15 (#811)

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

Author: eirsep

src/main/kotlin/org/opensearch/commons/alerting/model/DocLevelMonitorInput.kt

@@ -1,5 +1,6 @@
 package org.opensearch.commons.alerting.model
 
+import org.opensearch.Version
 import org.opensearch.common.CheckedFunction
 import org.opensearch.core.ParseField
 import org.opensearch.core.common.io.stream.StreamInput
@@ -23,7 +24,7 @@ data class DocLevelMonitorInput(
         sin.readString(), // description
         sin.readStringList(), // indices
         sin.readList(::DocLevelQuery), // docLevelQueries
-        sin.readOptionalBoolean() // fanoutEnabled
+        if (sin.version.onOrAfter(Version.V_2_15_0)) sin.readOptionalBoolean() else true // fanoutEnabled
     )
 
     override fun asTemplateArg(): Map<String, Any> {
@@ -43,7 +44,9 @@ data class DocLevelMonitorInput(
         out.writeString(description)
         out.writeStringCollection(indices)
         out.writeCollection(queries)
-        out.writeOptionalBoolean(fanoutEnabled)
+        if (out.version.onOrAfter(Version.V_2_15_0)) {
+            out.writeOptionalBoolean(fanoutEnabled)
+        }
     }
 
     override fun toXContent(builder: XContentBuilder, params: ToXContent.Params): XContentBuilder {

src/main/kotlin/org/opensearch/commons/alerting/model/IndexExecutionContext.kt

@@ -5,6 +5,7 @@
 
 package org.opensearch.commons.alerting.model
 
+import org.opensearch.Version
 import org.opensearch.core.common.io.stream.StreamInput
 import org.opensearch.core.common.io.stream.StreamOutput
 import org.opensearch.core.common.io.stream.Writeable
@@ -36,7 +37,7 @@ data class IndexExecutionContext(
         concreteIndexNames = sin.readStringList(),
         conflictingFields = sin.readStringList(),
         docIds = sin.readOptionalStringList(),
-        findingIds = sin.readOptionalStringList()
+        findingIds = if (sin.version.onOrAfter(Version.V_2_15_0)) sin.readOptionalStringList() else emptyList()
     )
 
     override fun writeTo(out: StreamOutput?) {
@@ -49,7 +50,9 @@ data class IndexExecutionContext(
         out.writeStringCollection(concreteIndexNames)
         out.writeStringCollection(conflictingFields)
         out.writeOptionalStringCollection(docIds)
-        out.writeOptionalStringCollection(findingIds)
+        if (out.version.onOrAfter(Version.V_2_15_0)) {
+            out.writeOptionalStringCollection(findingIds)
+        }
     }
 
     override fun toXContent(builder: XContentBuilder?, params: ToXContent.Params?): XContentBuilder {

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

@@ -1,5 +1,6 @@
 package org.opensearch.commons.alerting.model
 
+import org.opensearch.Version
 import org.opensearch.common.CheckedFunction
 import org.opensearch.commons.alerting.model.remote.monitors.RemoteMonitorTrigger
 import org.opensearch.commons.alerting.util.IndexUtils.Companion.MONITOR_MAX_INPUTS
@@ -112,8 +113,16 @@ data class Monitor(
         } else {
             DataSources()
         },
-        deleteQueryIndexInEveryRun = sin.readOptionalBoolean(),
-        shouldCreateSingleAlertForFindings = sin.readOptionalBoolean(),
+        deleteQueryIndexInEveryRun = if (sin.version.onOrAfter(Version.V_2_15_0)) {
+            sin.readOptionalBoolean()
+        } else {
+            false
+        },
+        shouldCreateSingleAlertForFindings = if (sin.version.onOrAfter(Version.V_2_15_0)) {
+            sin.readOptionalBoolean()
+        } else {
+            false
+        },
         owner = sin.readOptionalString()
     )
 
@@ -226,8 +235,12 @@ data class Monitor(
         out.writeMap(uiMetadata)
         out.writeBoolean(dataSources != null) // for backward compatibility with pre-existing monitors which don't have datasources field
         dataSources.writeTo(out)
-        out.writeOptionalBoolean(deleteQueryIndexInEveryRun)
-        out.writeOptionalBoolean(shouldCreateSingleAlertForFindings)
+        if (out.version.onOrAfter(Version.V_2_15_0)) {
+            out.writeOptionalBoolean(deleteQueryIndexInEveryRun)
+        }
+        if (out.version.onOrAfter(Version.V_2_15_0)) {
+            out.writeOptionalBoolean(shouldCreateSingleAlertForFindings)
+        }
         out.writeOptionalString(owner)
     }
 

src/main/kotlin/org/opensearch/commons/alerting/model/WorkflowRunContext.kt

@@ -5,6 +5,7 @@
 
 package org.opensearch.commons.alerting.model
 
+import org.opensearch.Version
 import org.opensearch.core.common.io.stream.StreamInput
 import org.opensearch.core.common.io.stream.StreamOutput
 import org.opensearch.core.common.io.stream.Writeable
@@ -28,12 +29,12 @@ data class WorkflowRunContext(
     }
 
     constructor(sin: StreamInput) : this(
-        sin.readString(),
-        sin.readString(),
-        sin.readOptionalString(),
-        sin.readMap() as Map<String, List<String>>,
-        sin.readBoolean(),
-        sin.readOptionalStringList()
+        workflowId = sin.readString(),
+        workflowMetadataId = sin.readString(),
+        chainedMonitorId = sin.readOptionalString(),
+        matchingDocIdsPerIndex = sin.readMap() as Map<String, List<String>>,
+        auditDelegateMonitorAlerts = sin.readBoolean(),
+        findingIds = if (sin.version.onOrAfter(Version.V_2_15_0)) sin.readOptionalStringList() else emptyList()
     )
 
     override fun writeTo(out: StreamOutput) {
@@ -42,7 +43,9 @@ data class WorkflowRunContext(
         out.writeOptionalString(chainedMonitorId)
         out.writeMap(matchingDocIdsPerIndex)
         out.writeBoolean(auditDelegateMonitorAlerts)
-        out.writeOptionalStringCollection(findingIds)
+        if (out.version.onOrAfter(Version.V_2_15_0)) {
+            out.writeOptionalStringCollection(findingIds)
+        }
     }
 
     override fun toXContent(builder: XContentBuilder, params: ToXContent.Params?): XContentBuilder {