From 0d659f2dfabedaf54c4d0ef6f2aeb6bf5b0fc8de Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Mon, 3 Nov 2025 02:08:01 +0000 Subject: [PATCH] PPL Alerting: Delete Monitor, More V1/V2 Separation (#1968) * PPL Alerting: Delete Monitor, More V1/V2 Separation Signed-off-by: Dennis Toepker * making v1 v2 separation error messaging more actionable Signed-off-by: Dennis Toepker --------- Signed-off-by: Dennis Toepker Co-authored-by: Dennis Toepker (cherry picked from commit fd151de408487280c01854e909885bfdfa180049) Signed-off-by: github-actions[bot] --- .../org/opensearch/alerting/AlertingPlugin.kt | 5 + .../opensearch/alerting/AlertingV2Utils.kt | 24 +++- .../opensearch/alerting/WorkflowService.kt | 8 +- .../actionv2/DeleteMonitorV2Action.kt | 15 +++ .../actionv2/DeleteMonitorV2Request.kt | 39 ++++++ .../actionv2/DeleteMonitorV2Response.kt | 43 ++++++ .../RestDeleteMonitorV2Action.kt | 52 ++++++++ .../alerting/service/DeleteMonitorService.kt | 21 ++- .../transport/TransportDeleteMonitorAction.kt | 18 ++- .../TransportDeleteWorkflowAction.kt | 18 ++- .../transport/TransportGetMonitorAction.kt | 27 ++-- .../transport/TransportSearchMonitorAction.kt | 47 +------ .../TransportDeleteMonitorV2Action.kt | 126 ++++++++++++++++++ .../resthandler/MonitorV2RestApiIT.kt | 21 +++ .../resthandler/SecureMonitorV2RestApiIT.kt | 93 +++++++++++++ 15 files changed, 482 insertions(+), 75 deletions(-) create mode 100644 alerting/src/main/kotlin/org/opensearch/alerting/actionv2/DeleteMonitorV2Action.kt create mode 100644 alerting/src/main/kotlin/org/opensearch/alerting/actionv2/DeleteMonitorV2Request.kt create mode 100644 alerting/src/main/kotlin/org/opensearch/alerting/actionv2/DeleteMonitorV2Response.kt create mode 100644 alerting/src/main/kotlin/org/opensearch/alerting/resthandlerv2/RestDeleteMonitorV2Action.kt create mode 100644 alerting/src/main/kotlin/org/opensearch/alerting/transportv2/TransportDeleteMonitorV2Action.kt diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/AlertingPlugin.kt b/alerting/src/main/kotlin/org/opensearch/alerting/AlertingPlugin.kt index 9e8dfab6a..93c17ab14 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.DeleteMonitorV2Action import org.opensearch.alerting.actionv2.GetMonitorV2Action import org.opensearch.alerting.actionv2.IndexMonitorV2Action import org.opensearch.alerting.actionv2.SearchMonitorV2Action @@ -55,6 +56,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.RestDeleteMonitorV2Action import org.opensearch.alerting.resthandlerv2.RestGetMonitorV2Action import org.opensearch.alerting.resthandlerv2.RestIndexMonitorV2Action import org.opensearch.alerting.resthandlerv2.RestSearchMonitorV2Action @@ -90,6 +92,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.TransportDeleteMonitorV2Action import org.opensearch.alerting.transportv2.TransportGetMonitorV2Action import org.opensearch.alerting.transportv2.TransportIndexMonitorV2Action import org.opensearch.alerting.transportv2.TransportSearchMonitorV2Action @@ -233,6 +236,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R // Alerting V2 RestIndexMonitorV2Action(), + RestDeleteMonitorV2Action(), RestGetMonitorV2Action(), RestSearchMonitorV2Action(settings, clusterService), ) @@ -273,6 +277,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R ActionPlugin.ActionHandler(IndexMonitorV2Action.INSTANCE, TransportIndexMonitorV2Action::class.java), ActionPlugin.ActionHandler(GetMonitorV2Action.INSTANCE, TransportGetMonitorV2Action::class.java), ActionPlugin.ActionHandler(SearchMonitorV2Action.INSTANCE, TransportSearchMonitorV2Action::class.java), + ActionPlugin.ActionHandler(DeleteMonitorV2Action.INSTANCE, TransportDeleteMonitorV2Action::class.java), ) } diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/AlertingV2Utils.kt b/alerting/src/main/kotlin/org/opensearch/alerting/AlertingV2Utils.kt index 692b7db6c..a0b43b697 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/AlertingV2Utils.kt +++ b/alerting/src/main/kotlin/org/opensearch/alerting/AlertingV2Utils.kt @@ -9,6 +9,8 @@ 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.AlertingPlugin.Companion.MONITOR_BASE_URI +import org.opensearch.alerting.AlertingPlugin.Companion.MONITOR_V2_BASE_URI import org.opensearch.alerting.modelv2.MonitorV2 import org.opensearch.commons.alerting.model.Monitor import org.opensearch.commons.alerting.model.ScheduledJob @@ -27,9 +29,16 @@ object AlertingV2Utils { // 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") + return IllegalStateException( + "The ID given corresponds to an Alerting V2 Monitor, but a V1 Monitor was expected. " + + "If you wish to operate on a V1 Monitor (e.g. Per Query, Per Document, etc), please use " + + "the Alerting V1 APIs with endpoint prefix: $MONITOR_BASE_URI." + ) } 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 IllegalStateException( + "The ID given corresponds to a scheduled job of unknown type: ${scheduledJob.javaClass.name}. " + + "Please validate the ID and ensure it corresponds to a valid Monitor." + ) } return null } @@ -38,9 +47,16 @@ object AlertingV2Utils { // 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") + return IllegalStateException( + "The ID given corresponds to an Alerting V1 Monitor, but a V2 Monitor was expected. " + + "If you wish to operate on a V2 Monitor (e.g. PPL Monitor), please use " + + "the Alerting V2 APIs with endpoint prefix: $MONITOR_V2_BASE_URI." + ) } else if (scheduledJob !is MonitorV2) { - return IllegalStateException("The ID given corresponds to a scheduled job of unknown type: ${scheduledJob.javaClass.name}") + return IllegalStateException( + "The ID given corresponds to a scheduled job of unknown type: ${scheduledJob.javaClass.name}. " + + "Please validate the ID and ensure it corresponds to a valid Monitor." + ) } return null } diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/WorkflowService.kt b/alerting/src/main/kotlin/org/opensearch/alerting/WorkflowService.kt index 15f1e192e..83f8b56e0 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/WorkflowService.kt +++ b/alerting/src/main/kotlin/org/opensearch/alerting/WorkflowService.kt @@ -11,6 +11,7 @@ import org.opensearch.action.admin.indices.exists.indices.IndicesExistsRequest import org.opensearch.action.admin.indices.exists.indices.IndicesExistsResponse import org.opensearch.action.search.SearchRequest import org.opensearch.action.search.SearchResponse +import org.opensearch.alerting.AlertingV2Utils.validateMonitorV1 import org.opensearch.alerting.opensearchapi.suspendUntil import org.opensearch.common.xcontent.LoggingDeprecationHandler import org.opensearch.common.xcontent.XContentType @@ -132,7 +133,12 @@ class WorkflowService( 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/actionv2/DeleteMonitorV2Action.kt b/alerting/src/main/kotlin/org/opensearch/alerting/actionv2/DeleteMonitorV2Action.kt new file mode 100644 index 000000000..b182d87d4 --- /dev/null +++ b/alerting/src/main/kotlin/org/opensearch/alerting/actionv2/DeleteMonitorV2Action.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 DeleteMonitorV2Action private constructor() : ActionType(NAME, ::DeleteMonitorV2Response) { + companion object { + val INSTANCE = DeleteMonitorV2Action() + const val NAME = "cluster:admin/opensearch/alerting/v2/monitor/delete" + } +} diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/actionv2/DeleteMonitorV2Request.kt b/alerting/src/main/kotlin/org/opensearch/alerting/actionv2/DeleteMonitorV2Request.kt new file mode 100644 index 000000000..7024842ac --- /dev/null +++ b/alerting/src/main/kotlin/org/opensearch/alerting/actionv2/DeleteMonitorV2Request.kt @@ -0,0 +1,39 @@ +/* + * 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.core.common.io.stream.StreamInput +import org.opensearch.core.common.io.stream.StreamOutput +import java.io.IOException + +class DeleteMonitorV2Request : ActionRequest { + val monitorV2Id: String + val refreshPolicy: WriteRequest.RefreshPolicy + + constructor(monitorV2Id: String, refreshPolicy: WriteRequest.RefreshPolicy) : super() { + this.monitorV2Id = monitorV2Id + this.refreshPolicy = refreshPolicy + } + + @Throws(IOException::class) + constructor(sin: StreamInput) : this( + monitorV2Id = sin.readString(), + refreshPolicy = WriteRequest.RefreshPolicy.readFrom(sin) + ) + + override fun validate(): ActionRequestValidationException? { + return null + } + + @Throws(IOException::class) + override fun writeTo(out: StreamOutput) { + out.writeString(monitorV2Id) + refreshPolicy.writeTo(out) + } +} diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/actionv2/DeleteMonitorV2Response.kt b/alerting/src/main/kotlin/org/opensearch/alerting/actionv2/DeleteMonitorV2Response.kt new file mode 100644 index 000000000..b4850b662 --- /dev/null +++ b/alerting/src/main/kotlin/org/opensearch/alerting/actionv2/DeleteMonitorV2Response.kt @@ -0,0 +1,43 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.alerting.actionv2 + +import org.opensearch.commons.alerting.util.IndexUtils +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 + +class DeleteMonitorV2Response : BaseResponse { + var id: String + var version: Long + + constructor( + id: String, + version: Long + ) : super() { + this.id = id + this.version = version + } + + constructor(sin: StreamInput) : this( + sin.readString(), // id + sin.readLong() // version + ) + + override fun writeTo(out: StreamOutput) { + out.writeString(id) + out.writeLong(version) + } + + override fun toXContent(builder: XContentBuilder, params: ToXContent.Params): XContentBuilder { + return builder.startObject() + .field(IndexUtils._ID, id) + .field(IndexUtils._VERSION, version) + .endObject() + } +} diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/resthandlerv2/RestDeleteMonitorV2Action.kt b/alerting/src/main/kotlin/org/opensearch/alerting/resthandlerv2/RestDeleteMonitorV2Action.kt new file mode 100644 index 000000000..92a49b003 --- /dev/null +++ b/alerting/src/main/kotlin/org/opensearch/alerting/resthandlerv2/RestDeleteMonitorV2Action.kt @@ -0,0 +1,52 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.alerting.resthandlerv2 + +import org.apache.logging.log4j.LogManager +import org.apache.logging.log4j.Logger +import org.opensearch.action.support.WriteRequest.RefreshPolicy +import org.opensearch.alerting.AlertingPlugin +import org.opensearch.alerting.actionv2.DeleteMonitorV2Action +import org.opensearch.alerting.actionv2.DeleteMonitorV2Request +import org.opensearch.alerting.util.REFRESH +import org.opensearch.rest.BaseRestHandler +import org.opensearch.rest.RestHandler.Route +import org.opensearch.rest.RestRequest +import org.opensearch.rest.RestRequest.Method.DELETE +import org.opensearch.rest.action.RestToXContentListener +import org.opensearch.transport.client.node.NodeClient +import java.io.IOException + +private val log: Logger = LogManager.getLogger(RestDeleteMonitorV2Action::class.java) + +class RestDeleteMonitorV2Action : BaseRestHandler() { + + override fun getName(): String { + return "delete_monitor_v2_action" + } + + override fun routes(): List { + return mutableListOf( + Route( + DELETE, + "${AlertingPlugin.MONITOR_V2_BASE_URI}/{monitorV2Id}" + ) + ) + } + + @Throws(IOException::class) + override fun prepareRequest(request: RestRequest, client: NodeClient): RestChannelConsumer { + val monitorV2Id = request.param("monitorV2Id") + log.info("${request.method()} ${AlertingPlugin.MONITOR_V2_BASE_URI}/$monitorV2Id") + + val refreshPolicy = RefreshPolicy.parse(request.param(REFRESH, RefreshPolicy.IMMEDIATE.value)) + val deleteMonitorV2Request = DeleteMonitorV2Request(monitorV2Id, refreshPolicy) + + return RestChannelConsumer { channel -> + client.execute(DeleteMonitorV2Action.INSTANCE, deleteMonitorV2Request, RestToXContentListener(channel)) + } + } +} diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/service/DeleteMonitorService.kt b/alerting/src/main/kotlin/org/opensearch/alerting/service/DeleteMonitorService.kt index fbc655543..c1a56a4c6 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/service/DeleteMonitorService.kt +++ b/alerting/src/main/kotlin/org/opensearch/alerting/service/DeleteMonitorService.kt @@ -22,6 +22,7 @@ import org.opensearch.action.support.IndicesOptions import org.opensearch.action.support.WriteRequest.RefreshPolicy import org.opensearch.action.support.clustermanager.AcknowledgedResponse import org.opensearch.alerting.MonitorMetadataService +import org.opensearch.alerting.actionv2.DeleteMonitorV2Response import org.opensearch.alerting.core.lock.LockModel import org.opensearch.alerting.core.lock.LockService import org.opensearch.alerting.opensearchapi.suspendUntil @@ -74,6 +75,19 @@ object DeleteMonitorService : return DeleteMonitorResponse(deleteResponse.id, deleteResponse.version) } + /** + * Deletes the monitorV2, which does not come with other metadata and queries + * like doc level monitors + * @param monitorV2Id monitorV2 ID to be deleted + * @param refreshPolicy + */ + suspend fun deleteMonitorV2(monitorV2Id: String, refreshPolicy: RefreshPolicy): DeleteMonitorV2Response { + val deleteResponse = deleteMonitor(monitorV2Id, refreshPolicy) + deleteLock(monitorV2Id) + return DeleteMonitorV2Response(deleteResponse.id, deleteResponse.version) + } + + // both Alerting v1 and v2 workflows flow through this function private suspend fun deleteMonitor(monitorId: String, refreshPolicy: RefreshPolicy): DeleteResponse { val deleteMonitorRequest = DeleteRequest(ScheduledJob.SCHEDULED_JOBS_INDEX, monitorId) .setRefreshPolicy(refreshPolicy) @@ -167,7 +181,12 @@ object DeleteMonitorService : } private suspend fun deleteLock(monitor: Monitor) { - client.suspendUntil { lockService.deleteLock(LockModel.generateLockId(monitor.id), it) } + deleteLock(monitor.id) + } + + // both Alerting v1 and v2 workflows flow through this function + private suspend fun deleteLock(monitorId: String) { + client.suspendUntil { lockService.deleteLock(LockModel.generateLockId(monitorId), it) } } /** diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportDeleteMonitorAction.kt b/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportDeleteMonitorAction.kt index b28311bd0..39b96f2b0 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportDeleteMonitorAction.kt +++ b/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportDeleteMonitorAction.kt @@ -16,6 +16,7 @@ import org.opensearch.action.get.GetResponse import org.opensearch.action.support.ActionFilters import org.opensearch.action.support.HandledTransportAction import org.opensearch.action.support.WriteRequest.RefreshPolicy +import org.opensearch.alerting.AlertingV2Utils.validateMonitorV1 import org.opensearch.alerting.opensearchapi.suspendUntil import org.opensearch.alerting.service.DeleteMonitorService import org.opensearch.alerting.settings.AlertingSettings @@ -87,7 +88,7 @@ class TransportDeleteMonitorAction @Inject constructor( ) { suspend fun resolveUserAndStart(refreshPolicy: RefreshPolicy) { try { - val monitor = getMonitor() + val monitor = getMonitor() ?: return // null means there was an issue retrieving the Monitor val canDelete = user == null || !doFilterForUser(user) || checkUserPermissionsWithResource(user, monitor.user, actionListener, "monitor", monitorId) @@ -115,11 +116,11 @@ class TransportDeleteMonitorAction @Inject constructor( } } - private suspend fun getMonitor(): Monitor { + private suspend fun getMonitor(): Monitor? { val getRequest = GetRequest(ScheduledJob.SCHEDULED_JOBS_INDEX, monitorId) val getResponse: GetResponse = client.suspendUntil { get(getRequest, it) } - if (getResponse.isExists == false) { + if (!getResponse.isExists) { actionListener.onFailure( AlertingException.wrap( OpenSearchStatusException("Monitor with $monitorId is not found", RestStatus.NOT_FOUND) @@ -130,7 +131,16 @@ class TransportDeleteMonitorAction @Inject constructor( xContentRegistry, LoggingDeprecationHandler.INSTANCE, getResponse.sourceAsBytesRef, XContentType.JSON ) - return 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 null + } + + val monitor = scheduledJob as Monitor + + return monitor } } } diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportDeleteWorkflowAction.kt b/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportDeleteWorkflowAction.kt index bf0d44eab..60e018e50 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportDeleteWorkflowAction.kt +++ b/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportDeleteWorkflowAction.kt @@ -23,6 +23,7 @@ import org.opensearch.action.search.SearchResponse import org.opensearch.action.support.ActionFilters import org.opensearch.action.support.HandledTransportAction import org.opensearch.action.support.WriteRequest.RefreshPolicy +import org.opensearch.alerting.AlertingV2Utils.validateMonitorV1 import org.opensearch.alerting.core.lock.LockModel import org.opensearch.alerting.core.lock.LockService import org.opensearch.alerting.opensearchapi.addFilter @@ -299,7 +300,13 @@ class TransportDeleteWorkflowAction @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 deletableMonitors.add(monitor) } } @@ -327,12 +334,17 @@ class TransportDeleteWorkflowAction @Inject constructor( ) } - private fun parseWorkflow(getResponse: GetResponse): Workflow { + private fun parseWorkflow(getResponse: GetResponse): Workflow? { val xcp = XContentHelper.createParser( xContentRegistry, LoggingDeprecationHandler.INSTANCE, getResponse.sourceAsBytesRef, XContentType.JSON ) - return 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 null + } + return scheduledJob as Workflow } private suspend fun deleteWorkflow(deleteRequest: DeleteRequest): DeleteResponse { diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportGetMonitorAction.kt b/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportGetMonitorAction.kt index 97e65ec50..ad8c8d021 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportGetMonitorAction.kt +++ b/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportGetMonitorAction.kt @@ -18,6 +18,8 @@ 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.alerting.AlertingV2Utils.isIndexNotFoundException +import org.opensearch.alerting.AlertingV2Utils.validateMonitorV1 import org.opensearch.alerting.opensearchapi.suspendUntil import org.opensearch.alerting.settings.AlertingSettings import org.opensearch.alerting.util.ScheduledJobUtils.Companion.WORKFLOW_DELEGATE_PATH @@ -41,11 +43,9 @@ import org.opensearch.commons.utils.recreateObject import org.opensearch.core.action.ActionListener import org.opensearch.core.rest.RestStatus import org.opensearch.core.xcontent.NamedXContentRegistry -import org.opensearch.index.IndexNotFoundException import org.opensearch.index.query.QueryBuilders import org.opensearch.search.builder.SearchSourceBuilder import org.opensearch.tasks.Task -import org.opensearch.transport.RemoteTransportException import org.opensearch.transport.TransportService import org.opensearch.transport.client.Client @@ -116,7 +116,14 @@ class TransportGetMonitorAction @Inject constructor( response.sourceAsBytesRef, XContentType.JSON ).use { xcp -> - monitor = ScheduledJob.parse(xcp, response.id, response.version) as Monitor + val scheduledJob = ScheduledJob.parse(xcp, response.id, response.version) + + validateMonitorV1(scheduledJob)?.let { + actionListener.onFailure(AlertingException.wrap(it)) + return + } + + monitor = scheduledJob as Monitor // security is enabled and filterby is enabled if (!checkUserPermissionsWithResource( @@ -168,20 +175,6 @@ class TransportGetMonitorAction @Inject constructor( } } - // Checks if the exception is caused by an IndexNotFoundException (directly or nested). - private 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 - } - private suspend fun getAssociatedWorkflows(id: String): List { try { val associatedWorkflows = mutableListOf() diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportSearchMonitorAction.kt b/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportSearchMonitorAction.kt index 21c633553..099fd9a2b 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportSearchMonitorAction.kt +++ b/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportSearchMonitorAction.kt @@ -6,15 +6,13 @@ package org.opensearch.alerting.transport import org.apache.logging.log4j.LogManager -import org.apache.lucene.search.TotalHits -import org.apache.lucene.search.TotalHits.Relation import org.opensearch.action.ActionRequest import org.opensearch.action.search.SearchRequest import org.opensearch.action.search.SearchResponse -import org.opensearch.action.search.SearchResponse.Clusters -import org.opensearch.action.search.ShardSearchFailure import org.opensearch.action.support.ActionFilters import org.opensearch.action.support.HandledTransportAction +import org.opensearch.alerting.AlertingV2Utils.getEmptySearchResponse +import org.opensearch.alerting.AlertingV2Utils.isIndexNotFoundException import org.opensearch.alerting.opensearchapi.addFilter import org.opensearch.alerting.settings.AlertingSettings import org.opensearch.alerting.util.use @@ -31,21 +29,13 @@ import org.opensearch.commons.authuser.User import org.opensearch.commons.utils.recreateObject import org.opensearch.core.action.ActionListener import org.opensearch.core.common.io.stream.NamedWriteableRegistry -import org.opensearch.index.IndexNotFoundException import org.opensearch.index.query.BoolQueryBuilder import org.opensearch.index.query.ExistsQueryBuilder import org.opensearch.index.query.MatchQueryBuilder import org.opensearch.index.query.QueryBuilders -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.tasks.Task -import org.opensearch.transport.RemoteTransportException import org.opensearch.transport.TransportService import org.opensearch.transport.client.Client -import java.util.Collections private val log = LogManager.getLogger(TransportSearchMonitorAction::class.java) @@ -112,39 +102,6 @@ class TransportSearchMonitorAction @Inject constructor( } } - 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 - ) - } - - // Checks if the exception is caused by an IndexNotFoundException (directly or nested). - private 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 - } - fun search(searchRequest: SearchRequest, actionListener: ActionListener) { client.search( searchRequest, diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/transportv2/TransportDeleteMonitorV2Action.kt b/alerting/src/main/kotlin/org/opensearch/alerting/transportv2/TransportDeleteMonitorV2Action.kt new file mode 100644 index 000000000..8e2ed496b --- /dev/null +++ b/alerting/src/main/kotlin/org/opensearch/alerting/transportv2/TransportDeleteMonitorV2Action.kt @@ -0,0 +1,126 @@ +/* + * 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 org.apache.logging.log4j.LogManager +import org.opensearch.OpenSearchStatusException +import org.opensearch.action.get.GetRequest +import org.opensearch.action.get.GetResponse +import org.opensearch.action.support.ActionFilters +import org.opensearch.action.support.HandledTransportAction +import org.opensearch.alerting.AlertingV2Utils +import org.opensearch.alerting.actionv2.DeleteMonitorV2Action +import org.opensearch.alerting.actionv2.DeleteMonitorV2Request +import org.opensearch.alerting.actionv2.DeleteMonitorV2Response +import org.opensearch.alerting.modelv2.MonitorV2 +import org.opensearch.alerting.opensearchapi.suspendUntil +import org.opensearch.alerting.service.DeleteMonitorService +import org.opensearch.alerting.settings.AlertingSettings +import org.opensearch.alerting.transport.SecureTransportAction +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.XContentHelper +import org.opensearch.common.xcontent.XContentType +import org.opensearch.commons.alerting.model.ScheduledJob +import org.opensearch.commons.alerting.util.AlertingException +import org.opensearch.core.action.ActionListener +import org.opensearch.core.rest.RestStatus +import org.opensearch.core.xcontent.NamedXContentRegistry +import org.opensearch.tasks.Task +import org.opensearch.transport.TransportService +import org.opensearch.transport.client.Client + +private val scope: CoroutineScope = CoroutineScope(Dispatchers.IO) +private val log = LogManager.getLogger(TransportDeleteMonitorV2Action::class.java) + +class TransportDeleteMonitorV2Action @Inject constructor( + transportService: TransportService, + val client: Client, + actionFilters: ActionFilters, + val clusterService: ClusterService, + settings: Settings, + val xContentRegistry: NamedXContentRegistry +) : HandledTransportAction( + DeleteMonitorV2Action.NAME, transportService, actionFilters, ::DeleteMonitorV2Request +), + SecureTransportAction { + + @Volatile override var filterByEnabled = AlertingSettings.FILTER_BY_BACKEND_ROLES.get(settings) + + init { + listenFilterBySettingChange(clusterService) + } + + override fun doExecute(task: Task, request: DeleteMonitorV2Request, actionListener: ActionListener) { + val user = readUserFromThreadContext(client) + + if (!validateUserBackendRoles(user, actionListener)) { + return + } + + scope.launch { + try { + val monitorV2 = getMonitorV2(request.monitorV2Id, actionListener) ?: return@launch + + val canDelete = user == null || !doFilterForUser(user) || + checkUserPermissionsWithResource(user, monitorV2!!.user, actionListener, "monitor_v2", request.monitorV2Id) + + if (canDelete) { + val deleteResponse = + DeleteMonitorService.deleteMonitorV2(request.monitorV2Id, request.refreshPolicy) + actionListener.onResponse(deleteResponse) + } else { + actionListener.onFailure( + AlertingException( + "Not allowed to delete this Monitor V2", + RestStatus.FORBIDDEN, + IllegalStateException() + ) + ) + } + } catch (e: Exception) { + actionListener.onFailure(e) + } + + // scheduled AlertV2Mover will sweep the alerts and find that this monitor no longer exists, + // and expire this monitor's alerts accordingly + } + } + + private suspend fun getMonitorV2(monitorV2Id: String, actionListener: ActionListener): MonitorV2? { + val getRequest = GetRequest(ScheduledJob.SCHEDULED_JOBS_INDEX, monitorV2Id) + + val getResponse: GetResponse = client.suspendUntil { get(getRequest, it) } + if (!getResponse.isExists) { + actionListener.onFailure( + AlertingException.wrap( + OpenSearchStatusException("Monitor V2 with $monitorV2Id is not found", RestStatus.NOT_FOUND) + ) + ) + return null + } + + val xcp = XContentHelper.createParser( + xContentRegistry, LoggingDeprecationHandler.INSTANCE, + getResponse.sourceAsBytesRef, XContentType.JSON + ) + val scheduledJob = ScheduledJob.parse(xcp, getResponse.id, getResponse.version) + + AlertingV2Utils.validateMonitorV2(scheduledJob)?.let { + actionListener.onFailure(AlertingException.wrap(it)) + return null + } + + val monitorV2 = scheduledJob as MonitorV2 + + return monitorV2 + } +} diff --git a/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/MonitorV2RestApiIT.kt b/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/MonitorV2RestApiIT.kt index de8112722..737dbed6e 100644 --- a/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/MonitorV2RestApiIT.kt +++ b/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/MonitorV2RestApiIT.kt @@ -171,6 +171,16 @@ class MonitorV2RestApiIT : AlertingRestTestCase() { assertEquals("PPL Monitor not found during search", 1, numberDocsFound) } + fun `test delete ppl monitor`() { + val pplMonitor = createRandomPPLMonitor() + + val deleteResponse = client().makeRequest("DELETE", "$MONITOR_V2_BASE_URI/${pplMonitor.id}") + assertEquals("Delete failed", RestStatus.OK, deleteResponse.restStatus()) + + val getResponse = client().makeRequest("HEAD", "$MONITOR_V2_BASE_URI/${pplMonitor.id}") + assertEquals("Deleted monitor still exists", RestStatus.NOT_FOUND, getResponse.restStatus()) + } + fun `test parsing ppl monitor as a scheduled job`() { val monitorV2 = createRandomPPLMonitor() @@ -469,4 +479,15 @@ class MonitorV2RestApiIT : AlertingRestTestCase() { assertEquals("Unexpected status", RestStatus.NOT_FOUND, e.response.restStatus()) } } + + fun `test delete nonexistent ppl monitor fails`() { + val randomId = UUIDs.base64UUID() + + try { + client().makeRequest("DELETE", "$MONITOR_V2_BASE_URI/$randomId") + 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/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/SecureMonitorV2RestApiIT.kt b/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/SecureMonitorV2RestApiIT.kt index 46c7dc02d..bc4178784 100644 --- a/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/SecureMonitorV2RestApiIT.kt +++ b/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/SecureMonitorV2RestApiIT.kt @@ -526,4 +526,97 @@ class SecureMonitorV2RestApiIT : AlertingRestTestCase() { // cleanup searchUserClient.close() } + + fun `test RBAC delete monitorV2 as user with correct backend roles succeeds`() { + enableFilterBy() + if (!isHttps()) { + return + } + val pplMonitorConfig = randomPPLMonitor(enabled = true) + + createUserWithRoles( + user, + listOf(ALERTING_FULL_ACCESS_ROLE, PPL_FULL_ACCESS_ROLE), + listOf("backend_role_a", "backend_role_b"), + false + ) + + val pplMonitor = createMonitorV2WithClient(userClient!!, pplMonitorConfig, listOf("backend_role_a", "backend_role_b")) + + // getUser should have access to the monitor above created by user + val deleteUser = "deleteUser" + + createUserWithRoles( + deleteUser, + listOf(ALERTING_FULL_ACCESS_ROLE, PPL_FULL_ACCESS_ROLE), + listOf("backend_role_a"), + true + ) + + val deleteUserClient = SecureRestClientBuilder(clusterHosts.toTypedArray(), isHttps(), deleteUser, password) + .setSocketTimeout(60000) + .setConnectionRequestTimeout(180000) + .build() + + val getMonitorResponse = deleteUserClient!!.makeRequest( + "DELETE", + "$MONITOR_V2_BASE_URI/${pplMonitor.id}", + null, + BasicHeader(HttpHeaders.CONTENT_TYPE, "application/json") + ) + assertEquals("Get monitorV2 failed", RestStatus.OK, getMonitorResponse.restStatus()) + + ensureNumMonitorV2s(0) + + // cleanup + deleteUserClient.close() + } + + fun `test RBAC delete monitorV2 as user without correct backend roles fails`() { + enableFilterBy() + if (!isHttps()) { + return + } + val pplMonitorConfig = randomPPLMonitor(enabled = true) + + createUserWithRoles( + user, + listOf(ALERTING_FULL_ACCESS_ROLE, PPL_FULL_ACCESS_ROLE), + listOf("backend_role_a", "backend_role_b"), + false + ) + + val pplMonitor = createMonitorV2WithClient(userClient!!, pplMonitorConfig, listOf("backend_role_a", "backend_role_b")) + + // getUser should not have access to the monitor above created by user + val deleteUser = "deleteUser" + + createUserWithRoles( + deleteUser, + listOf(ALERTING_FULL_ACCESS_ROLE, PPL_FULL_ACCESS_ROLE), + listOf("backend_role_c"), + true + ) + + val deleteUserClient = SecureRestClientBuilder(clusterHosts.toTypedArray(), isHttps(), deleteUser, password) + .setSocketTimeout(60000) + .setConnectionRequestTimeout(180000) + .build() + + try { + deleteUserClient!!.makeRequest( + "DELETE", + "$MONITOR_V2_BASE_URI/${pplMonitor.id}", + null, + BasicHeader(HttpHeaders.CONTENT_TYPE, "application/json") + ) + fail("Expected Forbidden exception") + } catch (e: ResponseException) { + assertEquals("Unexpected delete monitor status", RestStatus.FORBIDDEN.status, e.response.statusLine.statusCode) + } finally { + deleteUserClient?.close() + } + + ensureNumMonitorV2s(1) + } }