diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/DocumentLevelMonitorRunner.kt b/alerting/src/main/kotlin/org/opensearch/alerting/DocumentLevelMonitorRunner.kt index f7ed8a4c5..96437f433 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/DocumentLevelMonitorRunner.kt +++ b/alerting/src/main/kotlin/org/opensearch/alerting/DocumentLevelMonitorRunner.kt @@ -104,8 +104,14 @@ class DocumentLevelMonitorRunner : MonitorRunner() { val queries: List = docLevelMonitorInput.queries - val lastRunContext = if (monitorMetadata.lastRunContext.isNullOrEmpty()) mutableMapOf() - else monitorMetadata.lastRunContext.toMutableMap() as MutableMap> + // Safe reconstruction: filter non-Map values to avoid ClassCastException on stale or malformed entries. + @Suppress("UNCHECKED_CAST") + val lastRunContext: MutableMap> = if (monitorMetadata.lastRunContext.isNullOrEmpty()) mutableMapOf() + else monitorMetadata.lastRunContext + .entries + .filter { it.value is Map<*, *> } + .associate { it.key to (it.value as Map).toMutableMap() } + .toMutableMap() val updatedLastRunContext = lastRunContext.toMutableMap() @@ -461,7 +467,8 @@ class DocumentLevelMonitorRunner : MonitorRunner() { for (fanOutResponse in docLevelMonitorFanOutResponses) { if (fanOutResponse.exception == null) { // fanOutResponse.lastRunContexts //updatedContexts for relevant shards - val indexLastRunContext = updatedLastRunContext[indexName] as MutableMap + // indexName is always a key of updatedLastRunContext (we iterate its keys above) + val indexLastRunContext = updatedLastRunContext[indexName]!! if (fanOutResponse.lastRunContexts.contains("index") && fanOutResponse.lastRunContexts["index"] == indexName) { fanOutResponse.lastRunContexts.keys.forEach { diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/MonitorFanOutUtils.kt b/alerting/src/main/kotlin/org/opensearch/alerting/MonitorFanOutUtils.kt index 294d154d9..61520d333 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/MonitorFanOutUtils.kt +++ b/alerting/src/main/kotlin/org/opensearch/alerting/MonitorFanOutUtils.kt @@ -51,15 +51,15 @@ fun distributeShards( * It preserves existing sequence numbers and initializes new shards with UNASSIGNED_SEQ_NO. * * @param lastRunContext The previous run context from monitor metadata - * @param monitorCtx The execution context containing cluster service * @param index The name of the index for which the context is being initialized - * @return A map containing the initialized last run context + * @param shardCount The total shard count for the index + * @return A map containing the initialized last run context; a new MutableMap owned by the caller — mutations to the returned map do not affect the input [lastRunContext]. */ fun initializeNewLastRunContext( lastRunContext: Map, index: String, shardCount: Int, -): Map { +): MutableMap { val updatedLastRunContext = lastRunContext.toMutableMap() // Only initialize shards that don't already have a sequence number diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/MonitorMetadataService.kt b/alerting/src/main/kotlin/org/opensearch/alerting/MonitorMetadataService.kt index 1de16d48e..63df843dc 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/MonitorMetadataService.kt +++ b/alerting/src/main/kotlin/org/opensearch/alerting/MonitorMetadataService.kt @@ -210,9 +210,19 @@ object MonitorMetadataService : else if (monitor.monitorType.endsWith(Monitor.MonitorType.DOC_LEVEL_MONITOR.value)) (monitor.inputs[0] as RemoteDocLevelMonitorInput).docLevelMonitorInput.indices[0] else null - val runContext = if (monitor.monitorType.endsWith(Monitor.MonitorType.DOC_LEVEL_MONITOR.value)) - createFullRunContext(monitorIndex, metadata.lastRunContext.toMutableMap() as MutableMap>) - else null + val runContext = if (monitor.monitorType.endsWith(Monitor.MonitorType.DOC_LEVEL_MONITOR.value)) { + // Safe reconstruction: filter out any entries whose value is not a Map (e.g. stale scalar entries + // from prior migrations or XContent parse paths that produce non-MutableMap values), then copy + // each surviving value into a fresh MutableMap. This avoids ClassCastException on non-empty but + // wrongly-shaped lastRunContext. + @Suppress("UNCHECKED_CAST") + val lastRunCtx: MutableMap> = metadata.lastRunContext + .entries + .filter { it.value is Map<*, *> } + .associate { it.key to (it.value as Map).toMutableMap() } + .toMutableMap() + createFullRunContext(monitorIndex, lastRunCtx.ifEmpty { null }) + } else null return if (runContext != null) { metadata.copy( lastRunContext = runContext diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/remote/monitors/RemoteDocumentLevelMonitorRunner.kt b/alerting/src/main/kotlin/org/opensearch/alerting/remote/monitors/RemoteDocumentLevelMonitorRunner.kt index c12356cab..b8a94372b 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/remote/monitors/RemoteDocumentLevelMonitorRunner.kt +++ b/alerting/src/main/kotlin/org/opensearch/alerting/remote/monitors/RemoteDocumentLevelMonitorRunner.kt @@ -63,8 +63,14 @@ class RemoteDocumentLevelMonitorRunner : MonitorRunner() { workflowRunContext?.workflowMetadataId ) logger.info(monitorMetadata.lastRunContext.toMutableMap().toString()) - val lastRunContext = if (monitorMetadata.lastRunContext.isNullOrEmpty()) mutableMapOf() - else monitorMetadata.lastRunContext.toMutableMap() as MutableMap> + // Safe reconstruction: filter non-Map values to avoid ClassCastException on stale or malformed entries. + @Suppress("UNCHECKED_CAST") + val lastRunContext: MutableMap> = if (monitorMetadata.lastRunContext.isNullOrEmpty()) mutableMapOf() + else monitorMetadata.lastRunContext + .entries + .filter { it.value is Map<*, *> } + .associate { it.key to (it.value as Map).toMutableMap() } + .toMutableMap() val updatedLastRunContext = lastRunContext.toMutableMap() val remoteDocLevelMonitorInput = monitor.inputs[0] as RemoteDocLevelMonitorInput @@ -126,7 +132,7 @@ class RemoteDocumentLevelMonitorRunner : MonitorRunner() { indexLastRunContext.toMutableMap(), concreteIndexName, shardCount - ) as MutableMap + ) if (IndexUtils.isAlias(indexName, monitorCtx.clusterService!!.state()) || IndexUtils.isDataStream(indexName, monitorCtx.clusterService!!.state()) ) { @@ -264,7 +270,8 @@ class RemoteDocumentLevelMonitorRunner : MonitorRunner() { for (fanOutResponse in docLevelMonitorFanOutResponses) { if (fanOutResponse.exception == null) { // fanOutResponse.lastRunContexts //updatedContexts for relevant shards - val indexLastRunContext = updatedLastRunContext[indexName] as MutableMap + // indexName is always a key of updatedLastRunContext (we iterate its keys above) + val indexLastRunContext = updatedLastRunContext[indexName]!! if (fanOutResponse.lastRunContexts.contains(indexName)) { (fanOutResponse.lastRunContexts[indexName] as Map).forEach { diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/util/DocLevelMonitorQueries.kt b/alerting/src/main/kotlin/org/opensearch/alerting/util/DocLevelMonitorQueries.kt index fa62a23dc..5436f9da7 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/util/DocLevelMonitorQueries.kt +++ b/alerting/src/main/kotlin/org/opensearch/alerting/util/DocLevelMonitorQueries.kt @@ -228,18 +228,26 @@ class DocLevelMonitorQueries(private val client: Client, private val clusterServ // Compute full path relative to root val fullPath = if (currentPath.isEmpty()) it.key else "$currentPath.${it.key}" - val nodeProps = it.value as MutableMap + val nodeProps = (it.value as Map).toMutableMap() // If it has type property and type is not "nested" then this is a leaf if (nodeProps.containsKey(TYPE) && nodeProps[TYPE] != NESTED) { // At this point we know full path of node, so we add it to output array flattenPaths.put(fullPath, nodeProps) // Calls processLeafFn and gets old node name, new node name and new properties of node. // This is all information we need to update this node - val (oldName, newName, props) = processLeafFn(it.key, fullPath, it.value as MutableMap) + // Reuse nodeProps rather than creating a second independent copy from it.value. + val (oldName, newName, props) = processLeafFn(it.key, fullPath, nodeProps) newNodes.add(Triple(oldName, newName, props)) } else if (nodeProps.containsKey(PROPERTIES) && nodeProps[PROPERTIES] != null) { - // Internal(non-leaf) node - visit children - traverseMappingsAndUpdate(nodeProps[PROPERTIES] as MutableMap, fullPath, processLeafFn, flattenPaths) + // Internal(non-leaf) node - visit children. + // Copy nodeProps[PROPERTIES] into childProps, recurse on childProps, then write + // childProps back into nodeProps[PROPERTIES] so the parent node sees updated children. + @Suppress("UNCHECKED_CAST") + val childProps = (nodeProps[PROPERTIES] as Map).toMutableMap() + traverseMappingsAndUpdate(childProps, fullPath, processLeafFn, flattenPaths) + // Write the mutated child properties back so the parent node sees the updates. + nodeProps[PROPERTIES] = childProps + newNodes.add(Triple(it.key, it.key, nodeProps as Any)) } } // Here we can update all processed leaves in tree diff --git a/alerting/src/test/kotlin/org/opensearch/alerting/MonitorFanOutUtilsTests.kt b/alerting/src/test/kotlin/org/opensearch/alerting/MonitorFanOutUtilsTests.kt index 91afb23b9..601090342 100644 --- a/alerting/src/test/kotlin/org/opensearch/alerting/MonitorFanOutUtilsTests.kt +++ b/alerting/src/test/kotlin/org/opensearch/alerting/MonitorFanOutUtilsTests.kt @@ -6,6 +6,7 @@ package org.opensearch.alerting import org.opensearch.core.index.Index import org.opensearch.core.index.shard.ShardId +import org.opensearch.index.seqno.SequenceNumbers import org.opensearch.test.OpenSearchTestCase class MonitorFanOutUtilsTests : OpenSearchTestCase() { @@ -131,4 +132,45 @@ class MonitorFanOutUtilsTests : OpenSearchTestCase() { } assertEquals(expectedTotalShardCount, shardCount) } + + fun `test initializeNewLastRunContext returns mutable map`() { + val result = initializeNewLastRunContext(emptyMap(), "test-index", 3) + + assertTrue("result must be a MutableMap", result is MutableMap) + assertTrue("result must contain shard key '0'", result.containsKey("0")) + assertTrue("result must contain shard key '1'", result.containsKey("1")) + assertTrue("result must contain shard key '2'", result.containsKey("2")) + assertEquals("shards_count must equal 3", 3, result["shards_count"]) + assertEquals("index must equal 'test-index'", "test-index", result["index"]) + + // Verify it is truly mutable — must not throw UnsupportedOperationException + result["extra_key"] = "extra_value" + assertEquals("extra_value", result["extra_key"]) + } + + fun `test initializeNewLastRunContext preserves existing seq numbers`() { + val existingContext = mapOf("0" to 42L) + + val result = initializeNewLastRunContext(existingContext, "preserve-index", 2) + + assertEquals("Existing sequence number for shard 0 must be preserved", 42L, result["0"]) + assertEquals( + "New shard 1 must be initialized to UNASSIGNED_SEQ_NO", + SequenceNumbers.UNASSIGNED_SEQ_NO, + result["1"] + ) + assertEquals(2, result["shards_count"]) + } + + fun `test initializeNewLastRunContext with Collections emptyMap does not throw`() { + val emptyJavaMap = java.util.Collections.emptyMap() + + val result = initializeNewLastRunContext(emptyJavaMap, "empty-map-index", 2) + + assertNotNull("result must not be null", result) + assertTrue("result must be mutable", result is MutableMap) + // Verify mutability by inserting a new entry + result["probe"] = "ok" + assertEquals("ok", result["probe"]) + } } diff --git a/alerting/src/test/kotlin/org/opensearch/alerting/util/DocLevelMonitorQueriesTraversalTests.kt b/alerting/src/test/kotlin/org/opensearch/alerting/util/DocLevelMonitorQueriesTraversalTests.kt new file mode 100644 index 000000000..bbbd99c1a --- /dev/null +++ b/alerting/src/test/kotlin/org/opensearch/alerting/util/DocLevelMonitorQueriesTraversalTests.kt @@ -0,0 +1,115 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.alerting.util + +import org.mockito.Mockito.mock +import org.opensearch.cluster.service.ClusterService +import org.opensearch.test.OpenSearchTestCase +import org.opensearch.transport.client.Client + +/** + * Unit tests for [DocLevelMonitorQueries.traverseMappingsAndUpdate] focused on the + * ClassCastException guard: values returned by OpenSearch may be unmodifiable or empty maps, + * and the fix uses `(it.value as Map).toMutableMap()` instead of a direct + * `as MutableMap` cast. + */ +class DocLevelMonitorQueriesTraversalTests : OpenSearchTestCase() { + + private val client: Client = mock(Client::class.java) + private val clusterService: ClusterService = mock(ClusterService::class.java) + private val queries = DocLevelMonitorQueries(client, clusterService) + + private val identityLeafFn: (String, String, MutableMap) -> Triple> = + { key, _, props -> Triple(key, key, props) } + + fun `test traverseMappingsAndUpdate with unmodifiable map values does not throw`() { + // Simulates what OpenSearch cluster state may return: values wrapped in Collections.unmodifiableMap + val unmodifiableProps: Map = java.util.Collections.unmodifiableMap( + mapOf("type" to "keyword") + ) + val node = mutableMapOf("field1" to unmodifiableProps) + val flattenPaths = mutableMapOf>() + + // Must not throw ClassCastException + queries.traverseMappingsAndUpdate(node, "", identityLeafFn, flattenPaths) + + assertTrue("flattenPaths must contain leaf 'field1'", flattenPaths.containsKey("field1")) + } + + fun `test traverseMappingsAndUpdate with empty map value does not throw`() { + // An entry whose value is Collections.emptyMap() has no TYPE key and no PROPERTIES key, + // so it is silently skipped — no leaf added, no recursion. + val node = mutableMapOf( + "empty_field" to java.util.Collections.emptyMap() + ) + val flattenPaths = mutableMapOf>() + + // Must not throw ClassCastException or UnsupportedOperationException + queries.traverseMappingsAndUpdate(node, "", identityLeafFn, flattenPaths) + + assertTrue("No leaves should be registered for an empty-map entry", flattenPaths.isEmpty()) + } + + fun `test traverseMappingsAndUpdate renames leaf node`() { + val node = mutableMapOf( + "field1" to mutableMapOf("type" to "keyword") + ) + val flattenPaths = mutableMapOf>() + val renamingLeafFn: (String, String, MutableMap) -> Triple> = + { key, _, props -> + val newKey = "${key}_renamed" + Triple(key, newKey, props) + } + + queries.traverseMappingsAndUpdate(node, "", renamingLeafFn, flattenPaths) + + assertTrue("node must contain the renamed key", node.containsKey("field1_renamed")) + assertFalse("node must not retain the old key after rename", node.containsKey("field1")) + } + + fun `test traverseMappingsAndUpdate recurses into nested properties`() { + // Two-level map: outer -> { properties -> { inner -> { type -> text } } } + val innerField = mutableMapOf("type" to "text") + val innerProperties = mutableMapOf("inner" to innerField) + val outerField = mutableMapOf(DocLevelMonitorQueries.PROPERTIES to innerProperties) + val node = mutableMapOf("outer" to outerField) + val flattenPaths = mutableMapOf>() + + queries.traverseMappingsAndUpdate(node, "", identityLeafFn, flattenPaths) + + assertTrue( + "flattenPaths must contain fully-qualified path 'outer.inner'", + flattenPaths.containsKey("outer.inner") + ) + assertFalse( + "outer itself is not a leaf and must not appear in flattenPaths", + flattenPaths.containsKey("outer") + ) + } + + fun `test traverseMappingsAndUpdate with unmodifiable nested properties does not throw`() { + // Simulates nested properties returned as an unmodifiable map from OpenSearch + val innerField: Map = java.util.Collections.unmodifiableMap( + mapOf("type" to "text") + ) + val innerProperties: Map = java.util.Collections.unmodifiableMap( + mapOf("inner" to innerField) + ) + val outerField: Map = java.util.Collections.unmodifiableMap( + mapOf(DocLevelMonitorQueries.PROPERTIES to innerProperties) + ) + val node = mutableMapOf("outer" to outerField) + val flattenPaths = mutableMapOf>() + + // Must not throw ClassCastException when recursing into unmodifiable properties map + queries.traverseMappingsAndUpdate(node, "", identityLeafFn, flattenPaths) + + assertTrue( + "flattenPaths must contain 'outer.inner' even when all values are unmodifiable", + flattenPaths.containsKey("outer.inner") + ) + } +}