Percolate query optimization: Fetch fields mentioned in queries instead of entire doc and batch percolate query by heap-based threshold

alerting/build.gradle

@@ -424,9 +424,4 @@ run {
     useCluster testClusters.integTest
 }
 
-// Only apply jacoco test coverage if we are running a local single node cluster
-if (!usingRemoteCluster && !usingMultiNode) {
-    apply from: '../build-tools/opensearchplugin-coverage.gradle'
-}
-
 apply from: '../build-tools/pkgbuild.gradle'

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

@@ -100,6 +100,7 @@ import org.opensearch.core.xcontent.XContentParser
 import org.opensearch.env.Environment
 import org.opensearch.env.NodeEnvironment
 import org.opensearch.index.IndexModule
+import org.opensearch.monitor.jvm.JvmStats
 import org.opensearch.painless.spi.Allowlist
 import org.opensearch.painless.spi.AllowlistLoader
 import org.opensearch.painless.spi.PainlessExtension
@@ -263,6 +264,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
             .registerTriggerService(TriggerService(scriptService))
             .registerAlertService(AlertService(client, xContentRegistry, alertIndices))
             .registerDocLevelMonitorQueries(DocLevelMonitorQueries(client, clusterService))
+            .registerJvmStats(JvmStats.jvmStats())
             .registerWorkflowService(WorkflowService(client, xContentRegistry))
             .registerConsumers()
             .registerDestinationSettings()
@@ -320,6 +322,8 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
             AlertingSettings.ALERT_HISTORY_MAX_DOCS,
             AlertingSettings.ALERT_HISTORY_RETENTION_PERIOD,
             AlertingSettings.ALERTING_MAX_MONITORS,
+            AlertingSettings.PERCOLATE_QUERY_DOCS_SIZE_MEMORY_PERCENTAGE_LIMIT,
+            AlertingSettings.PERCOLATE_QUERY_MAX_NUM_DOCS_IN_MEMORY,
             AlertingSettings.REQUEST_TIMEOUT,
             AlertingSettings.MAX_ACTION_THROTTLE_VALUE,
             AlertingSettings.FILTER_BY_BACKEND_ROLES,
@@ -340,6 +344,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
             LegacyOpenDistroAlertingSettings.REQUEST_TIMEOUT,
             LegacyOpenDistroAlertingSettings.MAX_ACTION_THROTTLE_VALUE,
             LegacyOpenDistroAlertingSettings.FILTER_BY_BACKEND_ROLES,
+            AlertingSettings.DOC_LEVEL_MONITOR_FETCH_ONLY_QUERY_FIELDS_ENABLED,
             DestinationSettings.EMAIL_USERNAME,
             DestinationSettings.EMAIL_PASSWORD,
             DestinationSettings.ALLOW_LIST,

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

@@ -52,6 +52,7 @@ import org.opensearch.search.builder.SearchSourceBuilder
 import java.time.Instant
 import java.util.UUID
 
+// TODO raise PR for bucket level monitor optimization also. dont miss
 object BucketLevelMonitorRunner : MonitorRunner() {
     private val logger = LogManager.getLogger(javaClass)
 

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

@@ -18,6 +18,7 @@ import org.opensearch.alerting.opensearchapi.suspendUntil
 import org.opensearch.alerting.settings.AlertingSettings
 import org.opensearch.alerting.util.AggregationQueryRewriter
 import org.opensearch.alerting.util.CrossClusterMonitorUtils
+import org.opensearch.alerting.util.IndexUtils
 import org.opensearch.alerting.util.addUserBackendRolesFilter
 import org.opensearch.alerting.util.clusterMetricsMonitorHelpers.executeTransportAction
 import org.opensearch.alerting.util.clusterMetricsMonitorHelpers.toMap
@@ -112,6 +113,7 @@ class InputService(
                         val searchRequest = SearchRequest()
                             .indices(*indexes.toTypedArray())
                             .preference(Preference.PRIMARY_FIRST.type())
+                            .indices(*resolveIndices(monitor, input.indices).toTypedArray())
                         XContentType.JSON.xContent().createParser(xContentRegistry, LoggingDeprecationHandler.INSTANCE, searchSource).use {
                             searchRequest.source(SearchSourceBuilder.fromXContent(it))
                         }
@@ -232,7 +234,6 @@ class InputService(
 
             val searchRequest = SearchRequest()
                 .indices(*input.indices.toTypedArray())
-                .preference(Preference.PRIMARY_FIRST.type())
             XContentType.JSON.xContent().createParser(xContentRegistry, LoggingDeprecationHandler.INSTANCE, searchSource).use {
                 searchRequest.source(SearchSourceBuilder.fromXContent(it))
             }
@@ -259,4 +260,29 @@ class InputService(
             InputRunResults(emptyList(), e)
         }
     }
+
+    /** accepts list of index names and returns back a list where :
+     * i. if it's a concrete index or an index pattern - no changes is made
+     * ii. if it's a data stream or an alias - we return the active write index
+     */
+    private fun resolveIndices(
+        monitor: Monitor,
+        indices: List<String>,
+    ): List<String> {
+        val resolvedIndices = mutableListOf<String>()
+        for (it in indices) {
+            if (IndexUtils.isAlias(it, clusterService.state()) || IndexUtils.isDataStream(it, clusterService.state())) {
+                val writeIndex = IndexUtils.getWriteIndex(it, clusterService.state())
+                if (writeIndex == null) {
+                    logger.error("Monitor $monitor.id: Write Index not found for $it")
+                    continue
+                }
+                resolvedIndices.add(writeIndex)
+                // TODO add edge case where periodStart < writeIndex's creationTime
+            } else {
+                resolvedIndices.add(it)
+            }
+        }
+        return resolvedIndices
+    }
 }

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

@@ -125,6 +125,7 @@ object MonitorMetadataService :
                 throw OpenSearchStatusException(failureReason, RestStatus.INTERNAL_SERVER_ERROR)
             }
         } catch (e: Exception) {
+            log.error("Failed to upsert metadata", e)
             throw AlertingException.wrap(e)
         }
     }

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

@@ -18,6 +18,7 @@ import org.opensearch.cluster.service.ClusterService
 import org.opensearch.common.settings.Settings
 import org.opensearch.common.unit.TimeValue
 import org.opensearch.core.xcontent.NamedXContentRegistry
+import org.opensearch.monitor.jvm.JvmStats
 import org.opensearch.script.ScriptService
 import org.opensearch.threadpool.ThreadPool
 
@@ -36,6 +37,7 @@ data class MonitorRunnerExecutionContext(
     var alertService: AlertService? = null,
     var docLevelMonitorQueries: DocLevelMonitorQueries? = null,
     var workflowService: WorkflowService? = null,
+    var jvmStats: JvmStats? = null,
 
     @Volatile var retryPolicy: BackoffPolicy? = null,
     @Volatile var moveAlertsRetryPolicy: BackoffPolicy? = null,

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

@@ -50,6 +50,7 @@ import org.opensearch.commons.alerting.model.action.Action
 import org.opensearch.commons.alerting.util.isBucketLevelMonitor
 import org.opensearch.core.action.ActionListener
 import org.opensearch.core.xcontent.NamedXContentRegistry
+import org.opensearch.monitor.jvm.JvmStats
 import org.opensearch.script.Script
 import org.opensearch.script.ScriptService
 import org.opensearch.script.TemplateScript
@@ -134,6 +135,11 @@ object MonitorRunnerService : JobRunner, CoroutineScope, AbstractLifecycleCompon
         return this
     }
 
+    fun registerJvmStats(jvmStats: JvmStats): MonitorRunnerService {
+        this.monitorCtx.jvmStats = jvmStats
+        return this
+    }
+
     // Must be called after registerClusterService and registerSettings in AlertingPlugin
     fun registerConsumers(): MonitorRunnerService {
         monitorCtx.retryPolicy = BackoffPolicy.constantBackoff(
@@ -258,11 +264,19 @@ object MonitorRunnerService : JobRunner, CoroutineScope, AbstractLifecycleCompon
         when (job) {
             is Workflow -> {
                 launch {
+                    logger.debug(
+                        "PERF_DEBUG: executing workflow ${job.id} on node " +
+                            monitorCtx.clusterService!!.state().nodes().localNode.id
+                    )
                     runJob(job, periodStart, periodEnd, false)
                 }
             }
             is Monitor -> {
                 launch {
+                    logger.debug(
+                        "PERF_DEBUG: executing ${job.monitorType} ${job.id} on node " +
+                            monitorCtx.clusterService!!.state().nodes().localNode.id
+                    )
                     runJob(job, periodStart, periodEnd, false)
                 }
             }

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

@@ -0,0 +1,18 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.alerting.model
+
+import org.opensearch.commons.alerting.model.DocLevelQuery
+
+data class IndexExecutionContext(
+    val queries: List<DocLevelQuery>,
+    val lastRunContext: MutableMap<String, Any>,
+    val updatedLastRunContext: MutableMap<String, Any>,
+    val indexName: String,
+    val concreteIndexName: String,
+    val conflictingFields: List<String>,
+    val docIds: List<String>? = null
+)

alerting/src/main/kotlin/org/opensearch/alerting/settings/AlertingSettings.kt

@@ -25,6 +25,40 @@ class AlertingSettings {
             Setting.Property.NodeScope, Setting.Property.Dynamic
         )
 
+        /** Defines the threshold of the docs accumulated in memory to query against percolate query index in document
+         * level monitor execution. The docs are being collected from searching on shards of indices mentioned in the
+         * monitor input indices field.
+         */
+        val PERCOLATE_QUERY_DOCS_SIZE_MEMORY_PERCENTAGE_LIMIT = Setting.intSetting(
+            "plugins.alerting.monitor.percolate_query_docs_size_memory_percentage_limit",
+            10,
+            0,
+            100,
+            Setting.Property.NodeScope, Setting.Property.Dynamic
+        )
+
+        /** Defines the threshold of the maximum number of docs accumulated in memory to query against percolate query index in document
+         * level monitor execution. The docs are being collected from searching on shards of indices mentioned in the
+         * monitor input indices field. When the number of in-memory docs reaches or exceeds threshold we immediately perform percolate
+         * query with the current set of docs and clear the cache and repeat the process till we have queried all indices in current
+         * execution
+         */
+        val PERCOLATE_QUERY_MAX_NUM_DOCS_IN_MEMORY = Setting.intSetting(
+            "plugins.alerting.monitor.percolate_query_max_num_docs_in_memory",
+            10000, 1000,
+            Setting.Property.NodeScope, Setting.Property.Dynamic
+        )
+
+        /**
+         * Boolean setting to enable/disable optimizing doc level monitors by fetchign only fields mentioned in queries.
+         * Enabled by default. If disabled, will fetch entire source of documents while fetch data from shards.
+         */
+        val DOC_LEVEL_MONITOR_FETCH_ONLY_QUERY_FIELDS_ENABLED = Setting.boolSetting(
+            "plugins.alerting.monitor.doc_level_monitor_fetch_only_query_fields_enabled",
+            true,
+            Setting.Property.NodeScope, Setting.Property.Dynamic
+        )
+
         val INPUT_TIMEOUT = Setting.positiveTimeSetting(
             "plugins.alerting.input_timeout",
             LegacyOpenDistroAlertingSettings.INPUT_TIMEOUT,

alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportIndexMonitorAction.kt

@@ -189,6 +189,18 @@ class TransportIndexMonitorAction @Inject constructor(
             else (it as DocLevelMonitorInput).indices
             indices.addAll(inputIndices)
         }
+        if (
+            indices.size == 1 && (
+                IndexUtils.isAlias(indices[0], clusterService.state()) ||
+                    IndexUtils.isDataStream(indices[0], clusterService.state())
+                )
+        ) {
+            val metadata = clusterService.state().metadata.indicesLookup[indices[0]]?.writeIndex
+            if (metadata != null) {
+                indices.removeAt(0)
+                indices.add(metadata.index.name)
+            }
+        }
         val searchRequest = SearchRequest().indices(*indices.toTypedArray())
             .source(SearchSourceBuilder.searchSource().size(1).query(QueryBuilders.matchAllQuery()))
         client.search(

alerting/src/main/kotlin/org/opensearch/alerting/util/DocLevelMonitorQueries.kt

@@ -395,7 +395,7 @@ class DocLevelMonitorQueries(private val client: Client, private val clusterServ
             }
             bulkResponse.forEach { bulkItemResponse ->
                 if (bulkItemResponse.isFailed) {
-                    log.debug(bulkItemResponse.failureMessage)
+                    log.error("Failed to index doc level query for monitor $monitorId due to" + bulkItemResponse.failureMessage)
                 }
             }
         }

alerting/src/main/kotlin/org/opensearch/alerting/workflow/CompositeWorkflowRunner.kt

@@ -60,7 +60,10 @@ object CompositeWorkflowRunner : WorkflowRunner() {
         val isTempWorkflow = dryRun || workflow.id == Workflow.NO_ID
 
         val executionId = generateExecutionId(isTempWorkflow, workflow)
-
+        logger.debug(
+            "Workflow ${workflow.id} execution began at $workflowExecutionStartTime" +
+                " on node ${monitorCtx.clusterService!!.localNode().id}"
+        )
         val (workflowMetadata, _) = WorkflowMetadataService.getOrCreateWorkflowMetadata(
             workflow = workflow,
             skipIndex = isTempWorkflow,
@@ -227,6 +230,7 @@ object CompositeWorkflowRunner : WorkflowRunner() {
             )
         }
         workflowRunResult.executionEndTime = Instant.now()
+        logger.debug("Workflow ${workflow.id} execution completed at $workflowRunResult.executionEndTime")
         return workflowRunResult
     }
 

alerting/src/main/resources/org/opensearch/alerting/alerts/finding_mapping.json

@@ -49,6 +49,9 @@
         },
         "fields": {
           "type": "text"
+        },
+        "query_field_names": {
+          "type": "keyword"
         }
       }
     },

core/build.gradle

@@ -19,7 +19,7 @@ dependencies {
         exclude group: 'com.google.guava'
     }
     implementation 'com.google.guava:guava:32.0.1-jre'
-    api "org.opensearch:common-utils:${common_utils_version}@jar"
+    api files("/Users/snistala/Documents/opensearch/common-utils/build/libs/common-utils-3.0.0.0-SNAPSHOT.jar")
     implementation 'commons-validator:commons-validator:1.7'
 
     testImplementation "org.opensearch.test:framework:${opensearch_version}"

core/src/main/kotlin/org/opensearch/alerting/core/JobSweeper.kt

@@ -462,11 +462,13 @@ class JobSweeper(
 
     private fun isOwningNode(shardId: ShardId, jobId: JobId): Boolean {
         val localNodeId = clusterService.localNode().id
-        val shardNodeIds = clusterService.state().routingTable.shardRoutingTable(shardId)
+        val shardRoutingTable = clusterService.state().routingTable.shardRoutingTable(shardId)
+        val shardNodeIds = shardRoutingTable
             .filter { it.active() }
             .map { it.currentNodeId() }
         val shardNodes = ShardNodes(localNodeId, shardNodeIds)
-        return shardNodes.isOwningNode(jobId)
+        val owningNode = shardNodes.isOwningNode(jobId)
+        return owningNode
     }
 }