[BUG] Honor configured timeout for scheduled PPL execution

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

@@ -58,8 +58,8 @@ import org.opensearch.index.query.RangeQueryBuilder
 import org.opensearch.index.query.TermsQueryBuilder
 import org.opensearch.script.ScriptService
 import org.opensearch.search.builder.SearchSourceBuilder
+import org.opensearch.transport.TransportService
 import org.opensearch.transport.client.Client
-import org.opensearch.transport.client.node.NodeClient
 import java.time.Duration
 import java.time.Instant
 import kotlin.time.measureTimedValue
@@ -223,7 +223,8 @@ class InputService(
 
     suspend fun collectInputResultsForPPLMonitor(
         monitor: Monitor,
-        monitorCtx: MonitorRunnerExecutionContext
+        monitorCtx: MonitorRunnerExecutionContext,
+        transportService: TransportService
     ): InputRunResults {
         return try {
             if (onlyHasCustomTriggers(monitor)) {
@@ -243,7 +244,8 @@ class InputService(
             val basePplQueryResults = runPPLBaseQuery(
                 monitor,
                 (monitor.inputs[0] as PPLInput).query,
-                monitorCtx
+                monitorCtx,
+                transportService
             )
             val numPplResults = basePplQueryResults.get("total").asLong()
 
@@ -294,6 +296,7 @@ class InputService(
         pplMonitor: Monitor,
         baseQuery: String,
         monitorCtx: MonitorRunnerExecutionContext,
+        transportService: TransportService,
     ): JsonNode {
 
         val queryExecutionDuration = monitorCtx
@@ -313,7 +316,9 @@ class InputService(
                 executePplQuery(
                     limitedQueryToExecute,
                     false,
-                    monitorCtx.client!! as NodeClient
+                    transportService,
+                    monitorCtx.clusterService!!.localNode(),
+                    queryExecutionDuration
                 )
             }
             logger.debug("base query results: $queryResponseJsonReceived")

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

@@ -9,14 +9,20 @@ import com.fasterxml.jackson.databind.JsonNode
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.fasterxml.jackson.databind.node.ArrayNode
 import org.json.JSONObject
+import org.opensearch.action.ActionListenerResponseHandler
 import org.opensearch.alerting.core.ppl.PPLPluginInterface
 import org.opensearch.alerting.opensearchapi.suspendUntil
+import org.opensearch.cluster.node.DiscoveryNode
+import org.opensearch.common.unit.TimeValue
 import org.opensearch.commons.utils.recreateObject
 import org.opensearch.core.action.ActionListener
 import org.opensearch.core.action.ActionResponse
+import org.opensearch.core.common.io.stream.Writeable
 import org.opensearch.sql.plugin.transport.PPLQueryAction
 import org.opensearch.sql.plugin.transport.TransportPPLQueryRequest
 import org.opensearch.sql.plugin.transport.TransportPPLQueryResponse
+import org.opensearch.transport.TransportRequestOptions
+import org.opensearch.transport.TransportService
 import org.opensearch.transport.client.node.NodeClient
 
 object PPLUtils {
@@ -124,6 +130,27 @@ object PPLUtils {
         return mapper.readTree(transportPplQueryResponse.result)
     }
 
+    suspend fun executePplQuery(
+        query: String,
+        explain: Boolean,
+        transportService: TransportService,
+        localNode: DiscoveryNode,
+        requestTimeout: TimeValue
+    ): JsonNode {
+        val transportPplQueryResponse = PPLPluginInterface.suspendUntil {
+            executePplQuery(
+                query,
+                explain,
+                transportService,
+                localNode,
+                requestTimeout,
+                it
+            )
+        }
+
+        return mapper.readTree(transportPplQueryResponse.result)
+    }
+
     /**
      * Executes a PPL query, for callback style flows with an action listener
      *
@@ -172,6 +199,39 @@ object PPLUtils {
         client.execute(PPLQueryAction.INSTANCE, request, wrappedListener)
     }
 
+    fun executePplQuery(
+        query: String,
+        explain: Boolean,
+        transportService: TransportService,
+        localNode: DiscoveryNode,
+        requestTimeout: TimeValue,
+        listener: ActionListener<TransportPPLQueryResponse>
+    ) {
+        val path = if (explain) {
+            "/_plugins/_ppl/_explain"
+        } else {
+            "/_plugins/_ppl"
+        }
+
+        val request = TransportPPLQueryRequest(
+            query,
+            JSONObject(mapOf("query" to query)),
+            path
+        )
+
+        val responseReader = Writeable.Reader { TransportPPLQueryResponse(it) }
+        transportService.sendRequest(
+            localNode,
+            PPLQueryAction.NAME,
+            request,
+            TransportRequestOptions
+                .builder()
+                .withTimeout(requestTimeout)
+                .build(),
+            object : ActionListenerResponseHandler<TransportPPLQueryResponse>(listener, responseReader) {}
+        )
+    }
+
     fun capAndReformatPPLQueryResults(rawQueryResults: JsonNode, maxSize: Long): List<Map<String, Any?>> {
         val cappedQueryResults = capPPLQueryResultsSize(rawQueryResults, maxSize)
         val cappedMap = mapper.convertValue(cappedQueryResults, Map::class.java) as Map<String, Any>

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

@@ -78,7 +78,7 @@ object QueryLevelMonitorRunner : MonitorRunner() {
             ) {
                 reinjectHeaders(monitor, monitorCtx)
                 monitorResult = monitorResult.copy(
-                    inputResults = monitorCtx.inputService!!.collectInputResultsForPPLMonitor(monitor, monitorCtx)
+                    inputResults = monitorCtx.inputService!!.collectInputResultsForPPLMonitor(monitor, monitorCtx, transportService)
                 )
             }
         } else {
@@ -188,7 +188,8 @@ object QueryLevelMonitorRunner : MonitorRunner() {
                                 monitor,
                                 pplTrigger,
                                 (monitor.inputs[0] as PPLInput).query,
-                                monitorCtx
+                                monitorCtx,
+                                transportService
                             )
                         }
                     }

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

@@ -49,7 +49,7 @@ import org.opensearch.script.ScriptService
 import org.opensearch.search.aggregations.Aggregation
 import org.opensearch.search.aggregations.Aggregations
 import org.opensearch.search.aggregations.support.AggregationPath
-import org.opensearch.transport.client.node.NodeClient
+import org.opensearch.transport.TransportService
 import kotlin.time.measureTimedValue
 
 /** Service that handles executing Triggers */
@@ -340,7 +340,8 @@ class TriggerService(val scriptService: ScriptService) {
         pplMonitor: Monitor,
         pplTrigger: PPLTrigger,
         query: String,
-        monitorCtx: MonitorRunnerExecutionContext
+        monitorCtx: MonitorRunnerExecutionContext,
+        transportService: TransportService
     ): QueryLevelTriggerRunResult {
 
         if (pplTrigger.customCondition == null) {
@@ -397,7 +398,9 @@ class TriggerService(val scriptService: ScriptService) {
                         executePplQuery(
                             limitedQueryToExecute,
                             false,
-                            monitorCtx.client!! as NodeClient
+                            transportService,
+                            monitorCtx.clusterService!!.localNode(),
+                            pplTriggerExecutionDuration
                         )
                     }
                 }