Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -104,8 +104,14 @@ class DocumentLevelMonitorRunner : MonitorRunner() {

val queries: List<DocLevelQuery> = docLevelMonitorInput.queries

val lastRunContext = if (monitorMetadata.lastRunContext.isNullOrEmpty()) mutableMapOf()
else monitorMetadata.lastRunContext.toMutableMap() as MutableMap<String, MutableMap<String, Any>>
// Safe reconstruction: filter non-Map values to avoid ClassCastException on stale or malformed entries.
@Suppress("UNCHECKED_CAST")
val lastRunContext: MutableMap<String, MutableMap<String, Any>> = if (monitorMetadata.lastRunContext.isNullOrEmpty()) mutableMapOf()
else monitorMetadata.lastRunContext
.entries
.filter { it.value is Map<*, *> }
.associate { it.key to (it.value as Map<String, Any>).toMutableMap() }
.toMutableMap()

val updatedLastRunContext = lastRunContext.toMutableMap()

Expand Down Expand Up @@ -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<String, Any>
// 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 {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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<String, Any>,
index: String,
shardCount: Int,
): Map<String, Any> {
): MutableMap<String, Any> {
val updatedLastRunContext = lastRunContext.toMutableMap()

// Only initialize shards that don't already have a sequence number
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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<String, MutableMap<String, Any>>)
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<String, MutableMap<String, Any>> = metadata.lastRunContext
.entries
.filter { it.value is Map<*, *> }
.associate { it.key to (it.value as Map<String, Any>).toMutableMap() }
.toMutableMap()
createFullRunContext(monitorIndex, lastRunCtx.ifEmpty { null })
} else null
return if (runContext != null) {
metadata.copy(
lastRunContext = runContext
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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<String, MutableMap<String, Any>>
// Safe reconstruction: filter non-Map values to avoid ClassCastException on stale or malformed entries.
@Suppress("UNCHECKED_CAST")
val lastRunContext: MutableMap<String, MutableMap<String, Any>> = if (monitorMetadata.lastRunContext.isNullOrEmpty()) mutableMapOf()
else monitorMetadata.lastRunContext
.entries
.filter { it.value is Map<*, *> }
.associate { it.key to (it.value as Map<String, Any>).toMutableMap() }
.toMutableMap()
val updatedLastRunContext = lastRunContext.toMutableMap()

val remoteDocLevelMonitorInput = monitor.inputs[0] as RemoteDocLevelMonitorInput
Expand Down Expand Up @@ -126,7 +132,7 @@ class RemoteDocumentLevelMonitorRunner : MonitorRunner() {
indexLastRunContext.toMutableMap(),
concreteIndexName,
shardCount
) as MutableMap<String, Any>
)
if (IndexUtils.isAlias(indexName, monitorCtx.clusterService!!.state()) ||
IndexUtils.isDataStream(indexName, monitorCtx.clusterService!!.state())
) {
Expand Down Expand Up @@ -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<String, Any>
// 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<String, Any>).forEach {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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<String, Any>
val nodeProps = (it.value as Map<String, Any>).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<String, Any>)
// 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<String, Any>, 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<String, Any>).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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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() {
Expand Down Expand Up @@ -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<String, Any>("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<String, Any>()

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"])
}
}
Original file line number Diff line number Diff line change
@@ -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<String, Any>).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<String, Any>) -> Triple<String, String, MutableMap<String, Any>> =
{ 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<String, Any> = java.util.Collections.unmodifiableMap(
mapOf("type" to "keyword")
)
val node = mutableMapOf<String, Any>("field1" to unmodifiableProps)
val flattenPaths = mutableMapOf<String, MutableMap<String, Any>>()

// 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<String, Any>(
"empty_field" to java.util.Collections.emptyMap<String, Any>()
)
val flattenPaths = mutableMapOf<String, MutableMap<String, Any>>()

// 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<String, Any>(
"field1" to mutableMapOf<String, Any>("type" to "keyword")
)
val flattenPaths = mutableMapOf<String, MutableMap<String, Any>>()
val renamingLeafFn: (String, String, MutableMap<String, Any>) -> Triple<String, String, MutableMap<String, Any>> =
{ 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<String, Any>("type" to "text")
val innerProperties = mutableMapOf<String, Any>("inner" to innerField)
val outerField = mutableMapOf<String, Any>(DocLevelMonitorQueries.PROPERTIES to innerProperties)
val node = mutableMapOf<String, Any>("outer" to outerField)
val flattenPaths = mutableMapOf<String, MutableMap<String, Any>>()

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<String, Any> = java.util.Collections.unmodifiableMap(
mapOf("type" to "text")
)
val innerProperties: Map<String, Any> = java.util.Collections.unmodifiableMap(
mapOf("inner" to innerField)
)
val outerField: Map<String, Any> = java.util.Collections.unmodifiableMap(
mapOf(DocLevelMonitorQueries.PROPERTIES to innerProperties)
)
val node = mutableMapOf<String, Any>("outer" to outerField)
val flattenPaths = mutableMapOf<String, MutableMap<String, Any>>()

// 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")
)
}
}
Loading