address the comments for pr 1366

opensearch-project · Jan 9, 2024 · 86d45f8 · 86d45f8
1 parent a865512
commit 86d45f8
Show file tree

Hide file tree

Showing 4 changed files with 19 additions and 22 deletions.
diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/AlertService.kt b/alerting/src/main/kotlin/org/opensearch/alerting/AlertService.kt
@@ -67,7 +67,6 @@ import java.util.concurrent.TimeUnit
 import kotlin.coroutines.resume
 import kotlin.coroutines.resumeWithException
 import kotlin.coroutines.suspendCoroutine
-import kotlin.math.max
 
 /** Service that handles CRUD operations for alerts */
 class AlertService(
@@ -877,16 +876,6 @@ class AlertService(
         return searchResponse
     }
 
-    fun getCancelAfterTimeInterval(): Long {
-        // The default value for the cancelAfterTimeInterval is -1 and so, in this case
-        // we should ignore processing on the value
-        val givenInterval = MonitorRunnerService.monitorCtx.cancelAfterTimeInterval!!.minutes
-        if (givenInterval == -1L) {
-            return givenInterval
-        }
-        return max(givenInterval, ALERTS_SEARCH_TIMEOUT.minutes)
-    }
-
     private fun List<AlertError>?.update(alertError: AlertError?): List<AlertError> {
         return when {
             this == null && alertError == null -> emptyList()

diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/BucketLevelMonitorRunner.kt b/alerting/src/main/kotlin/org/opensearch/alerting/BucketLevelMonitorRunner.kt
@@ -24,6 +24,7 @@ import org.opensearch.alerting.script.BucketLevelTriggerExecutionContext
 import org.opensearch.alerting.util.defaultToPerExecutionAction
 import org.opensearch.alerting.util.getActionExecutionPolicy
 import org.opensearch.alerting.util.getBucketKeysHash
+import org.opensearch.alerting.util.getCancelAfterTimeInterval
 import org.opensearch.alerting.util.getCombinedTriggerRunResult
 import org.opensearch.alerting.workflow.WorkflowRunContext
 import org.opensearch.common.unit.TimeValue
@@ -411,8 +412,7 @@ object BucketLevelMonitorRunner : MonitorRunner() {
                             sr.source().query(queryBuilder)
                         }
                     sr.cancelAfterTimeInterval = TimeValue.timeValueMinutes(
-                        MonitorRunnerService
-                            .monitorCtx.alertService!!.getCancelAfterTimeInterval()
+                        getCancelAfterTimeInterval()
                     )
                     val searchResponse: SearchResponse = monitorCtx.client!!.suspendUntil { monitorCtx.client!!.search(sr, it) }
                     return createFindingPerIndex(searchResponse, monitor, monitorCtx, shouldCreateFinding, executionId)

diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/DocumentLevelMonitorRunner.kt b/alerting/src/main/kotlin/org/opensearch/alerting/DocumentLevelMonitorRunner.kt
@@ -26,6 +26,7 @@ import org.opensearch.alerting.util.AlertingException
 import org.opensearch.alerting.util.IndexUtils
 import org.opensearch.alerting.util.defaultToPerExecutionAction
 import org.opensearch.alerting.util.getActionExecutionPolicy
+import org.opensearch.alerting.util.getCancelAfterTimeInterval
 import org.opensearch.alerting.workflow.WorkflowRunContext
 import org.opensearch.client.Client
 import org.opensearch.client.node.NodeClient
@@ -582,10 +583,7 @@ object DocumentLevelMonitorRunner : MonitorRunner() {
                     .query(QueryBuilders.matchAllQuery())
                     .size(1)
             )
-        request.cancelAfterTimeInterval = TimeValue.timeValueMinutes(
-            MonitorRunnerService
-                .monitorCtx.alertService!!.getCancelAfterTimeInterval()
-        )
+        request.cancelAfterTimeInterval = TimeValue.timeValueMinutes(getCancelAfterTimeInterval())
 
         val response: SearchResponse = client.suspendUntil { client.search(request, it) }
 
@@ -675,10 +673,8 @@ object DocumentLevelMonitorRunner : MonitorRunner() {
             .preference(Preference.PRIMARY_FIRST.type())
 
         request.cancelAfterTimeInterval = TimeValue.timeValueMinutes(
-            MonitorRunnerService
-                .monitorCtx.alertService!!.getCancelAfterTimeInterval()
+            getCancelAfterTimeInterval()
         )
-
         val response: SearchResponse = monitorCtx.client!!.suspendUntil { monitorCtx.client!!.search(request, it) }
         if (response.status() !== RestStatus.OK) {
             throw IOException("Failed to search shard: $shard")
@@ -720,8 +716,7 @@ object DocumentLevelMonitorRunner : MonitorRunner() {
 
         try {
             searchRequest.cancelAfterTimeInterval = TimeValue.timeValueMinutes(
-                MonitorRunnerService
-                    .monitorCtx.alertService!!.getCancelAfterTimeInterval()
+                getCancelAfterTimeInterval()
             )
 
             response = monitorCtx.client!!.suspendUntil {

diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/util/AlertingUtils.kt b/alerting/src/main/kotlin/org/opensearch/alerting/util/AlertingUtils.kt
@@ -6,6 +6,8 @@
 package org.opensearch.alerting.util
 
 import org.apache.logging.log4j.LogManager
+import org.opensearch.alerting.AlertService
+import org.opensearch.alerting.MonitorRunnerService
 import org.opensearch.alerting.model.BucketLevelTriggerRunResult
 import org.opensearch.alerting.model.destination.Destination
 import org.opensearch.alerting.settings.DestinationSettings
@@ -18,6 +20,7 @@ import org.opensearch.commons.alerting.model.action.Action
 import org.opensearch.commons.alerting.model.action.ActionExecutionPolicy
 import org.opensearch.commons.alerting.model.action.ActionExecutionScope
 import org.opensearch.commons.alerting.util.isBucketLevelMonitor
+import kotlin.math.max
 
 private val logger = LogManager.getLogger("AlertingUtils")
 
@@ -163,6 +166,16 @@ inline fun <T : ThreadContext.StoredContext, R> T.use(block: (T) -> R): R {
     }
 }
 
+fun getCancelAfterTimeInterval(): Long {
+    // The default value for the cancelAfterTimeInterval is -1 and so, in this case
+    // we should ignore processing on the value
+    val givenInterval = MonitorRunnerService.monitorCtx.cancelAfterTimeInterval!!.minutes
+    if (givenInterval == -1L) {
+        return givenInterval
+    }
+    return max(givenInterval, AlertService.ALERTS_SEARCH_TIMEOUT.minutes)
+}
+
 /**
  * Closes this [AutoCloseable], suppressing possible exception or error thrown by [AutoCloseable.close] function when
  * it's being closed due to some other [cause] exception occurred.