Backport removal of runBlocking and Stats API fix

alerting/src/main/kotlin/org/opensearch/alerting/MonitorRunner.kt

@@ -31,7 +31,6 @@ import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.launch
-import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.withContext
 import org.apache.logging.log4j.LogManager
 import org.opensearch.action.bulk.BackoffPolicy
@@ -41,6 +40,7 @@ import org.opensearch.alerting.core.JobRunner
 import org.opensearch.alerting.core.model.ScheduledJob
 import org.opensearch.alerting.elasticapi.InjectorContextElement
 import org.opensearch.alerting.elasticapi.retry
+import org.opensearch.alerting.elasticapi.withClosableContext
 import org.opensearch.alerting.model.ActionRunResult
 import org.opensearch.alerting.model.Alert
 import org.opensearch.alerting.model.AlertingConfigAccessor
@@ -295,7 +295,7 @@ object MonitorRunner : JobRunner, CoroutineScope, AbstractLifecycleComponent() {
             return monitorResult.copy(error = e)
         }
         if (!isADMonitor(monitor)) {
-            runBlocking(InjectorContextElement(monitor.id, settings, threadPool.threadContext, roles)) {
+            withClosableContext(InjectorContextElement(monitor.id, settings, threadPool.threadContext, roles)) {
                 monitorResult = monitorResult.copy(inputResults = inputService.collectInputResults(monitor, periodStart, periodEnd))
             }
         } else {
@@ -384,7 +384,7 @@ object MonitorRunner : JobRunner, CoroutineScope, AbstractLifecycleComponent() {
             //  If a setting is imposed that limits buckets that can be processed for Bucket-Level Monitors, we'd need to iterate over
             //  the buckets until we hit that threshold. In that case, we'd want to exit the execution without creating any alerts since the
             //  buckets we iterate over before hitting the limit is not deterministic. Is there a better way to fail faster in this case?
-            runBlocking(InjectorContextElement(monitor.id, settings, threadPool.threadContext, roles)) {
+            withClosableContext(InjectorContextElement(monitor.id, settings, threadPool.threadContext, roles)) {
                 // Storing the first page of results in the case of pagination input results to prevent empty results
                 // in the final output of monitorResult which occurs when all pages have been exhausted.
                 // If it's favorable to return the last page, will need to check how to accomplish that with multiple aggregation paths

alerting/src/test/kotlin/org/opensearch/alerting/AlertingRestTestCase.kt

@@ -90,6 +90,9 @@ abstract class AlertingRestTestCase : ODFERestTestCase() {
     protected val numberOfNodes = System.getProperty("cluster.number_of_nodes", "1")!!.toInt()
     protected val isMultiNode = numberOfNodes > 1
 
+    protected val statsResponseOpendistroSweeperEnabledField = "opendistro.scheduled_jobs.enabled"
+    protected val statsResponseOpenSearchSweeperEnabledField = "plugins.scheduled_jobs.enabled"
+
     override fun xContentRegistry(): NamedXContentRegistry {
         return NamedXContentRegistry(
             mutableListOf(
@@ -845,6 +848,26 @@ abstract class AlertingRestTestCase : ODFERestTestCase() {
         client().performRequest(request)
     }
 
+    fun createIndexRoleWithDocLevelSecurity(name: String, index: String, dlsQuery: String) {
+        val request = Request("PUT", "/_plugins/_security/api/roles/$name")
+        val entity = """
+            {
+              "cluster_permissions": [],
+              "index_permissions": [{
+                "index_patterns": [
+                  "$index"
+                ],
+                "dls": "$dlsQuery",
+                "allowed_actions": [
+                  "read"
+                ]
+              }]
+            }
+        """.trimIndent()
+        request.setJsonEntity(entity)
+        client().performRequest(request)
+    }
+
     fun createUserRolesMapping(role: String, users: Array<String>) {
         val request = Request("PUT", "/_plugins/_security/api/rolesmapping/$role")
         val usersStr = users.joinToString { it -> "\"$it\"" }
@@ -876,6 +899,19 @@ abstract class AlertingRestTestCase : ODFERestTestCase() {
         createUserRolesMapping(role, arrayOf(user))
     }
 
+    fun createUserWithDocLevelSecurityTestData(
+        user: String,
+        index: String,
+        role: String,
+        backendRole: String,
+        dlsQuery: String
+    ) {
+        createUser(user, user, arrayOf(backendRole))
+        createTestIndex(index)
+        createIndexRoleWithDocLevelSecurity(role, index, dlsQuery)
+        createUserRolesMapping(role, arrayOf(user))
+    }
+
     companion object {
         internal interface IProxy {
             val version: String?

alerting/src/test/kotlin/org/opensearch/alerting/MonitorRunnerIT.kt

@@ -1011,7 +1011,6 @@ class MonitorRunnerIT : AlertingRestTestCase() {
         val monitor = createMonitor(randomBucketLevelMonitor(inputs = listOf(input), enabled = false, triggers = listOf(trigger)))
         val response = executeMonitor(monitor.id, params = DRYRUN_MONITOR)
         val output = entityAsMap(response)
-        // print("Output is: $output")
 
         assertEquals(monitor.name, output["monitor_name"])
         @Suppress("UNCHECKED_CAST")

alerting/src/test/kotlin/org/opensearch/alerting/resthandler/MonitorRestApiIT.kt

@@ -587,7 +587,9 @@ class MonitorRestApiIT : AlertingRestTestCase() {
         assertTrue("Expected $expectedAcknowledgedCount alerts to be acknowledged successfully.", acknowledgedAlerts.size == expectedAcknowledgedCount)
 
         val acknowledgedAlertsList = acknowledgedAlerts.toString()
-        alertsToAcknowledge.forEach { alert -> assertTrue("Alert with ID ${alert.id} not found in failed list.", acknowledgedAlertsList.contains(alert.id)) }
+        alertsToAcknowledge.forEach { alert ->
+            assertTrue("Alert with ID ${alert.id} not found in failed list.", acknowledgedAlertsList.contains(alert.id))
+        }
 
         val failedResponse = responseMap["failed"] as List<String>
         assertTrue("Expected 0 alerts to fail acknowledgment.", failedResponse.isEmpty())
@@ -615,15 +617,23 @@ class MonitorRestApiIT : AlertingRestTestCase() {
         assertTrue("Expected $expectedAcknowledgedCount alerts to be acknowledged successfully.", acknowledgedAlerts.size == expectedAcknowledgedCount)
 
         val acknowledgedAlertsList = acknowledgedAlerts.toString()
-        alertsGroup2.forEach { alert -> assertTrue("Alert with ID ${alert.id} not found in failed list.", acknowledgedAlertsList.contains(alert.id)) }
-        alertsGroup1.forEach { alert -> assertFalse("Alert with ID ${alert.id} found in failed list.", acknowledgedAlertsList.contains(alert.id)) }
+        alertsGroup2.forEach { alert ->
+            assertTrue("Alert with ID ${alert.id} not found in failed list.", acknowledgedAlertsList.contains(alert.id))
+        }
+        alertsGroup1.forEach { alert ->
+            assertFalse("Alert with ID ${alert.id} found in failed list.", acknowledgedAlertsList.contains(alert.id))
+        }
 
         val failedResponse = responseMap["failed"] as List<String>
         assertTrue("Expected ${alertsGroup1.size} alerts to fail acknowledgment.", failedResponse.size == alertsGroup1.size)
 
         val failedResponseList = failedResponse.toString()
-        alertsGroup1.forEach { alert -> assertTrue("Alert with ID ${alert.id} not found in failed list.", failedResponseList.contains(alert.id)) }
-        alertsGroup2.forEach { alert -> assertFalse("Alert with ID ${alert.id} found in failed list.", failedResponseList.contains(alert.id)) }
+        alertsGroup1.forEach { alert ->
+            assertTrue("Alert with ID ${alert.id} not found in failed list.", failedResponseList.contains(alert.id))
+        }
+        alertsGroup2.forEach { alert ->
+            assertFalse("Alert with ID ${alert.id} found in failed list.", failedResponseList.contains(alert.id))
+        }
     }
 
     @Throws(Exception::class)
@@ -883,8 +893,7 @@ class MonitorRestApiIT : AlertingRestTestCase() {
         disableScheduledJob()
 
         val responseMap = getAlertingStats()
-        // assertEquals("Cluster name is incorrect", responseMap["cluster_name"], "alerting_integTestCluster")
-        assertEquals("Scheduled job is not enabled", false, responseMap[ScheduledJobSettings.SWEEPER_ENABLED.key])
+        assertAlertingStatsSweeperEnabled(responseMap, false)
         assertEquals("Scheduled job index exists but there are no scheduled jobs.", false, responseMap["scheduled_job_index_exists"])
         val _nodes = responseMap["_nodes"] as Map<String, Int>
         assertEquals("Incorrect number of nodes", numberOfNodes, _nodes["total"])
@@ -897,8 +906,7 @@ class MonitorRestApiIT : AlertingRestTestCase() {
         enableScheduledJob()
 
         val responseMap = getAlertingStats()
-        // assertEquals("Cluster name is incorrect", responseMap["cluster_name"], "alerting_integTestCluster")
-        assertEquals("Scheduled job is not enabled", true, responseMap[ScheduledJobSettings.SWEEPER_ENABLED.key])
+        assertAlertingStatsSweeperEnabled(responseMap, true)
         assertEquals("Scheduled job index exists but there are no scheduled jobs.", false, responseMap["scheduled_job_index_exists"])
         val _nodes = responseMap["_nodes"] as Map<String, Int>
         assertEquals("Incorrect number of nodes", numberOfNodes, _nodes["total"])
@@ -912,8 +920,7 @@ class MonitorRestApiIT : AlertingRestTestCase() {
         createRandomMonitor(refresh = true)
 
         val responseMap = getAlertingStats()
-        // assertEquals("Cluster name is incorrect", responseMap["cluster_name"], "alerting_integTestCluster")
-        assertEquals("Scheduled job is not enabled", true, responseMap[ScheduledJobSettings.SWEEPER_ENABLED.key])
+        assertAlertingStatsSweeperEnabled(responseMap, true)
         assertEquals("Scheduled job index does not exist", true, responseMap["scheduled_job_index_exists"])
         assertEquals("Scheduled job index is not yellow", "yellow", responseMap["scheduled_job_index_status"])
         assertEquals("Nodes are not on schedule", numberOfNodes, responseMap["nodes_on_schedule"])
@@ -943,8 +950,7 @@ class MonitorRestApiIT : AlertingRestTestCase() {
         createRandomMonitor(refresh = true)
 
         val responseMap = getAlertingStats("/jobs_info")
-        // assertEquals("Cluster name is incorrect", responseMap["cluster_name"], "alerting_integTestCluster")
-        assertEquals("Scheduled job is not enabled", true, responseMap[ScheduledJobSettings.SWEEPER_ENABLED.key])
+        assertAlertingStatsSweeperEnabled(responseMap, true)
         assertEquals("Scheduled job index does not exist", true, responseMap["scheduled_job_index_exists"])
         assertEquals("Scheduled job index is not yellow", "yellow", responseMap["scheduled_job_index_status"])
         assertEquals("Nodes not on schedule", numberOfNodes, responseMap["nodes_on_schedule"])
@@ -1045,4 +1051,17 @@ class MonitorRestApiIT : AlertingRestTestCase() {
             assertEquals("Unexpected status", RestStatus.BAD_REQUEST, e.response.restStatus())
         }
     }
+
+    private fun assertAlertingStatsSweeperEnabled(alertingStatsResponse: Map<String, Any>, expected: Boolean) {
+        assertEquals(
+            "Legacy scheduled job enabled field is not set to $expected",
+            expected,
+            alertingStatsResponse[statsResponseOpendistroSweeperEnabledField]
+        )
+        assertEquals(
+            "Scheduled job is not ${if (expected) "enabled" else "disabled"}",
+            expected,
+            alertingStatsResponse[statsResponseOpenSearchSweeperEnabledField]
+        )
+    }
 }

alerting/src/test/kotlin/org/opensearch/alerting/resthandler/SecureMonitorRestApiIT.kt

@@ -19,12 +19,15 @@ import org.opensearch.alerting.ALERTING_BASE_URI
 import org.opensearch.alerting.ALWAYS_RUN
 import org.opensearch.alerting.AlertingRestTestCase
 import org.opensearch.alerting.DRYRUN_MONITOR
+import org.opensearch.alerting.aggregation.bucketselectorext.BucketSelectorExtAggregationBuilder
 import org.opensearch.alerting.assertUserNull
 import org.opensearch.alerting.core.model.SearchInput
 import org.opensearch.alerting.makeRequest
 import org.opensearch.alerting.model.Alert
 import org.opensearch.alerting.randomAction
 import org.opensearch.alerting.randomAlert
+import org.opensearch.alerting.randomBucketLevelMonitor
+import org.opensearch.alerting.randomBucketLevelTrigger
 import org.opensearch.alerting.randomQueryLevelMonitor
 import org.opensearch.alerting.randomQueryLevelTrigger
 import org.opensearch.alerting.randomTemplateScript
@@ -39,6 +42,9 @@ import org.opensearch.commons.authuser.User
 import org.opensearch.commons.rest.SecureRestClientBuilder
 import org.opensearch.index.query.QueryBuilders
 import org.opensearch.rest.RestStatus
+import org.opensearch.script.Script
+import org.opensearch.search.aggregations.bucket.composite.CompositeAggregationBuilder
+import org.opensearch.search.aggregations.bucket.composite.TermsValuesSourceBuilder
 import org.opensearch.search.builder.SearchSourceBuilder
 import org.opensearch.test.junit.annotations.TestLogging
 
@@ -521,4 +527,129 @@ class SecureMonitorRestApiIT : AlertingRestTestCase() {
             deleteRoleMapping("alerting_full_access")
         }
     }
+
+    fun `test execute query-level monitor with user having partial index permissions`() {
+        if (!securityEnabled()) return
+
+        val testIndex = "hr_data"
+        createUserWithDocLevelSecurityTestData(
+            user,
+            testIndex,
+            "hr_role",
+            "HR",
+            "{\"term\": { \"accessible\": true}}"
+        )
+        createUserRolesMapping("alerting_full_access", arrayOf(user))
+
+        // Add a doc that is accessible to the user
+        indexDoc(
+            testIndex, "1",
+            """
+            {
+              "test_field": "a",
+              "accessible": true
+            }
+            """.trimIndent()
+        )
+
+        // Add a second doc that is not accesible to the user
+        indexDoc(
+            testIndex, "2",
+            """
+            {
+              "test_field": "b",
+              "accessible": false
+            }
+            """.trimIndent()
+        )
+
+        val input = SearchInput(indices = listOf(testIndex), query = SearchSourceBuilder().query(QueryBuilders.matchAllQuery()))
+        val triggerScript = """
+            // make sure there is exactly one hit
+            return ctx.results[0].hits.hits.size() == 1
+        """.trimIndent()
+
+        val trigger = randomQueryLevelTrigger(condition = Script(triggerScript))
+        val monitor = randomQueryLevelMonitor(inputs = listOf(input), triggers = listOf(trigger))
+
+        try {
+            executeMonitor(monitor.id, params = DRYRUN_MONITOR)
+            val alerts = searchAlerts(monitor)
+            assertEquals("Incorrect number of alerts", 1, alerts.size)
+        } finally {
+            deleteRoleMapping("hr_role")
+            deleteRole("hr_role")
+            deleteRoleMapping("alerting_full_access")
+        }
+    }
+
+    fun `test execute bucket-level monitor with user having partial index permissions`() {
+        if (!securityEnabled()) return
+
+        val testIndex = "hr_data"
+        createUserWithDocLevelSecurityTestData(
+            user,
+            testIndex,
+            "hr_role",
+            "HR",
+            "{\"term\": { \"accessible\": true}}"
+        )
+        createUserRolesMapping("alerting_full_access", arrayOf(user))
+
+        // Add a doc that is accessible to the user
+        indexDoc(
+            testIndex, "1",
+            """
+            {
+              "test_field": "a",
+              "accessible": true
+            }
+            """.trimIndent()
+        )
+
+        // Add a second doc that is not accesible to the user
+        indexDoc(
+            testIndex, "2",
+            """
+            {
+              "test_field": "b",
+              "accessible": false
+            }
+            """.trimIndent()
+        )
+
+        val compositeSources = listOf(
+            TermsValuesSourceBuilder("test_field").field("test_field")
+        )
+        val compositeAgg = CompositeAggregationBuilder("composite_agg", compositeSources)
+        val input = SearchInput(
+            indices = listOf(testIndex),
+            query = SearchSourceBuilder().size(0).query(QueryBuilders.matchAllQuery()).aggregation(compositeAgg)
+        )
+        val triggerScript = """
+            params.docCount > 0
+        """.trimIndent()
+
+        var trigger = randomBucketLevelTrigger()
+        trigger = trigger.copy(
+            bucketSelector = BucketSelectorExtAggregationBuilder(
+                name = trigger.id,
+                bucketsPathsMap = mapOf("docCount" to "_count"),
+                script = Script(triggerScript),
+                parentBucketPath = "composite_agg",
+                filter = null
+            )
+        )
+        val monitor = createMonitor(randomBucketLevelMonitor(inputs = listOf(input), enabled = false, triggers = listOf(trigger)))
+
+        try {
+            executeMonitor(monitor.id, params = DRYRUN_MONITOR)
+            val alerts = searchAlerts(monitor)
+            assertEquals("Incorrect number of alerts", 1, alerts.size)
+        } finally {
+            deleteRoleMapping("hr_role")
+            deleteRole("hr_role")
+            deleteRoleMapping("alerting_full_access")
+        }
+    }
 }

core/src/main/kotlin/org/opensearch/alerting/core/action/node/ScheduledJobsStatsResponse.kt

@@ -28,6 +28,7 @@ package org.opensearch.alerting.core.action.node
 
 import org.opensearch.action.FailedNodeException
 import org.opensearch.action.support.nodes.BaseNodesResponse
+import org.opensearch.alerting.core.settings.LegacyOpenDistroScheduledJobSettings
 import org.opensearch.alerting.core.settings.ScheduledJobSettings
 import org.opensearch.cluster.ClusterName
 import org.opensearch.cluster.health.ClusterIndexHealth
@@ -77,6 +78,7 @@ class ScheduledJobsStatsResponse : BaseNodesResponse<ScheduledJobStats>, ToXCont
     }
 
     override fun toXContent(builder: XContentBuilder, params: ToXContent.Params): XContentBuilder {
+        builder.field(LegacyOpenDistroScheduledJobSettings.SWEEPER_ENABLED.key, scheduledJobEnabled)
         builder.field(ScheduledJobSettings.SWEEPER_ENABLED.key, scheduledJobEnabled)
         builder.field("scheduled_job_index_exists", indexExists)
         builder.field("scheduled_job_index_status", indexHealth?.status?.name?.toLowerCase())

core/src/main/kotlin/org/opensearch/alerting/elasticapi/ElasticExtensions.kt

@@ -26,8 +26,10 @@
 
 package org.opensearch.alerting.elasticapi
 
+import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.ThreadContextElement
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.withContext
 import org.apache.logging.log4j.Logger
 import org.opensearch.OpenSearchException
 import org.opensearch.action.ActionListener
@@ -218,3 +220,14 @@ class InjectorContextElement(id: String, settings: Settings, threadContext: Thre
         rolesInjectorHelper.close()
     }
 }
+
+suspend fun <T> withClosableContext(
+    context: InjectorContextElement,
+    block: suspend CoroutineScope.() -> T
+): T {
+    try {
+        return withContext(context) { block() }
+    } finally {
+        context.rolesInjectorHelper.close()
+    }
+}