From 306ba812e4f4a72e12b016df32eb7a43caa40d92 Mon Sep 17 00:00:00 2001 From: toepkerd <120457569+toepkerd@users.noreply.github.com> Date: Wed, 29 Oct 2025 22:58:35 -0700 Subject: [PATCH] PPL Alerting: Create and Update Monitor V2 (#1961) * PPL Alerting: Create and Update Monitor V2 Signed-off-by: Dennis Toepker * addressing PR comments Signed-off-by: Dennis Toepker * removing AlertingV1Utils Signed-off-by: Dennis Toepker --------- Signed-off-by: Dennis Toepker Co-authored-by: Dennis Toepker (cherry picked from commit c911cfa625c8c3de2b47aefc0962cda25b426051) --- alerting/build.gradle | 27 + .../org/opensearch/alerting/AlertingPlugin.kt | 37 +- .../opensearch/alerting/AlertingV2Utils.kt | 87 ++ .../org/opensearch/alerting/PPLUtils.kt | 275 ++++++ .../alerting/actionv2/IndexMonitorV2Action.kt | 15 + .../actionv2/IndexMonitorV2Request.kt | 69 ++ .../actionv2/IndexMonitorV2Response.kt | 73 ++ .../resthandlerv2/RestIndexMonitorV2Action.kt | 84 ++ .../alerting/settings/AlertingSettings.kt | 103 +++ .../transport/TransportIndexMonitorAction.kt | 11 +- .../transport/TransportIndexWorkflowAction.kt | 16 +- .../TransportIndexMonitorV2Action.kt | 841 ++++++++++++++++++ .../opensearch/alerting/util/IndexUtils.kt | 15 + .../org/opensearch/alerting/AccessRoles.kt | 2 + .../alerting/AlertingRestTestCase.kt | 218 ++++- .../alerting/resthandler/MonitorRestApiIT.kt | 17 +- .../resthandler/MonitorV2RestApiIT.kt | 63 ++ core/build.gradle | 51 +- .../alerting/core/ppl/PPLPluginInterface.kt | 50 ++ .../opensearchapi/OpenSearchExtensions.kt | 15 + 20 files changed, 2044 insertions(+), 25 deletions(-) create mode 100644 alerting/src/main/kotlin/org/opensearch/alerting/AlertingV2Utils.kt create mode 100644 alerting/src/main/kotlin/org/opensearch/alerting/PPLUtils.kt create mode 100644 alerting/src/main/kotlin/org/opensearch/alerting/actionv2/IndexMonitorV2Action.kt create mode 100644 alerting/src/main/kotlin/org/opensearch/alerting/actionv2/IndexMonitorV2Request.kt create mode 100644 alerting/src/main/kotlin/org/opensearch/alerting/actionv2/IndexMonitorV2Response.kt create mode 100644 alerting/src/main/kotlin/org/opensearch/alerting/resthandlerv2/RestIndexMonitorV2Action.kt create mode 100644 alerting/src/main/kotlin/org/opensearch/alerting/transportv2/TransportIndexMonitorV2Action.kt create mode 100644 alerting/src/test/kotlin/org/opensearch/alerting/resthandler/MonitorV2RestApiIT.kt create mode 100644 core/src/main/kotlin/org/opensearch/alerting/core/ppl/PPLPluginInterface.kt diff --git a/alerting/build.gradle b/alerting/build.gradle index 1ae549d5b..416e29df7 100644 --- a/alerting/build.gradle +++ b/alerting/build.gradle @@ -151,6 +151,8 @@ dependencies { // Needed for integ tests zipArchive group: 'org.opensearch.plugin', name:'opensearch-notifications-core', version: "${opensearch_build}" zipArchive group: 'org.opensearch.plugin', name:'notifications', version: "${opensearch_build}" + zipArchive group: 'org.opensearch.plugin', name:'opensearch-job-scheduler', version: "${opensearch_build}" + zipArchive group: 'org.opensearch.plugin', name:'opensearch-sql-plugin', version: "${opensearch_build}" // Needed for security tests if (securityEnabled) { @@ -168,7 +170,10 @@ dependencies { implementation "org.jetbrains.kotlin:kotlin-stdlib-common:${kotlin_version}" implementation "org.jetbrains:annotations:13.0" + // SQL/PPL plugin dependencies are included in alerting-core api project(":alerting-core") + implementation 'org.json:json:20240303' + implementation "com.github.seancfoley:ipaddress:5.4.1" implementation project(path: ":alerting-spi", configuration: 'shadow') @@ -246,6 +251,28 @@ testClusters.integTest { } })) + plugin(provider({ + new RegularFile() { + @Override + File getAsFile() { + return configurations.zipArchive.asFileTree.matching { + include '**/opensearch-job-scheduler*' + }.singleFile + } + } + })) + + plugin(provider({ + new RegularFile() { + @Override + File getAsFile() { + return configurations.zipArchive.asFileTree.matching { + include '**/opensearch-sql-plugin*' + }.singleFile + } + } + })) + if (securityEnabled) { plugin(provider({ new RegularFile() { diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/AlertingPlugin.kt b/alerting/src/main/kotlin/org/opensearch/alerting/AlertingPlugin.kt index 4ad7b6361..31c11c320 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/AlertingPlugin.kt +++ b/alerting/src/main/kotlin/org/opensearch/alerting/AlertingPlugin.kt @@ -14,6 +14,7 @@ import org.opensearch.alerting.action.GetEmailGroupAction import org.opensearch.alerting.action.GetRemoteIndexesAction import org.opensearch.alerting.action.SearchEmailAccountAction import org.opensearch.alerting.action.SearchEmailGroupAction +import org.opensearch.alerting.actionv2.IndexMonitorV2Action import org.opensearch.alerting.alerts.AlertIndices import org.opensearch.alerting.alerts.AlertIndices.Companion.ALL_ALERT_INDEX_PATTERN import org.opensearch.alerting.comments.CommentsIndices @@ -27,6 +28,7 @@ import org.opensearch.alerting.core.resthandler.RestScheduledJobStatsHandler import org.opensearch.alerting.core.schedule.JobScheduler import org.opensearch.alerting.core.settings.LegacyOpenDistroScheduledJobSettings import org.opensearch.alerting.core.settings.ScheduledJobSettings +import org.opensearch.alerting.modelv2.MonitorV2 import org.opensearch.alerting.remote.monitors.RemoteMonitorRegistry import org.opensearch.alerting.resthandler.RestAcknowledgeAlertAction import org.opensearch.alerting.resthandler.RestAcknowledgeChainedAlertAction @@ -51,6 +53,7 @@ import org.opensearch.alerting.resthandler.RestSearchAlertingCommentAction import org.opensearch.alerting.resthandler.RestSearchEmailAccountAction import org.opensearch.alerting.resthandler.RestSearchEmailGroupAction import org.opensearch.alerting.resthandler.RestSearchMonitorAction +import org.opensearch.alerting.resthandlerv2.RestIndexMonitorV2Action import org.opensearch.alerting.script.TriggerScript import org.opensearch.alerting.service.DeleteMonitorService import org.opensearch.alerting.settings.AlertingSettings @@ -83,6 +86,7 @@ import org.opensearch.alerting.transport.TransportSearchAlertingCommentAction import org.opensearch.alerting.transport.TransportSearchEmailAccountAction import org.opensearch.alerting.transport.TransportSearchEmailGroupAction import org.opensearch.alerting.transport.TransportSearchMonitorAction +import org.opensearch.alerting.transportv2.TransportIndexMonitorV2Action import org.opensearch.alerting.util.DocLevelMonitorQueries import org.opensearch.alerting.util.destinationmigration.DestinationMigrationCoordinator import org.opensearch.cluster.metadata.IndexNameExpressionResolver @@ -157,6 +161,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R @JvmField val OPEN_SEARCH_DASHBOARDS_USER_AGENT = "OpenSearch-Dashboards" @JvmField val UI_METADATA_EXCLUDE = arrayOf("monitor.${Monitor.UI_METADATA_FIELD}") @JvmField val MONITOR_BASE_URI = "/_plugins/_alerting/monitors" + @JvmField val MONITOR_V2_BASE_URI = "/_plugins/_alerting/v2/monitors" @JvmField val WORKFLOW_BASE_URI = "/_plugins/_alerting/workflows" @JvmField val REMOTE_BASE_URI = "/_plugins/_alerting/remote" @JvmField val DESTINATION_BASE_URI = "/_plugins/_alerting/destinations" @@ -169,7 +174,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R @JvmField val FINDING_BASE_URI = "/_plugins/_alerting/findings" @JvmField val COMMENTS_BASE_URI = "/_plugins/_alerting/comments" - @JvmField val ALERTING_JOB_TYPES = listOf("monitor", "workflow") + @JvmField val ALERTING_JOB_TYPES = listOf("monitor", "workflow", "monitor_v2") } lateinit var runner: MonitorRunnerService @@ -194,6 +199,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R nodesInCluster: Supplier ): List { return listOf( + // Alerting V1 RestGetMonitorAction(), RestDeleteMonitorAction(), RestIndexMonitorAction(), @@ -218,11 +224,15 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R RestIndexAlertingCommentAction(), RestSearchAlertingCommentAction(), RestDeleteAlertingCommentAction(), + + // Alerting V2 + RestIndexMonitorV2Action(), ) } override fun getActions(): List> { return listOf( + // Alerting V1 ActionPlugin.ActionHandler(ScheduledJobsStatsAction.INSTANCE, ScheduledJobsStatsTransportAction::class.java), ActionPlugin.ActionHandler(AlertingActions.INDEX_MONITOR_ACTION_TYPE, TransportIndexMonitorAction::class.java), ActionPlugin.ActionHandler(AlertingActions.GET_MONITOR_ACTION_TYPE, TransportGetMonitorAction::class.java), @@ -249,13 +259,17 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R ActionPlugin.ActionHandler(AlertingActions.DELETE_COMMENT_ACTION_TYPE, TransportDeleteAlertingCommentAction::class.java), ActionPlugin.ActionHandler(ExecuteWorkflowAction.INSTANCE, TransportExecuteWorkflowAction::class.java), ActionPlugin.ActionHandler(GetRemoteIndexesAction.INSTANCE, TransportGetRemoteIndexesAction::class.java), - ActionPlugin.ActionHandler(DocLevelMonitorFanOutAction.INSTANCE, TransportDocLevelMonitorFanOutAction::class.java) + ActionPlugin.ActionHandler(DocLevelMonitorFanOutAction.INSTANCE, TransportDocLevelMonitorFanOutAction::class.java), + + // Alerting V2 + ActionPlugin.ActionHandler(IndexMonitorV2Action.INSTANCE, TransportIndexMonitorV2Action::class.java), ) } override fun getNamedXContent(): List { return listOf( Monitor.XCONTENT_REGISTRY, + MonitorV2.XCONTENT_REGISTRY, SearchInput.XCONTENT_REGISTRY, DocLevelMonitorInput.XCONTENT_REGISTRY, QueryLevelTrigger.XCONTENT_REGISTRY, @@ -431,7 +445,22 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R AlertingSettings.COMMENTS_HISTORY_RETENTION_PERIOD, AlertingSettings.COMMENTS_MAX_CONTENT_SIZE, AlertingSettings.MAX_COMMENTS_PER_ALERT, - AlertingSettings.MAX_COMMENTS_PER_NOTIFICATION + AlertingSettings.MAX_COMMENTS_PER_NOTIFICATION, + AlertingSettings.ALERT_V2_HISTORY_ENABLED, + AlertingSettings.ALERT_V2_HISTORY_ROLLOVER_PERIOD, + AlertingSettings.ALERT_V2_HISTORY_INDEX_MAX_AGE, + AlertingSettings.ALERT_V2_HISTORY_MAX_DOCS, + AlertingSettings.ALERT_V2_HISTORY_RETENTION_PERIOD, + AlertingSettings.ALERTING_V2_MAX_MONITORS, + AlertingSettings.ALERTING_V2_MAX_THROTTLE_DURATION, + AlertingSettings.ALERTING_V2_MAX_EXPIRE_DURATION, + AlertingSettings.ALERTING_V2_MAX_LOOK_BACK_WINDOW, + AlertingSettings.ALERTING_V2_MAX_QUERY_LENGTH, + AlertingSettings.ALERTING_V2_QUERY_RESULTS_MAX_DATAROWS, + AlertingSettings.ALERT_V2_QUERY_RESULTS_MAX_SIZE, + AlertingSettings.ALERT_V2_PER_RESULT_TRIGGER_MAX_ALERTS, + AlertingSettings.NOTIFICATION_SUBJECT_SOURCE_MAX_LENGTH, + AlertingSettings.NOTIFICATION_MESSAGE_SOURCE_MAX_LENGTH ) } @@ -449,7 +478,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R return listOf( SystemIndexDescriptor(ALL_ALERT_INDEX_PATTERN, "Alerting Plugin system index pattern"), SystemIndexDescriptor(SCHEDULED_JOBS_INDEX, "Alerting Plugin Configuration index"), - SystemIndexDescriptor(ALL_COMMENTS_INDEX_PATTERN, "Alerting Comments system index pattern") + SystemIndexDescriptor(ALL_COMMENTS_INDEX_PATTERN, "Alerting Comments system index pattern"), ) } diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/AlertingV2Utils.kt b/alerting/src/main/kotlin/org/opensearch/alerting/AlertingV2Utils.kt new file mode 100644 index 000000000..692b7db6c --- /dev/null +++ b/alerting/src/main/kotlin/org/opensearch/alerting/AlertingV2Utils.kt @@ -0,0 +1,87 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.alerting + +import org.apache.lucene.search.TotalHits +import org.apache.lucene.search.TotalHits.Relation +import org.opensearch.action.search.SearchResponse +import org.opensearch.action.search.ShardSearchFailure +import org.opensearch.alerting.modelv2.MonitorV2 +import org.opensearch.commons.alerting.model.Monitor +import org.opensearch.commons.alerting.model.ScheduledJob +import org.opensearch.commons.alerting.model.Workflow +import org.opensearch.index.IndexNotFoundException +import org.opensearch.search.SearchHits +import org.opensearch.search.aggregations.InternalAggregations +import org.opensearch.search.internal.InternalSearchResponse +import org.opensearch.search.profile.SearchProfileShardResults +import org.opensearch.search.suggest.Suggest +import org.opensearch.transport.RemoteTransportException +import java.util.Collections + +object AlertingV2Utils { + // Validates that the given scheduled job is a Monitor + // returns the exception to pass into actionListener.onFailure if not. + fun validateMonitorV1(scheduledJob: ScheduledJob): Exception? { + if (scheduledJob is MonitorV2) { + return IllegalStateException("The ID given corresponds to a V2 Monitor, but a V1 Monitor was expected") + } else if (scheduledJob !is Monitor && scheduledJob !is Workflow) { + return IllegalStateException("The ID given corresponds to a scheduled job of unknown type: ${scheduledJob.javaClass.name}") + } + return null + } + + // Validates that the given scheduled job is a MonitorV2 + // returns the exception to pass into actionListener.onFailure if not. + fun validateMonitorV2(scheduledJob: ScheduledJob): Exception? { + if (scheduledJob is Monitor || scheduledJob is Workflow) { + return IllegalStateException("The ID given corresponds to a V1 Monitor, but a V2 Monitor was expected") + } else if (scheduledJob !is MonitorV2) { + return IllegalStateException("The ID given corresponds to a scheduled job of unknown type: ${scheduledJob.javaClass.name}") + } + return null + } + + // Checks if the exception is caused by an IndexNotFoundException (directly or nested). + // Used in Get and Search monitor functionalities to determine whether a "no results" + // response should be returned + fun isIndexNotFoundException(e: Exception): Boolean { + if (e is IndexNotFoundException) { + return true + } + if (e is RemoteTransportException) { + val cause = e.cause + if (cause is IndexNotFoundException) { + return true + } + } + return false + } + + // Used in Get and Search monitor functionalities to return a "no results" response + fun getEmptySearchResponse(): SearchResponse { + val internalSearchResponse = InternalSearchResponse( + SearchHits(emptyArray(), TotalHits(0L, Relation.EQUAL_TO), 0.0f), + InternalAggregations.from(Collections.emptyList()), + Suggest(Collections.emptyList()), + SearchProfileShardResults(Collections.emptyMap()), + false, + false, + 0 + ) + + return SearchResponse( + internalSearchResponse, + "", + 0, + 0, + 0, + 0, + ShardSearchFailure.EMPTY_ARRAY, + SearchResponse.Clusters.EMPTY + ) + } +} diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/PPLUtils.kt b/alerting/src/main/kotlin/org/opensearch/alerting/PPLUtils.kt new file mode 100644 index 000000000..c4d818136 --- /dev/null +++ b/alerting/src/main/kotlin/org/opensearch/alerting/PPLUtils.kt @@ -0,0 +1,275 @@ +package org.opensearch.alerting + +import org.json.JSONArray +import org.json.JSONObject +import org.opensearch.alerting.core.ppl.PPLPluginInterface +import org.opensearch.alerting.opensearchapi.suspendUntil +import org.opensearch.sql.plugin.transport.TransportPPLQueryRequest +import org.opensearch.transport.client.node.NodeClient + +object PPLUtils { + /** + * Appends a user-defined custom condition to a PPL query. + * + * This method is used exclusively for custom condition triggers. It concatenates + * the custom condition to the base PPL query using the pipe operator (|), allowing the condition + * to evaluate each query result data row. + * + * @param query The base PPL query string (e.g., "source=logs | where status=error") + * @param customCondition The custom trigger condition to append (e.g., "eval result = avg > 3") + * @return The combined PPL query with the custom condition appended + * + * @example + * ``` + * val baseQuery = "source=logs | stats max(price) as max_price by region" + * val condition = "eval result = max_price > 300" + * val result = appendCustomCondition(baseQuery, condition) + * // Returns: "source=logs | stats max(price) as max_price by region | eval result = max_price > 300" + * ``` + * + * @note This method does not validate the syntax of either the query or custom condition. + * It is assumed that upstream workflows have already validated the base query, + * and that downstream workflows will validate the constructed query + */ + fun appendCustomCondition(query: String, customCondition: String): String { + return "$query | $customCondition" + } + + /** + * Appends a limit on the number of documents/data rows to retrieve from a PPL query. + * + * This method uses the PPL `head` command to restrict the number of rows returned by + * the query. This is used to prevent memory issues and improving performance when + * only a subset of results is needed for alert evaluation. + * + * @param query The base PPL query string + * @param maxDataRows The maximum number of data rows to retrieve + * @return The PPL query with a head limit appended (e.g., "source=logs | head 1000") + * + * @example + * ``` + * val query = "source=logs | where status=error" + * val limitedQuery = appendDataRowsLimit(query, 100) + * // Returns: "source=logs | where status=error | head 100" + * ``` + */ + fun appendDataRowsLimit(query: String, maxDataRows: Long): String { + return "$query | head $maxDataRows" + } + + /** + * Executes a PPL query and returns the response as a parsable JSONObject. + * + * This method calls the PPL Plugin's Execute API via the transport layer to execute the provided query + * and parses the response into a structured JSON format suitable for trigger evaluation + * + * @param query The PPL query string to execute + * @param client The NodeClient used to communicate with the PPL plugin + * @return A JSONObject containing the query execution results + * + * @throws Exception if the query execution fails or the response cannot be parsed as JSON + * + * @note The response format follows the PPL plugin's Execute API response structure with + * "schema", "datarows", "total", and "size" fields. + */ + suspend fun executePplQuery(query: String, client: NodeClient): JSONObject { + // call PPL plugin to execute query + val transportPplQueryRequest = TransportPPLQueryRequest( + query, + JSONObject(mapOf("query" to query)), + null // null path falls back to a default path internal to SQL/PPL Plugin + ) + + val transportPplQueryResponse = PPLPluginInterface.suspendUntil { + this.executeQuery( + client, + transportPplQueryRequest, + it + ) + } + + val queryResponseJson = JSONObject(transportPplQueryResponse.result) + + return queryResponseJson + } + + /** + * Searches a custom condition eval statement for the name of the eval result variable. + * + * Parses a PPL eval expression to extract the variable name being assigned. The eval + * statement must follow the format: `eval = `. This variable + * name is needed to reference the evaluation result in subsequent trigger condition checks. + * + * @param customCondition The PPL custom condition string containing an eval statement (e.g. eval result = avg > 3) + * @return The name of the eval result variable + * @throws IllegalArgumentException if no valid eval statement is found or the syntax is invalid + * + * @example + * ``` + * val condition = "eval error_rate = errors / total" + * val varName = findEvalResultVar(condition) + * // Returns: "error_rate" + * ``` + * + * @note A precheck of the base query + custom condition is assumed to have been done already. + * The function thus expects the PPL keyword "eval" followed by whitespace. Without the + * whitespace (e.g., "evalresult"), the PPL plugin would have thrown a syntax error + * during upstream validations + * @note Variable names must follow standard identifier rules: start with a letter or underscore, + * followed by letters, digits, or underscores (matching `[a-zA-Z_][a-zA-Z0-9_]*`). + * + * TODO: Replace this in-house parser with a PPL plugin dependency that provides proper + * query parsing functionality. + */ + fun findEvalResultVar(customCondition: String): String { + // TODO: these are in-house PPL query parsers, find a PPL plugin dependency that does this for us + val regex = """\beval\s+([a-zA-Z_][a-zA-Z0-9_]*)\s*=""".toRegex() + val evalResultVar = regex.find(customCondition)?.groupValues?.get(1) + ?: throw IllegalArgumentException("Given custom condition is invalid, could not find eval result variable") + return evalResultVar + } + + /** + * Finds the index of the eval result variable in the PPL query response schema. + * + * Searches through the schema array in the PPL query response to locate the column + * corresponding to the eval result variable. This index is used to extract the + * eval result values from the datarows in the query response. + * + * @param customConditionQueryResponse The JSONObject containing the PPL query response + * with "schema" and "datarows" fields + * @param evalResultVarName The name of the eval result variable to locate in the schema + * @return The zero-based index of the eval result variable in the schema array + * @throws IllegalStateException if the eval result variable is not found in the schema + * + * @note The eval result variable should always be present in the schema if the query + * executed successfully. If not found, this indicates an unexpected state. + * @note The query response schema is assumed to follow PPL plugin Execute API response schema + */ + fun findEvalResultVarIdxInSchema(customConditionQueryResponse: JSONObject, evalResultVarName: String): Int { + // find the index eval statement result variable in the PPL query response schema + val schemaList = customConditionQueryResponse.getJSONArray("schema") + var evalResultVarIdx = -1 + for (i in 0 until schemaList.length()) { + val schemaObj = schemaList.getJSONObject(i) + val columnName = schemaObj.getString("name") + + if (columnName == evalResultVarName) { + evalResultVarIdx = i + break + } + } + + // eval statement result variable should always be found + if (evalResultVarIdx == -1) { + throw IllegalStateException( + "Expected to find eval statement results variable \"$evalResultVarName\" in results " + + "of PPL query with custom condition, but did not." + ) + } + + return evalResultVarIdx + } + + /** + * Extracts the list of indices from a PPL query's source statement. + * + * Parses the PPL `source=` clause to identify which indices, index patterns, or index + * aliases are being queried. This information is primarily used for permission checks. + * Supports comma-separated lists of indices and wildcard patterns. + * + * @param pplQuery The complete PPL query string containing a source statement + * @return A list of index names, patterns, or aliases (e.g., ["logs-*", "metrics-2024"]) + * @throws IllegalStateException if no valid source statement is found, even after + * the query has been validated by the SQL/PPL plugin + * + * @example + * ``` + * val query = "source=logs-* | where level='ERROR'" + * val indices = getIndicesFromPplQuery(query) + * // Returns: ["logs-*"] + * + * val multiQuery = "source=logs-*, metrics-2024, .kibana | stats count()" + * val multiIndices = getIndicesFromPplQuery(multiQuery) + * // Returns: ["logs-*", "metrics-2024", ".kibana"] + * ``` + * + * @note Supports concrete indices, wildcard patterns (*), dot-prefixed system indices, + * and index aliases + * @note PPL queries contain exactly one source statement, so only the first match is used + * @note The regex pattern handles optional whitespace around `=` and commas + * + */ + fun getIndicesFromPplQuery(pplQuery: String): List { + // captures comma-separated concrete indices, index patterns, and index aliases + // TODO: these are in-house PPL query parsers, find a PPL plugin dependency that does this for us + val indicesRegex = """(?i)source(?:\s*)=(?:\s*)([-\w.*'+]+(?:\*)?(?:\s*,\s*[-\w.*'+]+\*?)*)\s*\|*""".toRegex() + + // use find() instead of findAll() because a PPL query only ever has one source statement + // the only capture group specified in the regex captures the comma separated string of indices/index patterns + val indices = indicesRegex.find(pplQuery)?.groupValues?.get(1)?.split(",")?.map { it.trim() } + ?: throw IllegalStateException( + "Could not find indices that PPL Monitor query searches even " + + "after validating the query through SQL/PPL plugin." + ) + + return indices + } + + /** + * Caps the size of PPL query results to prevent memory issues and oversized alert payloads. + * + * Checks if the serialized query results exceed a specified size limit. If the results + * are within the limit, they are returned unchanged. If they exceed the limit, the datarows + * are replaced with an informational message while preserving the schema and metadata fields. + * This ensures alerts can still be created even when query results are too large. + * + * @param pplQueryResults The PPL query response JSONObject + * @param maxSize The maximum allowed size in bytes (estimated by serialized string length) + * @return The original results if under the limit, or a modified version with datarows replaced by a message + * + * @example + * ``` + * val queryResults = executePplQuery(query, client) + * val cappedResults = capPPLQueryResultsSize(queryResults, maxSize = 5000L) + * + * // If results were too large, datarows will contain: + * // [["The PPL Query results were too large and thus excluded"]] + * // But schema, total, and size fields are preserved + * ``` + * + * @note Size is estimated using `toString().length`, which approximates byte size but may + * not be exact for multi-byte characters + * @note The PPL query results structure includes: + * - `schema`: Array of objects storing data types for each column + * - `datarows`: Array of arrays containing the actual query result rows + * - `total`: Total number of result rows + * - `size`: Same as `total` (redundant field in PPL response) + */ + fun capPPLQueryResultsSize(pplQueryResults: JSONObject, maxSize: Long): JSONObject { + // estimate byte size with serialized string length + // if query results size are already under the limit, do nothing + // and return the query results as is + val pplQueryResultsSize = pplQueryResults.toString().length + if (pplQueryResultsSize <= maxSize) { + return pplQueryResults + } + + // if the query results exceed the limit, we need to replace the query results + // with a message that says the results were too large, but still retain the other + // ppl query response fields like schema, total, and size + val limitExceedMessageQueryResults = JSONObject() + + val schema = JSONArray(pplQueryResults.getJSONArray("schema").toList()) + val datarows = JSONArray().put(JSONArray(listOf("The PPL Query results were too large and thus excluded"))) + val total = pplQueryResults.getInt("total") + val size = pplQueryResults.getInt("size") + + limitExceedMessageQueryResults.put("schema", schema) + limitExceedMessageQueryResults.put("datarows", datarows) + limitExceedMessageQueryResults.put("total", total) + limitExceedMessageQueryResults.put("size", size) + + return limitExceedMessageQueryResults + } +} diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/actionv2/IndexMonitorV2Action.kt b/alerting/src/main/kotlin/org/opensearch/alerting/actionv2/IndexMonitorV2Action.kt new file mode 100644 index 000000000..aab23b631 --- /dev/null +++ b/alerting/src/main/kotlin/org/opensearch/alerting/actionv2/IndexMonitorV2Action.kt @@ -0,0 +1,15 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.alerting.actionv2 + +import org.opensearch.action.ActionType + +class IndexMonitorV2Action private constructor() : ActionType(NAME, ::IndexMonitorV2Response) { + companion object { + val INSTANCE = IndexMonitorV2Action() + const val NAME = "cluster:admin/opensearch/alerting/v2/monitor/write" + } +} diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/actionv2/IndexMonitorV2Request.kt b/alerting/src/main/kotlin/org/opensearch/alerting/actionv2/IndexMonitorV2Request.kt new file mode 100644 index 000000000..105408d07 --- /dev/null +++ b/alerting/src/main/kotlin/org/opensearch/alerting/actionv2/IndexMonitorV2Request.kt @@ -0,0 +1,69 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.alerting.actionv2 + +import org.opensearch.action.ActionRequest +import org.opensearch.action.ActionRequestValidationException +import org.opensearch.action.support.WriteRequest +import org.opensearch.alerting.modelv2.MonitorV2 +import org.opensearch.core.common.io.stream.StreamInput +import org.opensearch.core.common.io.stream.StreamOutput +import org.opensearch.rest.RestRequest +import java.io.IOException + +class IndexMonitorV2Request : ActionRequest { + val monitorId: String + val seqNo: Long + val primaryTerm: Long + val refreshPolicy: WriteRequest.RefreshPolicy + val method: RestRequest.Method + var monitorV2: MonitorV2 + val rbacRoles: List? + + constructor( + monitorId: String, + seqNo: Long, + primaryTerm: Long, + refreshPolicy: WriteRequest.RefreshPolicy, + method: RestRequest.Method, + monitorV2: MonitorV2, + rbacRoles: List? = null + ) : super() { + this.monitorId = monitorId + this.seqNo = seqNo + this.primaryTerm = primaryTerm + this.refreshPolicy = refreshPolicy + this.method = method + this.monitorV2 = monitorV2 + this.rbacRoles = rbacRoles + } + + @Throws(IOException::class) + constructor(sin: StreamInput) : this( + monitorId = sin.readString(), + seqNo = sin.readLong(), + primaryTerm = sin.readLong(), + refreshPolicy = WriteRequest.RefreshPolicy.readFrom(sin), + method = sin.readEnum(RestRequest.Method::class.java), + monitorV2 = MonitorV2.readFrom(sin), + rbacRoles = sin.readOptionalStringList() + ) + + override fun validate(): ActionRequestValidationException? { + return null + } + + @Throws(IOException::class) + override fun writeTo(out: StreamOutput) { + out.writeString(monitorId) + out.writeLong(seqNo) + out.writeLong(primaryTerm) + refreshPolicy.writeTo(out) + out.writeEnum(method) + MonitorV2.writeTo(out, monitorV2) + out.writeOptionalStringCollection(rbacRoles) + } +} diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/actionv2/IndexMonitorV2Response.kt b/alerting/src/main/kotlin/org/opensearch/alerting/actionv2/IndexMonitorV2Response.kt new file mode 100644 index 000000000..99d076334 --- /dev/null +++ b/alerting/src/main/kotlin/org/opensearch/alerting/actionv2/IndexMonitorV2Response.kt @@ -0,0 +1,73 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.alerting.actionv2 + +import org.opensearch.alerting.modelv2.MonitorV2 +import org.opensearch.commons.alerting.util.IndexUtils.Companion._ID +import org.opensearch.commons.alerting.util.IndexUtils.Companion._PRIMARY_TERM +import org.opensearch.commons.alerting.util.IndexUtils.Companion._SEQ_NO +import org.opensearch.commons.alerting.util.IndexUtils.Companion._VERSION +import org.opensearch.commons.notifications.action.BaseResponse +import org.opensearch.core.common.io.stream.StreamInput +import org.opensearch.core.common.io.stream.StreamOutput +import org.opensearch.core.xcontent.ToXContent +import org.opensearch.core.xcontent.XContentBuilder +import java.io.IOException + +class IndexMonitorV2Response : BaseResponse { + var id: String + var version: Long + var seqNo: Long + var primaryTerm: Long + var monitorV2: MonitorV2 + + constructor( + id: String, + version: Long, + seqNo: Long, + primaryTerm: Long, + monitorV2: MonitorV2 + ) : super() { + this.id = id + this.version = version + this.seqNo = seqNo + this.primaryTerm = primaryTerm + this.monitorV2 = monitorV2 + } + + @Throws(IOException::class) + constructor(sin: StreamInput) : this( + sin.readString(), // id + sin.readLong(), // version + sin.readLong(), // seqNo + sin.readLong(), // primaryTerm + MonitorV2.readFrom(sin) // monitorV2 + ) + + @Throws(IOException::class) + override fun writeTo(out: StreamOutput) { + out.writeString(id) + out.writeLong(version) + out.writeLong(seqNo) + out.writeLong(primaryTerm) + MonitorV2.writeTo(out, monitorV2) + } + + @Throws(IOException::class) + override fun toXContent(builder: XContentBuilder, params: ToXContent.Params): XContentBuilder { + return builder.startObject() + .field(_ID, id) + .field(_VERSION, version) + .field(_SEQ_NO, seqNo) + .field(_PRIMARY_TERM, primaryTerm) + .field(MONITOR_V2_FIELD, monitorV2) + .endObject() + } + + companion object { + const val MONITOR_V2_FIELD = "monitor_v2" + } +} diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/resthandlerv2/RestIndexMonitorV2Action.kt b/alerting/src/main/kotlin/org/opensearch/alerting/resthandlerv2/RestIndexMonitorV2Action.kt new file mode 100644 index 000000000..adafb924a --- /dev/null +++ b/alerting/src/main/kotlin/org/opensearch/alerting/resthandlerv2/RestIndexMonitorV2Action.kt @@ -0,0 +1,84 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.alerting.resthandlerv2 + +import org.apache.logging.log4j.LogManager +import org.opensearch.action.support.WriteRequest +import org.opensearch.alerting.AlertingPlugin +import org.opensearch.alerting.actionv2.IndexMonitorV2Action +import org.opensearch.alerting.actionv2.IndexMonitorV2Request +import org.opensearch.alerting.modelv2.MonitorV2 +import org.opensearch.alerting.util.IF_PRIMARY_TERM +import org.opensearch.alerting.util.IF_SEQ_NO +import org.opensearch.alerting.util.REFRESH +import org.opensearch.commons.alerting.util.AlertingException +import org.opensearch.core.xcontent.XContentParser.Token +import org.opensearch.core.xcontent.XContentParserUtils.ensureExpectedToken +import org.opensearch.index.seqno.SequenceNumbers +import org.opensearch.rest.BaseRestHandler +import org.opensearch.rest.RestHandler.Route +import org.opensearch.rest.RestRequest +import org.opensearch.rest.RestRequest.Method.POST +import org.opensearch.rest.RestRequest.Method.PUT +import org.opensearch.rest.action.RestToXContentListener +import org.opensearch.transport.client.node.NodeClient +import java.io.IOException + +private val log = LogManager.getLogger(RestIndexMonitorV2Action::class.java) + +/** + * Rest handlers to create and update V2 Monitors like PPL Monitors + */ +class RestIndexMonitorV2Action : BaseRestHandler() { + override fun getName(): String { + return "index_monitor_v2_action" + } + + override fun routes(): List { + return listOf( + Route( + POST, + AlertingPlugin.MONITOR_V2_BASE_URI + ), + Route( + PUT, + "${AlertingPlugin.MONITOR_V2_BASE_URI}/{monitorV2Id}" + ) + ) + } + + @Throws(IOException::class) + override fun prepareRequest(request: RestRequest, client: NodeClient): RestChannelConsumer { + log.debug("${request.method()} ${request.path()}") + + val xcp = request.contentParser() + ensureExpectedToken(Token.START_OBJECT, xcp.nextToken(), xcp) + + val monitorV2: MonitorV2 + val rbacRoles: List? + try { + monitorV2 = MonitorV2.parse(xcp) + rbacRoles = request.contentParser().map()["rbac_roles"] as List? + } catch (e: Exception) { + throw AlertingException.wrap(IllegalArgumentException(e.localizedMessage)) + } + + val id = request.param("monitorV2Id", MonitorV2.NO_ID) + val seqNo = request.paramAsLong(IF_SEQ_NO, SequenceNumbers.UNASSIGNED_SEQ_NO) + val primaryTerm = request.paramAsLong(IF_PRIMARY_TERM, SequenceNumbers.UNASSIGNED_PRIMARY_TERM) + val refreshPolicy = if (request.hasParam(REFRESH)) { + WriteRequest.RefreshPolicy.parse(request.param(REFRESH)) + } else { + WriteRequest.RefreshPolicy.IMMEDIATE + } + + val indexMonitorV2Request = IndexMonitorV2Request(id, seqNo, primaryTerm, refreshPolicy, request.method(), monitorV2, rbacRoles) + + return RestChannelConsumer { channel -> + client.execute(IndexMonitorV2Action.INSTANCE, indexMonitorV2Request, RestToXContentListener(channel)) + } + } +} diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/settings/AlertingSettings.kt b/alerting/src/main/kotlin/org/opensearch/alerting/settings/AlertingSettings.kt index 5a50ce632..d48552646 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/settings/AlertingSettings.kt +++ b/alerting/src/main/kotlin/org/opensearch/alerting/settings/AlertingSettings.kt @@ -293,5 +293,108 @@ class AlertingSettings { 0, Setting.Property.NodeScope, Setting.Property.Dynamic ) + + val ALERT_V2_HISTORY_ENABLED = Setting.boolSetting( + "plugins.alerting_v2.alert_history_enabled", + true, + Setting.Property.NodeScope, Setting.Property.Dynamic + ) + + val ALERT_V2_HISTORY_ROLLOVER_PERIOD = Setting.positiveTimeSetting( + "plugins.alerting_v2.alert_history_rollover_period", + TimeValue(12, TimeUnit.HOURS), + Setting.Property.NodeScope, Setting.Property.Dynamic + ) + + val ALERT_V2_HISTORY_INDEX_MAX_AGE = Setting.positiveTimeSetting( + "plugins.alerting_v2.alert_history_max_age", + TimeValue(30, TimeUnit.DAYS), + Setting.Property.NodeScope, Setting.Property.Dynamic + ) + + val ALERT_V2_HISTORY_MAX_DOCS = Setting.longSetting( + "plugins.alerting_v2.alert_history_max_docs", + 1000L, 0L, + Setting.Property.NodeScope, Setting.Property.Dynamic + ) + + val ALERT_V2_HISTORY_RETENTION_PERIOD = Setting.positiveTimeSetting( + "plugins.alerting_v2.alert_history_retention_period", + TimeValue(60, TimeUnit.DAYS), + Setting.Property.NodeScope, Setting.Property.Dynamic + ) + + val ALERTING_V2_MAX_MONITORS = Setting.intSetting( + "plugins.alerting_v2.monitor.max_monitors", + 1000, + 1, + Setting.Property.NodeScope, Setting.Property.Dynamic + ) + + val ALERTING_V2_MAX_THROTTLE_DURATION = Setting.longSetting( + "plugins.alerting_v2.monitor.max_throttle_duration", + 7200L, // 5 days, 7200 minutes + 2L, + Setting.Property.NodeScope, Setting.Property.Dynamic + ) + + val ALERTING_V2_MAX_EXPIRE_DURATION = Setting.longSetting( + "plugins.alerting_v2.monitor.max_expire_duration", + 43200L, // 30 days, 43200 minutes + 2L, + Setting.Property.NodeScope, Setting.Property.Dynamic + ) + + val ALERTING_V2_MAX_LOOK_BACK_WINDOW = Setting.longSetting( + "plugins.alerting_v2.monitor.max_look_back_window", + 10080L, // 7 days, 10080 minutes + 2L, + Setting.Property.NodeScope, Setting.Property.Dynamic + ) + + val ALERTING_V2_MAX_QUERY_LENGTH = Setting.longSetting( + "plugins.alerting_v2.monitor.max_query_length", + 2000L, + 0L, + Setting.Property.NodeScope, Setting.Property.Dynamic + ) + + // max data rows to retrieve when executing PPL query against + // SQL/PPL plugin during monitor execution + val ALERTING_V2_QUERY_RESULTS_MAX_DATAROWS = Setting.longSetting( + "plugins.alerting_v2.query_results_max_datarows", + 1000L, + 1L, + Setting.Property.NodeScope, Setting.Property.Dynamic + ) + + // max size of query results to store in alerts and notifications + val ALERT_V2_QUERY_RESULTS_MAX_SIZE = Setting.longSetting( + "plugins.alerting_v2.query_results_max_size", + 3000L, + 0L, + Setting.Property.NodeScope, Setting.Property.Dynamic + ) + + val ALERT_V2_PER_RESULT_TRIGGER_MAX_ALERTS = Setting.intSetting( + "plugins.alerting_v2.per_result_trigger_max_alerts", + 10, + 1, + Setting.Property.NodeScope, Setting.Property.Dynamic + ) + + val NOTIFICATION_SUBJECT_SOURCE_MAX_LENGTH = Setting.intSetting( + "plugins.alerting_v2.notification_subject_source_max_length", + 1000, + 100, + Setting.Property.NodeScope, Setting.Property.Dynamic + ) + + val NOTIFICATION_MESSAGE_SOURCE_MAX_LENGTH = Setting.intSetting( + "plugins.alerting_v2.notification_message_source_max_length", + 3000, + 1000, + Setting.Property.NodeScope, Setting.Property.Dynamic + ) } } diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportIndexMonitorAction.kt b/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportIndexMonitorAction.kt index c36fc4957..320b82ccc 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportIndexMonitorAction.kt +++ b/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportIndexMonitorAction.kt @@ -29,6 +29,7 @@ import org.opensearch.action.support.ActionFilters import org.opensearch.action.support.HandledTransportAction import org.opensearch.action.support.WriteRequest.RefreshPolicy import org.opensearch.action.support.clustermanager.AcknowledgedResponse +import org.opensearch.alerting.AlertingV2Utils.validateMonitorV1 import org.opensearch.alerting.MonitorMetadataService import org.opensearch.alerting.core.ScheduledJobIndices import org.opensearch.alerting.opensearchapi.suspendUntil @@ -614,7 +615,15 @@ class TransportIndexMonitorAction @Inject constructor( xContentRegistry, LoggingDeprecationHandler.INSTANCE, getResponse.sourceAsBytesRef, XContentType.JSON ) - val monitor = ScheduledJob.parse(xcp, getResponse.id, getResponse.version) as Monitor + val scheduledJob = ScheduledJob.parse(xcp, getResponse.id, getResponse.version) + + validateMonitorV1(scheduledJob)?.let { + actionListener.onFailure(AlertingException.wrap(it)) + return + } + + val monitor = scheduledJob as Monitor + onGetResponse(monitor) } catch (t: Exception) { actionListener.onFailure(AlertingException.wrap(t)) diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportIndexWorkflowAction.kt b/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportIndexWorkflowAction.kt index 3c80af129..ba1e17c57 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportIndexWorkflowAction.kt +++ b/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportIndexWorkflowAction.kt @@ -27,6 +27,7 @@ import org.opensearch.action.search.SearchResponse import org.opensearch.action.support.ActionFilters import org.opensearch.action.support.HandledTransportAction import org.opensearch.action.support.clustermanager.AcknowledgedResponse +import org.opensearch.alerting.AlertingV2Utils.validateMonitorV1 import org.opensearch.alerting.MonitorMetadataService import org.opensearch.alerting.MonitorRunnerService.monitorCtx import org.opensearch.alerting.WorkflowMetadataService @@ -442,7 +443,12 @@ class TransportIndexWorkflowAction @Inject constructor( xContentRegistry, LoggingDeprecationHandler.INSTANCE, getResponse.sourceAsBytesRef, XContentType.JSON ) - val workflow = ScheduledJob.parse(xcp, getResponse.id, getResponse.version) as Workflow + val scheduledJob = ScheduledJob.parse(xcp, getResponse.id, getResponse.version) + validateMonitorV1(scheduledJob)?.let { + actionListener.onFailure(AlertingException.wrap(it)) + return + } + val workflow = scheduledJob as Workflow onGetResponse(workflow) } catch (t: Exception) { actionListener.onFailure(AlertingException.wrap(t)) @@ -454,7 +460,7 @@ class TransportIndexWorkflowAction @Inject constructor( user, currentWorkflow.user, actionListener, - "workfklow", + "workflow", request.workflowId ) ) { @@ -715,7 +721,11 @@ class TransportIndexWorkflowAction @Inject constructor( xContentRegistry, LoggingDeprecationHandler.INSTANCE, hit.sourceAsString ).use { hitsParser -> - val monitor = ScheduledJob.parse(hitsParser, hit.id, hit.version) as Monitor + val scheduledJob = ScheduledJob.parse(hitsParser, hit.id, hit.version) + validateMonitorV1(scheduledJob)?.let { + throw OpenSearchException(it) + } + val monitor = scheduledJob as Monitor monitors.add(monitor) } } diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/transportv2/TransportIndexMonitorV2Action.kt b/alerting/src/main/kotlin/org/opensearch/alerting/transportv2/TransportIndexMonitorV2Action.kt new file mode 100644 index 000000000..a4ca63177 --- /dev/null +++ b/alerting/src/main/kotlin/org/opensearch/alerting/transportv2/TransportIndexMonitorV2Action.kt @@ -0,0 +1,841 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.alerting.transportv2 + +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.launch +import kotlinx.coroutines.newSingleThreadContext +import kotlinx.coroutines.withContext +import org.apache.logging.log4j.LogManager +import org.opensearch.ExceptionsHelper +import org.opensearch.OpenSearchException +import org.opensearch.OpenSearchStatusException +import org.opensearch.ResourceAlreadyExistsException +import org.opensearch.action.admin.cluster.health.ClusterHealthAction +import org.opensearch.action.admin.cluster.health.ClusterHealthRequest +import org.opensearch.action.admin.cluster.health.ClusterHealthResponse +import org.opensearch.action.admin.indices.create.CreateIndexResponse +import org.opensearch.action.admin.indices.mapping.get.GetMappingsRequest +import org.opensearch.action.get.GetRequest +import org.opensearch.action.get.GetResponse +import org.opensearch.action.index.IndexRequest +import org.opensearch.action.index.IndexResponse +import org.opensearch.action.search.SearchRequest +import org.opensearch.action.search.SearchResponse +import org.opensearch.action.support.ActionFilters +import org.opensearch.action.support.HandledTransportAction +import org.opensearch.action.support.clustermanager.AcknowledgedResponse +import org.opensearch.alerting.AlertingV2Utils.validateMonitorV2 +import org.opensearch.alerting.PPLUtils.appendCustomCondition +import org.opensearch.alerting.PPLUtils.executePplQuery +import org.opensearch.alerting.PPLUtils.findEvalResultVar +import org.opensearch.alerting.PPLUtils.findEvalResultVarIdxInSchema +import org.opensearch.alerting.PPLUtils.getIndicesFromPplQuery +import org.opensearch.alerting.actionv2.IndexMonitorV2Action +import org.opensearch.alerting.actionv2.IndexMonitorV2Request +import org.opensearch.alerting.actionv2.IndexMonitorV2Response +import org.opensearch.alerting.core.ScheduledJobIndices +import org.opensearch.alerting.modelv2.MonitorV2 +import org.opensearch.alerting.modelv2.MonitorV2.Companion.MONITOR_V2_TYPE +import org.opensearch.alerting.modelv2.PPLSQLMonitor +import org.opensearch.alerting.modelv2.PPLSQLTrigger.ConditionType +import org.opensearch.alerting.opensearchapi.suspendUntil +import org.opensearch.alerting.settings.AlertingSettings +import org.opensearch.alerting.settings.AlertingSettings.Companion.ALERTING_V2_MAX_EXPIRE_DURATION +import org.opensearch.alerting.settings.AlertingSettings.Companion.ALERTING_V2_MAX_LOOK_BACK_WINDOW +import org.opensearch.alerting.settings.AlertingSettings.Companion.ALERTING_V2_MAX_MONITORS +import org.opensearch.alerting.settings.AlertingSettings.Companion.ALERTING_V2_MAX_QUERY_LENGTH +import org.opensearch.alerting.settings.AlertingSettings.Companion.ALERTING_V2_MAX_THROTTLE_DURATION +import org.opensearch.alerting.settings.AlertingSettings.Companion.ALERTING_V2_QUERY_RESULTS_MAX_DATAROWS +import org.opensearch.alerting.settings.AlertingSettings.Companion.INDEX_TIMEOUT +import org.opensearch.alerting.settings.AlertingSettings.Companion.NOTIFICATION_MESSAGE_SOURCE_MAX_LENGTH +import org.opensearch.alerting.settings.AlertingSettings.Companion.NOTIFICATION_SUBJECT_SOURCE_MAX_LENGTH +import org.opensearch.alerting.settings.AlertingSettings.Companion.REQUEST_TIMEOUT +import org.opensearch.alerting.transport.SecureTransportAction +import org.opensearch.alerting.util.IndexUtils +import org.opensearch.alerting.util.use +import org.opensearch.cluster.service.ClusterService +import org.opensearch.common.inject.Inject +import org.opensearch.common.settings.Settings +import org.opensearch.common.xcontent.LoggingDeprecationHandler +import org.opensearch.common.xcontent.XContentFactory.jsonBuilder +import org.opensearch.common.xcontent.XContentHelper +import org.opensearch.common.xcontent.XContentType +import org.opensearch.commons.alerting.model.ScheduledJob +import org.opensearch.commons.alerting.model.ScheduledJob.Companion.SCHEDULED_JOBS_INDEX +import org.opensearch.commons.alerting.model.userErrorMessage +import org.opensearch.commons.alerting.util.AlertingException +import org.opensearch.commons.authuser.User +import org.opensearch.core.action.ActionListener +import org.opensearch.core.common.io.stream.NamedWriteableRegistry +import org.opensearch.core.rest.RestStatus +import org.opensearch.core.xcontent.NamedXContentRegistry +import org.opensearch.core.xcontent.ToXContent +import org.opensearch.index.query.QueryBuilders +import org.opensearch.rest.RestRequest +import org.opensearch.search.builder.SearchSourceBuilder +import org.opensearch.tasks.Task +import org.opensearch.transport.TransportService +import org.opensearch.transport.client.Client +import org.opensearch.transport.client.node.NodeClient + +private val log = LogManager.getLogger(TransportIndexMonitorV2Action::class.java) +private val scope: CoroutineScope = CoroutineScope(Dispatchers.IO) + +class TransportIndexMonitorV2Action @Inject constructor( + transportService: TransportService, + val client: Client, + actionFilters: ActionFilters, + val scheduledJobIndices: ScheduledJobIndices, + val clusterService: ClusterService, + val settings: Settings, + val xContentRegistry: NamedXContentRegistry, + val namedWriteableRegistry: NamedWriteableRegistry, +) : HandledTransportAction( + IndexMonitorV2Action.NAME, transportService, actionFilters, ::IndexMonitorV2Request +), + SecureTransportAction { + + // adjustable limits (via settings) + @Volatile private var maxMonitors = ALERTING_V2_MAX_MONITORS.get(settings) + @Volatile private var maxThrottleDuration = ALERTING_V2_MAX_THROTTLE_DURATION.get(settings) + @Volatile private var maxExpireDuration = ALERTING_V2_MAX_EXPIRE_DURATION.get(settings) + @Volatile private var maxLookBackWindow = ALERTING_V2_MAX_LOOK_BACK_WINDOW.get(settings) + @Volatile private var maxQueryLength = ALERTING_V2_MAX_QUERY_LENGTH.get(settings) + @Volatile private var maxQueryResults = ALERTING_V2_QUERY_RESULTS_MAX_DATAROWS.get(settings) + @Volatile private var notificationSubjectMaxLength = NOTIFICATION_SUBJECT_SOURCE_MAX_LENGTH.get(settings) + @Volatile private var notificationMessageMaxLength = NOTIFICATION_MESSAGE_SOURCE_MAX_LENGTH.get(settings) + @Volatile private var requestTimeout = REQUEST_TIMEOUT.get(settings) + @Volatile private var indexTimeout = INDEX_TIMEOUT.get(settings) + @Volatile override var filterByEnabled = AlertingSettings.FILTER_BY_BACKEND_ROLES.get(settings) + + init { + clusterService.clusterSettings.addSettingsUpdateConsumer(ALERTING_V2_MAX_MONITORS) { maxMonitors = it } + clusterService.clusterSettings.addSettingsUpdateConsumer(ALERTING_V2_MAX_THROTTLE_DURATION) { maxThrottleDuration = it } + clusterService.clusterSettings.addSettingsUpdateConsumer(ALERTING_V2_MAX_EXPIRE_DURATION) { maxExpireDuration = it } + clusterService.clusterSettings.addSettingsUpdateConsumer(ALERTING_V2_MAX_LOOK_BACK_WINDOW) { maxLookBackWindow = it } + clusterService.clusterSettings.addSettingsUpdateConsumer(ALERTING_V2_MAX_QUERY_LENGTH) { maxQueryLength = it } + clusterService.clusterSettings.addSettingsUpdateConsumer(ALERTING_V2_QUERY_RESULTS_MAX_DATAROWS) { maxQueryResults = it } + clusterService.clusterSettings.addSettingsUpdateConsumer(NOTIFICATION_SUBJECT_SOURCE_MAX_LENGTH) { + notificationSubjectMaxLength = it + } + clusterService.clusterSettings.addSettingsUpdateConsumer(NOTIFICATION_MESSAGE_SOURCE_MAX_LENGTH) { + notificationMessageMaxLength = it + } + clusterService.clusterSettings.addSettingsUpdateConsumer(REQUEST_TIMEOUT) { requestTimeout = it } + clusterService.clusterSettings.addSettingsUpdateConsumer(INDEX_TIMEOUT) { indexTimeout = it } + listenFilterBySettingChange(clusterService) + } + + override fun doExecute( + task: Task, + indexMonitorV2Request: IndexMonitorV2Request, + actionListener: ActionListener + ) { + // read the user from thread context immediately, before + // downstream flows spin up new threads with fresh context + val user = readUserFromThreadContext(client) + + // validate the MonitorV2 based on its type + when (indexMonitorV2Request.monitorV2) { + is PPLSQLMonitor -> validatePplSqlMonitorUserPermissionsAndQuery( + indexMonitorV2Request, + user, + object : ActionListener { // validationListener + override fun onResponse(response: Unit) { + // user permissions to indices have already been checked + // proceed without the context of the user, otherwise, + // we would get permissions errors trying to search the alerting-config + // index as the user. pass the user object itself so backend + // roles can be matched and checked downstream + client.threadPool().threadContext.stashContext().use { + val pplSqlMonitor = indexMonitorV2Request.monitorV2 as PPLSQLMonitor + if (user == null) { + indexMonitorV2Request.monitorV2 = pplSqlMonitor + .copy(user = User("", listOf(), listOf(), mapOf())) + } else { + indexMonitorV2Request.monitorV2 = pplSqlMonitor + .copy(user = User(user.name, user.backendRoles, user.roles, user.customAttributes)) + } + checkScheduledJobIndex(indexMonitorV2Request, actionListener, user) + } + } + + override fun onFailure(e: Exception) { + actionListener.onFailure(e) + } + } + ) + else -> actionListener.onFailure( + AlertingException.wrap( + IllegalStateException( + "unexpected MonitorV2 type: ${indexMonitorV2Request.monitorV2.javaClass.name}" + ) + ) + ) + } + } + + // validates the PPL Monitor, its query, and user's permissions to the indices it queries by submitting it to SQL/PPL plugin + private fun validatePplSqlMonitorUserPermissionsAndQuery( + indexMonitorV2Request: IndexMonitorV2Request, + user: User?, + validationListener: ActionListener + ) { + client.threadPool().threadContext.stashContext().use { + scope.launch { + val singleThreadContext = newSingleThreadContext("IndexMonitorV2ActionThread") + withContext(singleThreadContext) { + it.restore() + + val pplSqlMonitor = indexMonitorV2Request.monitorV2 as PPLSQLMonitor + + val pplQueryValid = validatePplSqlQuery(pplSqlMonitor, validationListener) + if (!pplQueryValid) { + return@withContext + } + + // run basic validations against the PPL/SQL Monitor + val pplSqlMonitorValid = validatePplSqlMonitor(pplSqlMonitor, validationListener) + if (!pplSqlMonitorValid) { + return@withContext + } + + // check the user for basic permissions + val userHasPermissions = checkUser(user, indexMonitorV2Request, validationListener) + if (!userHasPermissions) { + return@withContext + } + + // check that given timestamp field is valid + val timestampFieldValid = checkPplQueryIndicesForTimestampField(pplSqlMonitor, validationListener) + if (!timestampFieldValid) { + return@withContext + } + + validationListener.onResponse(Unit) + } + } + } + } + + private suspend fun validatePplSqlQuery(pplSqlMonitor: PPLSQLMonitor, validationListener: ActionListener): Boolean { + // first attempt to run the monitor query and all possible + // extensions of it (from custom conditions) + try { + val nodeClient = client as NodeClient + + // first run the base query as is. + // if there are any PPL syntax or index not found or other errors, + // this will throw an exception + executePplQuery(pplSqlMonitor.query, nodeClient) + + // now scan all the triggers with custom conditions, and ensure each query constructed + // from the base query + custom condition is valid + for (pplTrigger in pplSqlMonitor.triggers) { + if (pplTrigger.conditionType != ConditionType.CUSTOM) { + continue + } + + val evalResultVar = findEvalResultVar(pplTrigger.customCondition!!) + + val queryWithCustomCondition = appendCustomCondition(pplSqlMonitor.query, pplTrigger.customCondition!!) + + val executePplQueryResponse = executePplQuery(queryWithCustomCondition, nodeClient) + + val evalResultVarIdx = findEvalResultVarIdxInSchema(executePplQueryResponse, evalResultVar) + + val resultVarType = executePplQueryResponse + .getJSONArray("schema") + .getJSONObject(evalResultVarIdx) + .getString("type") + + // custom conditions must evaluate to a boolean result, otherwise it's invalid + if (resultVarType != "boolean") { + validationListener.onFailure( + AlertingException.wrap( + IllegalArgumentException( + "Custom condition in trigger ${pplTrigger.name} is invalid because it does not " + + "evaluate to a boolean, but instead to type: $resultVarType" + ) + ) + ) + return false + } + } + } catch (e: Exception) { + validationListener.onFailure( + AlertingException.wrap( + IllegalArgumentException("Validation error for PPL Query in PPL Monitor: ${e.userErrorMessage()}") + ) + ) + return false + } + + return true + } + + private fun validatePplSqlMonitor(pplSqlMonitor: PPLSQLMonitor, validationListener: ActionListener): Boolean { + // ensure the trigger throttle and expire durations are valid + pplSqlMonitor.triggers.forEach { trigger -> + trigger.throttleDuration?.let { throttleDuration -> + if (throttleDuration > maxThrottleDuration) { + validationListener.onFailure( + AlertingException.wrap( + IllegalArgumentException( + "Throttle duration must be at most $maxThrottleDuration but was $throttleDuration" + ) + ) + ) + return false + } + } + + if (trigger.expireDuration > maxExpireDuration) { + validationListener.onFailure( + AlertingException.wrap( + IllegalArgumentException( + "Expire duration must be at most $maxExpireDuration but was ${trigger.expireDuration}" + ) + ) + ) + return false + } + + if (trigger.conditionType == ConditionType.NUMBER_OF_RESULTS && + trigger.numResultsValue!! > maxQueryResults + ) { + validationListener.onFailure( + AlertingException.wrap( + IllegalArgumentException( + "Trigger ${trigger.id} checks for number of results threshold of ${trigger.numResultsValue}, " + + "but Alerting V2 is configured only to retrieve $maxQueryResults query results maximum. " + + "Please lower the number of results value to one below this maximum value, or adjust the cluster " + + "setting: $ALERTING_V2_QUERY_RESULTS_MAX_DATAROWS.key}" + ) + ) + ) + return false + } + + trigger.actions.forEach { action -> + if (action.subjectTemplate?.idOrCode?.length!! > notificationSubjectMaxLength) { + validationListener.onFailure( + AlertingException.wrap( + IllegalArgumentException( + "Notification subject source cannot exceed length: $notificationSubjectMaxLength" + ) + ) + ) + return false + } + + if (action.messageTemplate.idOrCode.length > notificationMessageMaxLength) { + validationListener.onFailure( + AlertingException.wrap( + IllegalArgumentException( + "Notification message source cannot exceed length: $notificationMessageMaxLength" + ) + ) + ) + return false + } + } + } + + // ensure the query length doesn't exceed the limit + if (pplSqlMonitor.query.length > maxQueryLength) { + validationListener.onFailure( + AlertingException.wrap( + IllegalArgumentException( + "PPL Query length must be at most $maxQueryLength but was ${pplSqlMonitor.query.length}" + ) + ) + ) + return false + } + + // ensure the look back window doesn't exceed the limit + pplSqlMonitor.lookBackWindow?.let { + if (pplSqlMonitor.lookBackWindow > maxLookBackWindow) { + validationListener.onFailure( + AlertingException.wrap( + IllegalArgumentException( + "Look back window must be at most $maxLookBackWindow minutes but was ${pplSqlMonitor.lookBackWindow}" + ) + ) + ) + return false + } + } + + return true + } + + private fun checkUser( + user: User?, + indexMonitorV2Request: IndexMonitorV2Request, + validationListener: ActionListener + ): Boolean { + /* check initial user permissions */ + if (!validateUserBackendRoles(user, validationListener)) { + return false + } + + if ( + user != null && + !isAdmin(user) && + indexMonitorV2Request.rbacRoles != null + ) { + if (indexMonitorV2Request.rbacRoles.stream().anyMatch { !user.backendRoles.contains(it) }) { + log.debug( + "User specified backend roles, ${indexMonitorV2Request.rbacRoles}, " + + "that they don't have access to. User backend roles: ${user.backendRoles}" + ) + validationListener.onFailure( + AlertingException.wrap( + OpenSearchStatusException( + "User specified backend roles that they don't have access to. Contact administrator", RestStatus.FORBIDDEN + ) + ) + ) + return false + } else if (indexMonitorV2Request.rbacRoles.isEmpty()) { + log.debug( + "Non-admin user are not allowed to specify an empty set of backend roles. " + + "Please don't pass in the parameter or pass in at least one backend role." + ) + validationListener.onFailure( + AlertingException.wrap( + OpenSearchStatusException( + "Non-admin user are not allowed to specify an empty set of backend roles.", RestStatus.FORBIDDEN + ) + ) + ) + return false + } + } + + return true + } + + // if look back window is specified, all the indices that the PPL query searches + // must contain the timestamp field specified in the PPL Monitor, and they must + // all be of OpenSearch data type "date" + private suspend fun checkPplQueryIndicesForTimestampField( + pplSqlMonitor: PPLSQLMonitor, + validationListener: ActionListener + ): Boolean { + if (pplSqlMonitor.lookBackWindow == null) { + // if no look back window was specified, no need + // to check for timestamp field in PPL query indices + return true + } + + val pplQuery = pplSqlMonitor.query + val timestampField = pplSqlMonitor.timestampField + + val indices = getIndicesFromPplQuery(pplQuery) + val getMappingsRequest = GetMappingsRequest().indices(*indices.toTypedArray()) + val getMappingsResponse = client.suspendUntil { admin().indices().getMappings(getMappingsRequest, it) } + + val metadataMap = getMappingsResponse.mappings + try { + for (index in metadataMap.keys) { + val metadata = metadataMap[index]!!.sourceAsMap["properties"] as Map + if (!metadata.keys.contains(timestampField)) { + validationListener.onFailure( + AlertingException.wrap( + IllegalArgumentException("Query index $index don't contain given timestamp field: $timestampField") + ) + ) + return false + } + val typeInfo = metadata[timestampField] as Map + val type = typeInfo["type"] + if (type != "date") { + validationListener.onFailure( + AlertingException.wrap( + IllegalArgumentException( + "Timestamp field: $timestampField is present in index $index but is of type $type instead of type date" + ) + ) + ) + return false + } + } + } catch (e: Exception) { + log.error("failed to read query indices' fields when checking for timestamp field: $timestampField") + validationListener.onFailure( + AlertingException.wrap( + IllegalArgumentException("failed to read query indices' fields when checking for timestamp field: $timestampField", e) + ) + ) + return false + } + + return true + } + + private fun checkScheduledJobIndex( + indexMonitorRequest: IndexMonitorV2Request, + actionListener: ActionListener, + user: User? + ) { + // user permissions to indices have already been checked + // proceed without the context of the user, otherwise, + // we would get permissions errors trying to search the alerting-config + // index as the user + client.threadPool().threadContext.stashContext().use { + /* check to see if alerting-config index (scheduled job index) is created and updated before indexing MonitorV2 into it */ + if (!scheduledJobIndices.scheduledJobIndexExists()) { // if alerting-config index doesn't exist, send request to create it + scheduledJobIndices.initScheduledJobIndex(object : ActionListener { + override fun onResponse(response: CreateIndexResponse) { + onCreateMappingsResponse(response.isAcknowledged, indexMonitorRequest, actionListener, user) + } + + override fun onFailure(e: Exception) { + if (ExceptionsHelper.unwrapCause(e) is ResourceAlreadyExistsException) { + scope.launch { + // Wait for the yellow status + val clusterHealthRequest = ClusterHealthRequest() + .indices(SCHEDULED_JOBS_INDEX) + .waitForYellowStatus() + val response: ClusterHealthResponse = client.suspendUntil { + execute(ClusterHealthAction.INSTANCE, clusterHealthRequest, it) + } + if (response.isTimedOut) { + actionListener.onFailure( + OpenSearchException("Cannot determine that the $SCHEDULED_JOBS_INDEX index is healthy") + ) + } + // Retry mapping of monitor + onCreateMappingsResponse(true, indexMonitorRequest, actionListener, user) + } + } else { + actionListener.onFailure(AlertingException.wrap(e)) + } + } + }) + } else if (!IndexUtils.scheduledJobIndexUpdated) { + IndexUtils.updateIndexMapping( + SCHEDULED_JOBS_INDEX, + ScheduledJobIndices.scheduledJobMappings(), clusterService.state(), client.admin().indices(), + object : ActionListener { + override fun onResponse(response: AcknowledgedResponse) { + onUpdateMappingsResponse(response, indexMonitorRequest, actionListener, user) + } + override fun onFailure(t: Exception) { + actionListener.onFailure(AlertingException.wrap(t)) + } + } + ) + } else { + prepareMonitorIndexing(indexMonitorRequest, actionListener, user) + } + } + } + + private fun onCreateMappingsResponse( + isAcknowledged: Boolean, + request: IndexMonitorV2Request, + actionListener: ActionListener, + user: User? + ) { + if (isAcknowledged) { + log.info("Created $SCHEDULED_JOBS_INDEX with mappings.") + prepareMonitorIndexing(request, actionListener, user) + IndexUtils.scheduledJobIndexUpdated() + } else { + log.info("Create $SCHEDULED_JOBS_INDEX mappings call not acknowledged.") + actionListener.onFailure( + AlertingException.wrap( + OpenSearchStatusException( + "Create $SCHEDULED_JOBS_INDEX mappings call not acknowledged", RestStatus.INTERNAL_SERVER_ERROR + ) + ) + ) + } + } + + private fun onUpdateMappingsResponse( + response: AcknowledgedResponse, + indexMonitorRequest: IndexMonitorV2Request, + actionListener: ActionListener, + user: User? + ) { + if (response.isAcknowledged) { + log.info("Updated $SCHEDULED_JOBS_INDEX with mappings.") + IndexUtils.scheduledJobIndexUpdated() + prepareMonitorIndexing(indexMonitorRequest, actionListener, user) + } else { + log.info("Update $SCHEDULED_JOBS_INDEX mappings call not acknowledged.") + actionListener.onFailure( + AlertingException.wrap( + OpenSearchStatusException( + "Updated $SCHEDULED_JOBS_INDEX mappings call not acknowledged.", + RestStatus.INTERNAL_SERVER_ERROR + ) + ) + ) + } + } + + private fun prepareMonitorIndexing( + indexMonitorRequest: IndexMonitorV2Request, + actionListener: ActionListener, + user: User? + ) { + if (indexMonitorRequest.method == RestRequest.Method.PUT) { // update monitor case + scope.launch { + updateMonitor(indexMonitorRequest, actionListener, user) + } + } else { // create monitor case + val query = QueryBuilders.boolQuery().filter(QueryBuilders.existsQuery(MONITOR_V2_TYPE)) + val searchSource = SearchSourceBuilder().query(query).timeout(requestTimeout) + val searchRequest = SearchRequest(SCHEDULED_JOBS_INDEX).source(searchSource) + + client.search( + searchRequest, + object : ActionListener { + override fun onResponse(searchResponse: SearchResponse) { + onMonitorCountSearchResponse(searchResponse, indexMonitorRequest, actionListener, user) + } + + override fun onFailure(t: Exception) { + actionListener.onFailure(AlertingException.wrap(t)) + } + } + ) + } + } + + /* Functions for Update Monitor flow */ + + private suspend fun updateMonitor( + indexMonitorRequest: IndexMonitorV2Request, + actionListener: ActionListener, + user: User? + ) { + val getRequest = GetRequest(SCHEDULED_JOBS_INDEX, indexMonitorRequest.monitorId) + try { + val getResponse: GetResponse = client.suspendUntil { client.get(getRequest, it) } + if (!getResponse.isExists) { + actionListener.onFailure( + AlertingException.wrap( + OpenSearchStatusException("MonitorV2 with ${indexMonitorRequest.monitorId} is not found", RestStatus.NOT_FOUND) + ) + ) + return + } + val xcp = XContentHelper.createParser( + xContentRegistry, LoggingDeprecationHandler.INSTANCE, + getResponse.sourceAsBytesRef, XContentType.JSON + ) + val scheduledJob = ScheduledJob.parse(xcp, getResponse.id, getResponse.version) + + validateMonitorV2(scheduledJob)?.let { + actionListener.onFailure(AlertingException.wrap(it)) + return + } + + val monitorV2 = scheduledJob as MonitorV2 + + onGetMonitorResponseForUpdate(monitorV2, indexMonitorRequest, actionListener, user) + } catch (e: Exception) { + actionListener.onFailure(AlertingException.wrap(e)) + } + } + + private suspend fun onGetMonitorResponseForUpdate( + existingMonitorV2: MonitorV2, + indexMonitorRequest: IndexMonitorV2Request, + actionListener: ActionListener, + user: User? + ) { + log.info("user: $user") + log.info("monitor user: ${existingMonitorV2.user}") + if ( + !checkUserPermissionsWithResource( + user, + existingMonitorV2.user, + actionListener, + "monitor_v2", + indexMonitorRequest.monitorId + ) + ) { + return + } + + var newMonitorV2 = indexMonitorRequest.monitorV2 + + // If both are enabled, use the current existing monitor enabled time, + // otherwise the next execution will be incorrect. + if (newMonitorV2.enabled && existingMonitorV2.enabled) { + newMonitorV2 = newMonitorV2.makeCopy(enabledTime = existingMonitorV2.enabledTime) + } + + /** + * On update monitor check which backend roles to associate to the monitor. + * Below are 2 examples of how the logic works + * + * Example 1, say we have a Monitor with backend roles [a, b, c, d] associated with it. + * If I'm User A (non-admin user) and I have backend roles [a, b, c] associated with me and I make a request to update + * the Monitor's backend roles to [a, b]. This would mean that the roles to remove are [c] and the roles to add are [a, b]. + * The Monitor's backend roles would then be [a, b, d]. + * + * Example 2, say we have a Monitor with backend roles [a, b, c, d] associated with it. + * If I'm User A (admin user) and I have backend roles [a, b, c] associated with me and I make a request to update + * the Monitor's backend roles to [a, b]. This would mean that the roles to remove are [c, d] and the roles to add are [a, b]. + * The Monitor's backend roles would then be [a, b]. + */ + if (user != null) { + if (indexMonitorRequest.rbacRoles != null) { + if (isAdmin(user)) { + newMonitorV2 = newMonitorV2.makeCopy( + user = User(user.name, indexMonitorRequest.rbacRoles, user.roles, user.customAttributes) + ) + } else { + // rolesToRemove: these are the backend roles to remove from the monitor + val rolesToRemove = user.backendRoles - indexMonitorRequest.rbacRoles + // remove the monitor's roles with rolesToRemove and add any roles passed into the request.rbacRoles + val updatedRbac = existingMonitorV2.user?.backendRoles.orEmpty() - rolesToRemove + indexMonitorRequest.rbacRoles + newMonitorV2 = newMonitorV2.makeCopy( + user = User(user.name, updatedRbac, user.roles, user.customAttributes) + ) + } + } else { + newMonitorV2 = newMonitorV2 + .makeCopy(user = User(user.name, existingMonitorV2.user!!.backendRoles, user.roles, user.customAttributes)) + } + log.info("Update monitor backend roles to: ${newMonitorV2.user?.backendRoles}") + } + + newMonitorV2 = newMonitorV2.makeCopy(schemaVersion = IndexUtils.scheduledJobIndexSchemaVersion) + val indexRequest = IndexRequest(SCHEDULED_JOBS_INDEX) + .setRefreshPolicy(indexMonitorRequest.refreshPolicy) + .source(newMonitorV2.toXContentWithUser(jsonBuilder(), ToXContent.MapParams(mapOf("with_type" to "true")))) + .id(indexMonitorRequest.monitorId) + .routing(indexMonitorRequest.monitorId) + .setIfSeqNo(indexMonitorRequest.seqNo) + .setIfPrimaryTerm(indexMonitorRequest.primaryTerm) + .timeout(indexTimeout) + + log.info( + "Updating monitor, ${existingMonitorV2.id}, from: ${existingMonitorV2.toXContentWithUser( + jsonBuilder(), + ToXContent.MapParams(mapOf("with_type" to "true")) + )} \n to: ${newMonitorV2.toXContentWithUser(jsonBuilder(), ToXContent.MapParams(mapOf("with_type" to "true")))}" + ) + + try { + val indexResponse: IndexResponse = client.suspendUntil { client.index(indexRequest, it) } + val failureReasons = IndexUtils.checkShardsFailure(indexResponse) + if (failureReasons != null) { + actionListener.onFailure( + AlertingException.wrap(OpenSearchStatusException(failureReasons.toString(), indexResponse.status())) + ) + return + } + + actionListener.onResponse( + IndexMonitorV2Response( + indexResponse.id, indexResponse.version, indexResponse.seqNo, + indexResponse.primaryTerm, newMonitorV2 + ) + ) + } catch (e: Exception) { + actionListener.onFailure(AlertingException.wrap(e)) + } + } + + /* Functions for Create Monitor flow */ + + /** + * After searching for all existing monitors we validate the system can support another monitor to be created. + */ + private fun onMonitorCountSearchResponse( + monitorCountSearchResponse: SearchResponse, + indexMonitorRequest: IndexMonitorV2Request, + actionListener: ActionListener, + user: User? + ) { + val totalHits = monitorCountSearchResponse.hits.totalHits?.value + if (totalHits != null && totalHits >= maxMonitors) { + log.info("This request would create more than the allowed monitors [$maxMonitors].") + actionListener.onFailure( + AlertingException.wrap( + IllegalArgumentException( + "This request would create more than the allowed monitors [$maxMonitors]." + ) + ) + ) + } else { + scope.launch { + indexMonitor(indexMonitorRequest, actionListener, user) + } + } + } + + private suspend fun indexMonitor( + indexMonitorRequest: IndexMonitorV2Request, + actionListener: ActionListener, + user: User? + ) { + var monitorV2 = indexMonitorRequest.monitorV2 + + if (user != null) { + // Use the backend roles which is an intersection of the requested backend roles and the user's backend roles. + // Admins can pass in any backend role. Also if no backend role is passed in, all the user's backend roles are used. + val rbacRoles = if (indexMonitorRequest.rbacRoles == null) user.backendRoles.toSet() + else if (!isAdmin(user)) indexMonitorRequest.rbacRoles.intersect(user.backendRoles).toSet() + else indexMonitorRequest.rbacRoles + + monitorV2 = monitorV2.makeCopy( + user = User(user.name, rbacRoles.toList(), user.roles, user.customAttributes) + ) + + log.debug("Created monitor's backend roles: $rbacRoles") + } + + val indexRequest = IndexRequest(SCHEDULED_JOBS_INDEX) + .setRefreshPolicy(indexMonitorRequest.refreshPolicy) + .source(monitorV2.toXContentWithUser(jsonBuilder(), ToXContent.MapParams(mapOf("with_type" to "true")))) + .routing(indexMonitorRequest.monitorId) + .setIfSeqNo(indexMonitorRequest.seqNo) + .setIfPrimaryTerm(indexMonitorRequest.primaryTerm) + .timeout(indexTimeout) + + log.info( + "Creating new monitorV2: ${monitorV2.toXContentWithUser( + jsonBuilder(), + ToXContent.MapParams(mapOf("with_type" to "true")) + )}" + ) + + try { + val indexResponse: IndexResponse = client.suspendUntil { client.index(indexRequest, it) } + val failureReasons = IndexUtils.checkShardsFailure(indexResponse) + if (failureReasons != null) { + log.info(failureReasons.toString()) + actionListener.onFailure( + AlertingException.wrap(OpenSearchStatusException(failureReasons.toString(), indexResponse.status())) + ) + return + } + + actionListener.onResponse( + IndexMonitorV2Response( + indexResponse.id, indexResponse.version, indexResponse.seqNo, + indexResponse.primaryTerm, monitorV2 + ) + ) + } catch (t: Exception) { + actionListener.onFailure(AlertingException.wrap(t)) + } + } +} diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/util/IndexUtils.kt b/alerting/src/main/kotlin/org/opensearch/alerting/util/IndexUtils.kt index 093b0bd39..994293f1d 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/util/IndexUtils.kt +++ b/alerting/src/main/kotlin/org/opensearch/alerting/util/IndexUtils.kt @@ -6,6 +6,7 @@ package org.opensearch.alerting.util import org.opensearch.action.admin.indices.mapping.put.PutMappingRequest +import org.opensearch.action.index.IndexResponse import org.opensearch.action.support.IndicesOptions import org.opensearch.action.support.clustermanager.AcknowledgedResponse import org.opensearch.alerting.alerts.AlertIndices @@ -47,6 +48,7 @@ class IndexUtils { private set var commentsIndexUpdated: Boolean = false private set + var lastUpdatedAlertHistoryIndex: String? = null var lastUpdatedFindingHistoryIndex: String? = null var lastUpdatedCommentsHistoryIndex: String? = null @@ -205,5 +207,18 @@ class IndexUtils { fun getCreationDateForIndex(index: String, clusterState: ClusterState): Long { return clusterState.metadata.index(index).creationDate } + + @JvmStatic + fun checkShardsFailure(response: IndexResponse): String? { + val failureReasons = StringBuilder() + if (response.shardInfo.failed > 0) { + response.shardInfo.failures.forEach { + entry -> + failureReasons.append(entry.reason()) + } + return failureReasons.toString() + } + return null + } } } diff --git a/alerting/src/test/kotlin/org/opensearch/alerting/AccessRoles.kt b/alerting/src/test/kotlin/org/opensearch/alerting/AccessRoles.kt index d14473884..133504168 100644 --- a/alerting/src/test/kotlin/org/opensearch/alerting/AccessRoles.kt +++ b/alerting/src/test/kotlin/org/opensearch/alerting/AccessRoles.kt @@ -9,6 +9,7 @@ import org.opensearch.alerting.action.ExecuteWorkflowAction import org.opensearch.commons.alerting.action.AlertingActions val ALL_ACCESS_ROLE = "all_access" +val PPL_FULL_ACCESS_ROLE = "ppl_full_access" val READALL_AND_MONITOR_ROLE = "readall_and_monitor" val ALERTING_FULL_ACCESS_ROLE = "alerting_full_access" val ALERTING_ACK_ALERTS_ROLE = "alerting_ack_alerts" @@ -31,6 +32,7 @@ val ALERTING_GET_ALERTS_ACCESS = "alerting_get_alerts_access" val ALERTING_INDEX_WORKFLOW_ACCESS = "alerting_index_workflow_access" val ROLE_TO_PERMISSION_MAPPING = mapOf( + ALL_ACCESS_ROLE to "*", ALERTING_NO_ACCESS_ROLE to "", ALERTING_GET_EMAIL_ACCOUNT_ACCESS to "cluster:admin/opendistro/alerting/destination/email_account/get", ALERTING_SEARCH_EMAIL_ACCOUNT_ACCESS to "cluster:admin/opendistro/alerting/destination/email_account/search", diff --git a/alerting/src/test/kotlin/org/opensearch/alerting/AlertingRestTestCase.kt b/alerting/src/test/kotlin/org/opensearch/alerting/AlertingRestTestCase.kt index 860f4ea6f..84433c779 100644 --- a/alerting/src/test/kotlin/org/opensearch/alerting/AlertingRestTestCase.kt +++ b/alerting/src/test/kotlin/org/opensearch/alerting/AlertingRestTestCase.kt @@ -17,6 +17,7 @@ import org.opensearch.action.search.SearchResponse import org.opensearch.alerting.AlertingPlugin.Companion.COMMENTS_BASE_URI import org.opensearch.alerting.AlertingPlugin.Companion.EMAIL_ACCOUNT_BASE_URI import org.opensearch.alerting.AlertingPlugin.Companion.EMAIL_GROUP_BASE_URI +import org.opensearch.alerting.AlertingPlugin.Companion.MONITOR_V2_BASE_URI import org.opensearch.alerting.alerts.AlertIndices import org.opensearch.alerting.alerts.AlertIndices.Companion.FINDING_HISTORY_WRITE_INDEX import org.opensearch.alerting.core.settings.ScheduledJobSettings @@ -26,6 +27,8 @@ import org.opensearch.alerting.model.destination.Destination import org.opensearch.alerting.model.destination.Slack import org.opensearch.alerting.model.destination.email.EmailAccount import org.opensearch.alerting.model.destination.email.EmailGroup +import org.opensearch.alerting.modelv2.MonitorV2 +import org.opensearch.alerting.modelv2.PPLSQLMonitor import org.opensearch.alerting.settings.AlertingSettings import org.opensearch.alerting.settings.DestinationSettings import org.opensearch.alerting.util.DestinationType @@ -66,23 +69,25 @@ import org.opensearch.core.xcontent.ToXContent import org.opensearch.core.xcontent.XContentBuilder import org.opensearch.core.xcontent.XContentParser import org.opensearch.core.xcontent.XContentParserUtils +import org.opensearch.index.query.QueryBuilders import org.opensearch.search.SearchModule import org.opensearch.search.builder.SearchSourceBuilder +import org.opensearch.test.OpenSearchTestCase import java.net.URLEncoder import java.nio.file.Files import java.time.Instant import java.time.ZonedDateTime import java.time.format.DateTimeFormatter import java.time.temporal.ChronoUnit +import java.time.temporal.ChronoUnit.MILLIS import java.util.Locale import java.util.UUID +import java.util.concurrent.TimeUnit import java.util.stream.Collectors import javax.management.MBeanServerInvocationHandler import javax.management.ObjectName import javax.management.remote.JMXConnectorFactory import javax.management.remote.JMXServiceURL -import kotlin.collections.ArrayList -import kotlin.collections.HashMap /** * Superclass for tests that interact with an external test cluster using OpenSearch's RestClient @@ -103,6 +108,7 @@ abstract class AlertingRestTestCase : ODFERestTestCase() { return NamedXContentRegistry( mutableListOf( Monitor.XCONTENT_REGISTRY, + MonitorV2.XCONTENT_REGISTRY, SearchInput.XCONTENT_REGISTRY, DocLevelMonitorInput.XCONTENT_REGISTRY, QueryLevelTrigger.XCONTENT_REGISTRY, @@ -129,6 +135,17 @@ abstract class AlertingRestTestCase : ODFERestTestCase() { return StringEntity(jsonString, APPLICATION_JSON) } + private fun createMonitorV2EntityWithBackendRoles(monitorV2: MonitorV2, rbacRoles: List?): HttpEntity { + if (rbacRoles == null) { + return monitorV2.toHttpEntity() + } + val temp = monitorV2.toJsonString() + val toReplace = temp.lastIndexOf("}") + val rbacString = rbacRoles.joinToString { "\"$it\"" } + val jsonString = temp.substring(0, toReplace) + ", \"rbac_roles\": [$rbacString] }" + return StringEntity(jsonString, APPLICATION_JSON) + } + protected fun createMonitorWithClient( client: RestClient, monitor: Monitor, @@ -150,10 +167,48 @@ abstract class AlertingRestTestCase : ODFERestTestCase() { return getMonitor(monitorId = monitorJson["_id"] as String) } + protected fun createMonitorV2WithClient( + client: RestClient, + monitorV2: MonitorV2, + rbacRoles: List? = null + ): MonitorV2 { + // every random ppl monitor's query searches index TEST_INDEX_NAME + // by default, so create that first before creating the monitor + val indexExistsResponse = client().makeRequest("HEAD", TEST_INDEX_NAME) + if (indexExistsResponse.restStatus() == RestStatus.NOT_FOUND) { + createIndex(TEST_INDEX_NAME, Settings.EMPTY, TEST_INDEX_MAPPINGS) + } + + // be sure to use the passed in client to send the create monitor request, + // as the user stored in this client is the user whose permissions we want + // to test, not client()'s admin level user + val response = client.makeRequest( + "POST", MONITOR_V2_BASE_URI, emptyMap(), + createMonitorV2EntityWithBackendRoles(monitorV2, rbacRoles) + ) + assertEquals("Unable to create a new monitor v2", RestStatus.OK, response.restStatus()) + + val monitorV2Json = jsonXContent.createParser( + NamedXContentRegistry.EMPTY, LoggingDeprecationHandler.INSTANCE, + response.entity.content + ).map() + assertUserNull(monitorV2Json as HashMap) + + return getMonitorV2(monitorV2Id = monitorV2Json["_id"] as String) + } + protected fun createMonitor(monitor: Monitor, refresh: Boolean = true): Monitor { return createMonitorWithClient(client(), monitor, emptyList(), refresh) } + protected fun createMonitorV2(monitorV2: MonitorV2): MonitorV2 { + val client = client() + val response = client.makeRequest("POST", MONITOR_V2_BASE_URI, emptyMap(), monitorV2.toHttpEntity()) + assertEquals("Unable to create a new monitor", RestStatus.OK, response.restStatus()) + + return getMonitorV2(monitorV2Id = response.asMap()["_id"] as String) + } + protected fun deleteMonitor(monitor: Monitor, refresh: Boolean = true): Response { val response = client().makeRequest( "DELETE", "$ALERTING_BASE_URI/${monitor.id}?refresh=$refresh", emptyMap(), @@ -164,6 +219,15 @@ abstract class AlertingRestTestCase : ODFERestTestCase() { return response } + protected fun deleteMonitorV2(monitorV2Id: String): Response { + val response = client().makeRequest( + "DELETE", "$MONITOR_V2_BASE_URI/$monitorV2Id?refresh=true", emptyMap() + ) + assertEquals("Unable to delete a monitor", RestStatus.OK, response.restStatus()) + + return response + } + protected fun deleteWorkflow(workflow: Workflow, deleteDelegates: Boolean = false, refresh: Boolean = true): Response { val response = client().makeRequest( "DELETE", @@ -535,6 +599,18 @@ abstract class AlertingRestTestCase : ODFERestTestCase() { return getMonitor(monitorId = monitorId) } + protected fun createRandomPPLMonitor(pplMonitorConfig: PPLSQLMonitor = randomPPLMonitor()): PPLSQLMonitor { + // every random ppl monitor's query searches index TEST_INDEX_NAME + // by default, so create that first before creating the monitor + val indexExistsResponse = adminClient().makeRequest("HEAD", TEST_INDEX_NAME) + if (indexExistsResponse.restStatus() == RestStatus.NOT_FOUND) { + createIndex(TEST_INDEX_NAME, Settings.EMPTY, TEST_INDEX_MAPPINGS) + } + logger.info("ppl monitor: $pplMonitorConfig") + val pplMonitorId = createMonitorV2(pplMonitorConfig).id + return getMonitorV2(monitorV2Id = pplMonitorId) as PPLSQLMonitor + } + protected fun createRandomDocumentMonitor(refresh: Boolean = false, withMetadata: Boolean = false): Monitor { val monitor = randomDocumentLevelMonitor(withMetadata = withMetadata) val monitorId = createMonitor(monitor, refresh).id @@ -568,6 +644,16 @@ abstract class AlertingRestTestCase : ODFERestTestCase() { return getWorkflow(workflowId = workflow.id) } + @Suppress("UNCHECKED_CAST") + protected fun updateMonitorV2(monitorV2: MonitorV2, refresh: Boolean = false): MonitorV2 { + val response = client().makeRequest( + "PUT", "$MONITOR_V2_BASE_URI/${monitorV2.id}?refresh=$refresh", + emptyMap(), monitorV2.toHttpEntity() + ) + assertEquals("Unable to update a monitorV2", RestStatus.OK, response.restStatus()) + return getMonitorV2(monitorV2Id = monitorV2.id) + } + protected fun updateMonitorWithClient( client: RestClient, monitor: Monitor, @@ -635,6 +721,33 @@ abstract class AlertingRestTestCase : ODFERestTestCase() { return monitor.copy(id = id, version = version) } + protected fun getMonitorV2( + monitorV2Id: String, + header: BasicHeader = BasicHeader(HttpHeaders.CONTENT_TYPE, "application/json") + ): MonitorV2 { + val response = client().makeRequest("GET", "$MONITOR_V2_BASE_URI/$monitorV2Id", null, header) + assertEquals("Unable to get monitorV2 $monitorV2Id", RestStatus.OK, response.restStatus()) + + val parser = createParser(XContentType.JSON.xContent(), response.entity.content) + XContentParserUtils.ensureExpectedToken(XContentParser.Token.START_OBJECT, parser.nextToken(), parser) + + lateinit var id: String + var version: Long = 0 + lateinit var monitorV2: MonitorV2 + + while (parser.nextToken() != XContentParser.Token.END_OBJECT) { + parser.nextToken() + + when (parser.currentName()) { + "_id" -> id = parser.text() + "_version" -> version = parser.longValue() + "monitorV2" -> monitorV2 = MonitorV2.parse(parser) + } + } + + return monitorV2.makeCopy(id = id, version = version) + } + // TODO: understand why doc alerts wont work with the normal search Alerts function protected fun searchAlertsWithFilter( monitor: Monitor, @@ -782,6 +895,17 @@ abstract class AlertingRestTestCase : ODFERestTestCase() { return getAlerts(client(), dataMap, header) } + protected fun getAlertV2s(): Response { + val response = client().makeRequest( + "GET", + "$MONITOR_V2_BASE_URI/alerts?", + null, + BasicHeader(HttpHeaders.CONTENT_TYPE, "application/json") + ) + assertEquals("Get call failed.", RestStatus.OK, response.restStatus()) + return response + } + protected fun refreshIndex(index: String): Response { val response = client().makeRequest("POST", "/$index/_refresh?expand_wildcards=all") assertEquals("Unable to refresh index", RestStatus.OK, response.restStatus()) @@ -843,6 +967,9 @@ abstract class AlertingRestTestCase : ODFERestTestCase() { protected fun executeMonitor(client: RestClient, monitor: Monitor, params: Map = mapOf()): Response = client.makeRequest("POST", "$ALERTING_BASE_URI/_execute", params, monitor.toHttpEntityWithUser()) + protected fun executeMonitorV2(monitorId: String, params: Map = mutableMapOf()): Response = + client().makeRequest("POST", "$MONITOR_V2_BASE_URI/$monitorId/_execute", params) + protected fun searchFindings(params: Map = mutableMapOf()): GetFindingsResponse { var baseEndpoint = "${AlertingPlugin.FINDING_BASE_URI}/_search?" @@ -1318,6 +1445,23 @@ abstract class AlertingRestTestCase : ODFERestTestCase() { return shuffleXContent(toXContentWithUser(builder, ToXContent.EMPTY_PARAMS)).string() } + protected fun MonitorV2.toHttpEntity(): HttpEntity { + return StringEntity(toJsonString(), APPLICATION_JSON) + } + + private fun MonitorV2.toJsonString(): String { + return shuffleXContent(toXContent(jsonBuilder(), ToXContent.EMPTY_PARAMS)).string() + } + + protected fun MonitorV2.toHttpEntityWithUser(): HttpEntity { + return StringEntity(toJsonStringWithUser(), APPLICATION_JSON) + } + + private fun MonitorV2.toJsonStringWithUser(): String { + val builder = jsonBuilder() + return shuffleXContent(toXContentWithUser(builder, ToXContent.EMPTY_PARAMS)).string() + } + protected fun Destination.toHttpEntity(): HttpEntity { return StringEntity(toJsonString(), APPLICATION_JSON) } @@ -1440,6 +1584,12 @@ abstract class AlertingRestTestCase : ODFERestTestCase() { return responseMap } + fun getAlertingV2Stats(metrics: String = ""): Map { + val monitorStatsResponse = client().makeRequest("GET", "/_plugins/_alerting/v2/stats$metrics") + val responseMap = createParser(XContentType.JSON.xContent(), monitorStatsResponse.entity.content).map() + return responseMap + } + fun enableScheduledJob(): Response { val updateResponse = client().makeRequest( "PUT", "_cluster/settings", @@ -2006,4 +2156,68 @@ abstract class AlertingRestTestCase : ODFERestTestCase() { return deletedCommentId } + + protected fun isMonitorScheduled(monitorId: String, alertingStatsResponse: Map): Boolean { + val nodesInfo = alertingStatsResponse["nodes"] as Map + for (nodeId in nodesInfo.keys) { + val nodeInfo = nodesInfo[nodeId] as Map + val jobsInfo = nodeInfo["jobs_info"] as Map + if (jobsInfo.keys.contains(monitorId)) { + return true + } + } + + return false + } + + // this function is used for PPL Alerting testing. + // precondition: TEST_INDEX_NAME must be created before calling this + // indexes a doc from some time ago into index TEST_INDEX_NAME. + // this function only works on the TEST_INDEX_NAME index created + // specifically for this IT suite. It has fields + // "timestamp" (date), "abc" (string), "number" (integer) + protected fun indexDocFromSomeTimeAgo(timeValue: Long, timeUnit: ChronoUnit, abc: String, number: Int) { + val someTimeAgo = ZonedDateTime.now().minus(timeValue, timeUnit).truncatedTo(MILLIS) + val testTime = DateTimeFormatter.ISO_OFFSET_DATE_TIME.format(someTimeAgo) // the timestamp string is given a random timezone offset + val testDoc = """{ "timestamp" : "$testTime", "abc": "$abc", "number" : "$number" }""" + indexDoc(TEST_INDEX_NAME, UUID.randomUUID().toString(), testDoc) + } + + protected fun ensureNumMonitorV2s(expectedNum: Int) { + // if a validation error is thrown but a monitor is still accidentally created, + // what happens is that this check runs before the workflows to create + // alerting-config index and index the monitor complete, meaning this check gets + // no search results, then afterwards, the monitor is created, leading this function + // to falsely believe no monitor was create. wait some amount of time to let the + // workflows incorrectly create whatever monitors it will + OpenSearchTestCase.waitUntil({ + return@waitUntil false + }, 10, TimeUnit.SECONDS) + + val search = SearchSourceBuilder().query(QueryBuilders.matchAllQuery()).toString() + val searchResponse = client().makeRequest( + "POST", "$MONITOR_V2_BASE_URI/_search", + StringEntity(search, ContentType.APPLICATION_JSON) + ) + + assertEquals("Search monitor failed", RestStatus.OK, searchResponse.restStatus()) + val xcp = createParser(XContentType.JSON.xContent(), searchResponse.entity.content) + val hits = xcp.map()["hits"]!! as Map> + val numberDocsFound = hits["total"]?.get("value") + assertEquals("Unexpected number of PPL Monitors found in Search Monitors", expectedNum, numberDocsFound) + } + + // takes in an execute monitor API response and returns true if the + // trigger condition was met. assumes the monitor executed only had 1 trigger + protected fun isTriggered(pplMonitor: PPLSQLMonitor, executeResponse: Response): Boolean { + val executeResponseMap = entityAsMap(executeResponse) + val triggerResultsObj = (executeResponseMap["trigger_results"] as Map)[pplMonitor.triggers[0].id] as Map + return triggerResultsObj["triggered"] as Boolean + } + + // takes in a get alerts API response and returns the current number of active alerts + protected fun numAlerts(getAlertsResponse: Response): Int { + logger.info("get alerts response: ${entityAsMap(getAlertsResponse)}") + return entityAsMap(getAlertsResponse)["totalAlertV2s"] as Int + } } diff --git a/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/MonitorRestApiIT.kt b/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/MonitorRestApiIT.kt index 5de980bfa..64a5b9c40 100644 --- a/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/MonitorRestApiIT.kt +++ b/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/MonitorRestApiIT.kt @@ -72,7 +72,9 @@ import java.util.concurrent.TimeUnit @Suppress("UNCHECKED_CAST") class MonitorRestApiIT : AlertingRestTestCase() { - val USE_TYPED_KEYS = ToXContent.MapParams(mapOf("with_type" to "true")) + companion object { + val USE_TYPED_KEYS = ToXContent.MapParams(mapOf("with_type" to "true")) + } @Throws(Exception::class) fun `test plugin is loaded`() { @@ -1534,19 +1536,6 @@ class MonitorRestApiIT : AlertingRestTestCase() { assertEquals("More than $numberOfNodes successful node", numberOfNodes, nodesResponse["successful"]) } - private fun isMonitorScheduled(monitorId: String, alertingStatsResponse: Map): Boolean { - val nodesInfo = alertingStatsResponse["nodes"] as Map - for (nodeId in nodesInfo.keys) { - val nodeInfo = nodesInfo[nodeId] as Map - val jobsInfo = nodeInfo["jobs_info"] as Map - if (jobsInfo.keys.contains(monitorId)) { - return true - } - } - - return false - } - private fun assertAlertingStatsSweeperEnabled(alertingStatsResponse: Map, expected: Boolean) { assertEquals( "Legacy scheduled job enabled field is not set to $expected", diff --git a/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/MonitorV2RestApiIT.kt b/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/MonitorV2RestApiIT.kt new file mode 100644 index 000000000..cf4959ea6 --- /dev/null +++ b/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/MonitorV2RestApiIT.kt @@ -0,0 +1,63 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.alerting.resthandler + +import org.opensearch.alerting.AlertingPlugin.Companion.MONITOR_V2_BASE_URI +import org.opensearch.alerting.AlertingRestTestCase +import org.opensearch.alerting.TEST_INDEX_MAPPINGS +import org.opensearch.alerting.TEST_INDEX_NAME +import org.opensearch.alerting.makeRequest +import org.opensearch.alerting.modelv2.MonitorV2 +import org.opensearch.alerting.randomPPLMonitor +import org.opensearch.client.ResponseException +import org.opensearch.common.UUIDs +import org.opensearch.common.settings.Settings +import org.opensearch.core.rest.RestStatus +import org.opensearch.test.junit.annotations.TestLogging + +/*** + * Tests Alerting V2 CRUD and validations + * + * Gradle command to run this suite: + * ./gradlew :alerting:integTest -Dhttps=true -Dsecurity=true -Duser=admin -Dpassword=admin \ + * --tests "org.opensearch.alerting.resthandler.MonitorV2RestApiIT" + */ +@TestLogging("level:DEBUG", reason = "Debug for tests.") +@Suppress("UNCHECKED_CAST") +class MonitorV2RestApiIT : AlertingRestTestCase() { + + /* Simple Case Tests */ + fun `test create ppl monitor`() { + createIndex(TEST_INDEX_NAME, Settings.EMPTY, TEST_INDEX_MAPPINGS) + val pplMonitor = randomPPLMonitor() + + val response = client().makeRequest("POST", MONITOR_V2_BASE_URI, emptyMap(), pplMonitor.toHttpEntity()) + assertEquals("Unable to create a new monitor v2", RestStatus.OK, response.restStatus()) + + val responseBody = response.asMap() + val createdId = responseBody["_id"] as String + val createdVersion = responseBody["_version"] as Int + assertNotEquals("response is missing Id", MonitorV2.NO_ID, createdId) + assertEquals("incorrect version", 1, createdVersion) + } + + /* Validation Tests */ + fun `test update nonexistent ppl monitor fails`() { + // the random monitor query searches index TEST_INDEX_NAME, + // so we need to create that first to ensure at least the request body is valid + createIndex(TEST_INDEX_NAME, Settings.EMPTY, TEST_INDEX_MAPPINGS) + + val monitorV2 = randomPPLMonitor() + val randomId = UUIDs.base64UUID() + + try { + client().makeRequest("PUT", "$MONITOR_V2_BASE_URI/$randomId", emptyMap(), monitorV2.toHttpEntity()) + fail("Expected request to fail with NOT_FOUND but it succeeded") + } catch (e: ResponseException) { + assertEquals("Unexpected status", RestStatus.NOT_FOUND, e.response.restStatus()) + } + } +} diff --git a/core/build.gradle b/core/build.gradle index cfce74c42..9aad7da88 100644 --- a/core/build.gradle +++ b/core/build.gradle @@ -8,7 +8,8 @@ apply plugin: 'opensearch.java-rest-test' apply plugin: 'org.jetbrains.kotlin.jvm' apply plugin: 'jacoco' -configurations{ +configurations { + zipArchive all { resolutionStrategy { // force commons-beanutils to a non-vulnerable version @@ -17,6 +18,18 @@ configurations{ } } +def sqlJarDirectory = "$buildDir/dependencies/opensearch-sql-plugin" + +task addJarsToClasspath(type: Copy) { + from(fileTree(dir: sqlJarDirectory)) { + include "opensearch-sql-${opensearch_build}.jar" + include "ppl-${opensearch_build}.jar" + include "protocol-${opensearch_build}.jar" + include "core-${opensearch_build}.jar" + } + into("$buildDir/classes") +} + dependencies { compileOnly "org.opensearch:opensearch:${opensearch_version}" implementation "org.jetbrains.kotlin:kotlin-stdlib:${kotlin_version}" @@ -26,8 +39,44 @@ dependencies { api "org.opensearch.client:opensearch-rest-client:${opensearch_version}" api "org.opensearch:common-utils:${common_utils_version}@jar" implementation 'commons-validator:commons-validator:1.7' + implementation 'org.json:json:20240303' + + api fileTree(dir: sqlJarDirectory, include: ["opensearch-sql-thin-${opensearch_build}.jar", "ppl-${opensearch_build}.jar", "protocol-${opensearch_build}.jar", "core-${opensearch_build}.jar"]) + + zipArchive group: 'org.opensearch.plugin', name:'opensearch-sql-plugin', version: "${opensearch_build}" testImplementation "org.opensearch.test:framework:${opensearch_version}" testImplementation "org.jetbrains.kotlin:kotlin-test:${kotlin_version}" testImplementation "org.jetbrains.kotlin:kotlin-test-junit:${kotlin_version}" } + +task extractSqlJar(type: Copy) { + mustRunAfter() + from(zipTree(configurations.zipArchive.find { it.name.startsWith("opensearch-sql-plugin") })) + into sqlJarDirectory +} + +task extractSqlClass(type: Copy, dependsOn: [extractSqlJar]) { + from zipTree("${sqlJarDirectory}/opensearch-sql-${opensearch_build}.jar") + into("$buildDir/opensearch-sql") + include 'org/opensearch/sql/**' +} + +task replaceSqlJar(type: Jar, dependsOn: [extractSqlClass]) { + from("$buildDir/opensearch-sql") + archiveFileName = "opensearch-sql-thin-${opensearch_build}.jar" + destinationDirectory = file(sqlJarDirectory) + doLast { + file("${sqlJarDirectory}/opensearch-sql-${opensearch_build}.jar").delete() + } +} + +tasks.addJarsToClasspath.dependsOn(replaceSqlJar) + +compileJava { + dependsOn addJarsToClasspath +} + +compileKotlin { + dependsOn addJarsToClasspath +} diff --git a/core/src/main/kotlin/org/opensearch/alerting/core/ppl/PPLPluginInterface.kt b/core/src/main/kotlin/org/opensearch/alerting/core/ppl/PPLPluginInterface.kt new file mode 100644 index 000000000..2176d3b31 --- /dev/null +++ b/core/src/main/kotlin/org/opensearch/alerting/core/ppl/PPLPluginInterface.kt @@ -0,0 +1,50 @@ +package org.opensearch.alerting.core.ppl + +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.client.node.NodeClient + +/** + * Transport action plugin interfaces for the SQL/PPL plugin + */ +object PPLPluginInterface { + fun executeQuery( + client: NodeClient, + request: TransportPPLQueryRequest, + listener: ActionListener + ) { + client.execute( + PPLQueryAction.INSTANCE, + request, + wrapActionListener(listener) { response -> recreateObject(response) { TransportPPLQueryResponse(it) } } + ) + } + + /** + * Wrap action listener on concrete response class by a new created one on ActionResponse. + * This is required because the response may be loaded by different classloader across plugins. + * The onResponse(ActionResponse) avoids type cast exception and give a chance to recreate + * the response object. + */ + @Suppress("UNCHECKED_CAST") + private fun wrapActionListener( + listener: ActionListener, + recreate: (Writeable) -> Response + ): ActionListener { + return object : ActionListener { + override fun onResponse(response: ActionResponse) { + val recreated = recreate(response) + listener.onResponse(recreated) + } + + override fun onFailure(exception: java.lang.Exception) { + listener.onFailure(exception) + } + } as ActionListener + } +} diff --git a/core/src/main/kotlin/org/opensearch/alerting/opensearchapi/OpenSearchExtensions.kt b/core/src/main/kotlin/org/opensearch/alerting/opensearchapi/OpenSearchExtensions.kt index 582d13fbe..fd500ef1d 100644 --- a/core/src/main/kotlin/org/opensearch/alerting/opensearchapi/OpenSearchExtensions.kt +++ b/core/src/main/kotlin/org/opensearch/alerting/opensearchapi/OpenSearchExtensions.kt @@ -14,6 +14,7 @@ import org.opensearch.OpenSearchException import org.opensearch.action.bulk.BackoffPolicy import org.opensearch.action.search.SearchResponse import org.opensearch.action.search.ShardSearchFailure +import org.opensearch.alerting.core.ppl.PPLPluginInterface import org.opensearch.common.settings.Settings import org.opensearch.common.util.concurrent.ThreadContext import org.opensearch.common.xcontent.XContentHelper @@ -170,6 +171,20 @@ suspend fun NotificationsPluginInterface.suspendUntil(block: NotificationsPl }) } +/** + * Converts [PPLPluginInterface] methods that take a callback into a kotlin suspending function. + * + * @param block - a block of code that is passed an [ActionListener] that should be passed to the PPLPluginInterface API. + */ +suspend fun PPLPluginInterface.suspendUntil(block: PPLPluginInterface.(ActionListener) -> Unit): T = + suspendCoroutine { cont -> + block(object : ActionListener { + override fun onResponse(response: T) = cont.resume(response) + + override fun onFailure(e: Exception) = cont.resumeWithException(e) + }) + } + class InjectorContextElement( id: String, settings: Settings,