Add connection to triggers for doc level alerting #316
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -30,6 +30,7 @@ import org.opensearch.alerting.model.Alert | |
| import org.opensearch.alerting.model.AlertingConfigAccessor | ||
| import org.opensearch.alerting.model.BucketLevelTrigger | ||
| import org.opensearch.alerting.model.BucketLevelTriggerRunResult | ||
| import org.opensearch.alerting.model.DocumentLevelTrigger | ||
| import org.opensearch.alerting.model.Finding | ||
| import org.opensearch.alerting.model.InputRunResults | ||
| import org.opensearch.alerting.model.Monitor | ||
|
|
@@ -48,6 +49,7 @@ import org.opensearch.alerting.model.destination.DestinationContextFactory | |
| import org.opensearch.alerting.model.docLevelInput.DocLevelMonitorInput | ||
| import org.opensearch.alerting.model.docLevelInput.DocLevelQuery | ||
| import org.opensearch.alerting.script.BucketLevelTriggerExecutionContext | ||
| import org.opensearch.alerting.script.DocumentLevelTriggerExecutionContext | ||
| import org.opensearch.alerting.script.QueryLevelTriggerExecutionContext | ||
| import org.opensearch.alerting.script.TriggerExecutionContext | ||
| import org.opensearch.alerting.settings.AlertingSettings.Companion.ALERT_BACKOFF_COUNT | ||
|
|
@@ -757,39 +759,119 @@ object MonitorRunner : JobRunner, CoroutineScope, AbstractLifecycleComponent() { | |
| return | ||
| } | ||
|
|
||
| val count: Int = lastRunContext["shards_count"] as Int | ||
|
There was a problem hiding this comment. How do we handle change in shard count between the runs? Are there edge scenarios which might need this handling. There was a problem hiding this comment. I created this github issue regarding this. Also this only happens when there is a reindexing operation, so there might still be issues during executions since the sequence numbers will be changed as well. |
||
| val updatedLastRunContext = lastRunContext.toMutableMap() | ||
AWSHurneyt marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| for (i: Int in 0 until count) { | ||
| val shard = i.toString() | ||
| val maxSeqNo: Long = getMaxSeqNo(index, shard) | ||
| updatedLastRunContext[shard] = maxSeqNo.toString() | ||
| } | ||
|
|
||
| val queryDocIds = mutableMapOf<DocLevelQuery, Set<String>>() | ||
|
There was a problem hiding this comment.
|
||
| val docsToQueries = mutableMapOf<String, MutableList<String>>() | ||
|
There was a problem hiding this comment.
|
||
| for (query in queries) { | ||
| val matchingDocIds = runForEachQuery(monitor, lastRunContext, updatedLastRunContext, index, query) | ||
| queryDocIds[query] = matchingDocIds | ||
| matchingDocIds.forEach { | ||
| if (docsToQueries.containsKey(it)) { | ||
| docsToQueries[it]?.add(query.id) | ||
| } else { | ||
| docsToQueries[it] = mutableListOf(query.id) | ||
| } | ||
| } | ||
|
There was a problem hiding this comment. Can we simplify this by passing a Supplier Function to There was a problem hiding this comment. By trying to add a supplier, the code started to seem more complex and I noticed we need to change the threads the plugin runs on and additionally add additional permissions for the plugin to parallelize on multiple threads in the cluster. We can do this as a separate PR |
||
| } | ||
|
|
||
| val queryIds = queries.map { it.id } | ||
|
|
||
| for (trigger in monitor.triggers) { | ||
| val triggerCtx = DocumentLevelTriggerExecutionContext(monitor, trigger as DocumentLevelTrigger) | ||
| val triggerResult = triggerService.runDocLevelTrigger(monitor, trigger, triggerCtx, docsToQueries, queryIds) | ||
|
|
||
| logger.info("trigger results") | ||
| logger.info(triggerResult.triggeredDocs.toString()) | ||
|
|
||
| queryDocIds.forEach { | ||
| val queryTriggeredDocs = it.value.intersect(triggerResult.triggeredDocs) | ||
| if (queryTriggeredDocs.isNotEmpty()) { | ||
| val findingId = createFindings(monitor, index, it.key, queryTriggeredDocs, trigger) | ||
| // TODO: check if need to create alert, if so create it and point it to FindingId | ||
| // TODO: run action as well, but this mat need to be throttled based on Mo's comment for bucket level alerting | ||
| } | ||
| } | ||
|
There was a problem hiding this comment. Abstract this out as a separate There was a problem hiding this comment. I have separated the function out for this. Also not sure if there is any benefit to consolidate all the findings in the end since we just want to run a single action based on each finding |
||
| } | ||
|
|
||
| // TODO: Check for race condition against the update monitor api | ||
| // This does the update at the end in case of errors and makes sure all the queries are executed | ||
| val updatedMonitor = monitor.copy(lastRunContext = updatedLastRunContext) | ||
| // note: update has to called in serial for shards of a given index. | ||
| // make sure this is just updated for the specific query or at the end of all the queries | ||
| updateMonitor(client, xContentRegistry, settings, updatedMonitor) | ||
|
There was a problem hiding this comment. What is there are race due to concurrent updates to monitor from user driven update vs last context updates? There was a problem hiding this comment. I will remove these comments since these have now been sorted. Yea there has been talks of having another index for additional metadata information and this can help with throttling. |
||
| } | ||
|
|
||
| private fun createFindings( | ||
| monitor: Monitor, | ||
| index: String, | ||
| docLevelQuery: DocLevelQuery, | ||
| matchingDocIds: Set<String>, | ||
| trigger: DocumentLevelTrigger | ||
| ): String { | ||
| val finding = Finding( | ||
| id = UUID.randomUUID().toString(), | ||
| relatedDocId = matchingDocIds.joinToString(","), | ||
| monitorId = monitor.id, | ||
| monitorName = monitor.name, | ||
| index = index, | ||
| queryId = docLevelQuery.id, | ||
| queryTags = docLevelQuery.tags, | ||
| severity = docLevelQuery.severity, | ||
| timestamp = Instant.now(), | ||
| triggerId = trigger.id, | ||
| triggerName = trigger.name | ||
| ) | ||
|
|
||
| val findingStr = finding.toXContent(XContentBuilder.builder(XContentType.JSON.xContent()), ToXContent.EMPTY_PARAMS).string() | ||
| // change this to debug. | ||
| logger.info("Findings: $findingStr") | ||
|
|
||
| // todo: below is all hardcoded, temp code and added only to test. replace this with proper Findings index lifecycle management. | ||
| val indexRequest = IndexRequest(".opensearch-alerting-findings") | ||
| .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE) | ||
| .source(findingStr, XContentType.JSON) | ||
|
|
||
| client.index(indexRequest).actionGet() | ||
| return finding.id | ||
| } | ||
|
|
||
| private suspend fun runForEachQuery( | ||
| monitor: Monitor, | ||
| lastRunContext: MutableMap<String, Any>, | ||
| updatedLastRunContext: MutableMap<String, Any>, | ||
| index: String, | ||
| query: DocLevelQuery | ||
| ): Set<String> { | ||
| val count: Int = lastRunContext["shards_count"] as Int | ||
| val matchingDocs = mutableSetOf<String>() | ||
| for (i: Int in 0 until count) { | ||
| val shard = i.toString() | ||
| try { | ||
| logger.info("Monitor execution for shard: $shard") | ||
|
|
||
| val maxSeqNo: Long = updatedLastRunContext[shard].toString().toLong() | ||
|
There was a problem hiding this comment. I like the idea that we are pre-caching the last run context per shard, as this allows uniform run-time across the shards since we its essential sequential run among the shards. However, I see the opportunity of creating an execution object which provides abstraction for all the information needed to run the queries per shard. Such as queries, lastRunContext, UpdatedContext, matchingDocs under a single wrapper. Think of it like the state object which you can pass to a runnable for asynchronous parallel executions. We will need this in very near future. This will help make the flow logic of this method stateless. There was a problem hiding this comment. I added a new object, but note that this object will be modified more later on as we optimize the flow and utilize the object. I didn't see any benefit to add the matchingDocs, so that was not added. |
||
| logger.info("MaxSeqNo of shard_$shard is $maxSeqNo") | ||
|
|
||
| val hits: SearchHits = searchShard(index, shard, lastRunContext[shard].toString().toLongOrNull(), maxSeqNo, query.query) | ||
|
There was a problem hiding this comment. handle current seq number less than current? Possible it also needs initial handling when contexts are created. There was a problem hiding this comment. Fail fast or return empty for cases where there are merges and the sequence numbers are messed up. This way next monitor execution will work correctly |
||
| logger.info("Search hits for shard_$shard is: ${hits.hits.size}") | ||
|
|
||
| if (hits.hits.isNotEmpty()) { | ||
| logger.info("found matches") | ||
| matchingDocs.addAll(getAllDocIds(hits)) | ||
| } | ||
| } catch (e: Exception) { | ||
| logger.info("Failed to run for shard $shard. Error: ${e.message}") | ||
| logger.debug("Failed to run for shard $shard", e) | ||
| } | ||
| } | ||
| return matchingDocs | ||
| } | ||
|
|
||
| // todo: add more validations. | ||
|
|
@@ -810,7 +892,7 @@ object MonitorRunner : JobRunner, CoroutineScope, AbstractLifecycleComponent() { | |
|
|
||
| private fun getShardsCount(index: String): Int { | ||
| val allShards: List<ShardRouting> = clusterService.state().routingTable().allShards(index) | ||
| return allShards.filter { it.primary() }.size | ||
| } | ||
|
|
||
| private fun createRunContext(index: String): HashMap<String, Any> { | ||
|
|
@@ -854,6 +936,9 @@ object MonitorRunner : JobRunner, CoroutineScope, AbstractLifecycleComponent() { | |
| } | ||
|
|
||
| private fun searchShard(index: String, shard: String, prevSeqNo: Long?, maxSeqNo: Long, query: String): SearchHits { | ||
| if (prevSeqNo?.equals(maxSeqNo) == true) { | ||
| return SearchHits.empty() | ||
| } | ||
| val boolQueryBuilder = BoolQueryBuilder() | ||
| boolQueryBuilder.filter(QueryBuilders.rangeQuery("_seq_no").gt(prevSeqNo).lte(maxSeqNo)) | ||
| boolQueryBuilder.must(QueryBuilders.queryStringQuery(query)) | ||
|
|
@@ -875,39 +960,7 @@ object MonitorRunner : JobRunner, CoroutineScope, AbstractLifecycleComponent() { | |
| return response.hits | ||
| } | ||
|
|
||
| private fun getAllDocIds(hits: SearchHits): List<String> { | ||
| return hits.map { hit -> hit.id } | ||
| } | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Do you think it's the time we break MonitorRunner into multiple concrete implementations for different types of alerting to make sure we are segregating responsibilities? This class is growing huge. It will also simplify start adding tests with dedicated scope.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes I agree and I have a task for refactoring to help cleanup monitor runner