diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/InputService.kt b/alerting/src/main/kotlin/org/opensearch/alerting/InputService.kt index baa9c0c8f..3d5aff9ea 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/InputService.kt +++ b/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") diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/PPLUtils.kt b/alerting/src/main/kotlin/org/opensearch/alerting/PPLUtils.kt index 82132cc76..fd4b331ca 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/PPLUtils.kt +++ b/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 + ) { + 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(listener, responseReader) {} + ) + } + fun capAndReformatPPLQueryResults(rawQueryResults: JsonNode, maxSize: Long): List> { val cappedQueryResults = capPPLQueryResultsSize(rawQueryResults, maxSize) val cappedMap = mapper.convertValue(cappedQueryResults, Map::class.java) as Map diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/QueryLevelMonitorRunner.kt b/alerting/src/main/kotlin/org/opensearch/alerting/QueryLevelMonitorRunner.kt index 83c44387b..04188dc23 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/QueryLevelMonitorRunner.kt +++ b/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 ) } } diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/TriggerService.kt b/alerting/src/main/kotlin/org/opensearch/alerting/TriggerService.kt index ecea465f5..142681b89 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/TriggerService.kt +++ b/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 ) } }