-
Notifications
You must be signed in to change notification settings - Fork 104
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Browse files
Browse the repository at this point in the history
* Added rest layer for the workflow. Added secure tests (#886) * Added rest layer for the workflow. Added secure tests * add execution_id field in alert mapping --------- Signed-off-by: Stevan Buzejic <[email protected]> Signed-off-by: Surya Sashank Nistala <[email protected]> Co-authored-by: Stevan Buzejic <[email protected]>
- Loading branch information
Showing
11 changed files
with
2,896 additions
and
9 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
60 changes: 60 additions & 0 deletions
60
alerting/src/main/kotlin/org/opensearch/alerting/resthandler/RestDeleteWorkflowAction.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,60 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.alerting.resthandler | ||
|
|
||
| import org.apache.logging.log4j.LogManager | ||
| import org.opensearch.action.support.WriteRequest | ||
| import org.opensearch.alerting.AlertingPlugin | ||
| import org.opensearch.alerting.util.REFRESH | ||
| import org.opensearch.client.node.NodeClient | ||
| import org.opensearch.commons.alerting.action.AlertingActions | ||
| import org.opensearch.commons.alerting.action.DeleteWorkflowRequest | ||
| import org.opensearch.rest.BaseRestHandler | ||
| import org.opensearch.rest.RestHandler | ||
| import org.opensearch.rest.RestRequest | ||
| import org.opensearch.rest.action.RestToXContentListener | ||
| import java.io.IOException | ||
|
|
||
| /** | ||
| * This class consists of the REST handler to delete workflows. | ||
| */ | ||
| class RestDeleteWorkflowAction : BaseRestHandler() { | ||
|
|
||
| private val log = LogManager.getLogger(javaClass) | ||
|
|
||
| override fun getName(): String { | ||
| return "delete_workflow_action" | ||
| } | ||
|
|
||
| override fun routes(): List<RestHandler.Route> { | ||
| return listOf( | ||
| RestHandler.Route( | ||
| RestRequest.Method.DELETE, | ||
| "${AlertingPlugin.WORKFLOW_BASE_URI}/{workflowID}" | ||
| ) | ||
| ) | ||
| } | ||
|
|
||
| @Throws(IOException::class) | ||
| override fun prepareRequest(request: RestRequest, client: NodeClient): RestChannelConsumer { | ||
| log.debug("${request.method()} ${AlertingPlugin.WORKFLOW_BASE_URI}/{workflowID}") | ||
|
|
||
| val workflowId = request.param("workflowID") | ||
| val deleteDelegateMonitors = request.paramAsBoolean("deleteDelegateMonitors", false) | ||
| log.debug("${request.method()} ${request.uri()}") | ||
|
|
||
| val refreshPolicy = | ||
| WriteRequest.RefreshPolicy.parse(request.param(REFRESH, WriteRequest.RefreshPolicy.IMMEDIATE.value)) | ||
| val deleteWorkflowRequest = DeleteWorkflowRequest(workflowId, deleteDelegateMonitors) | ||
|
|
||
| return RestChannelConsumer { channel -> | ||
| client.execute( | ||
| AlertingActions.DELETE_WORKFLOW_ACTION_TYPE, deleteWorkflowRequest, | ||
| RestToXContentListener(channel) | ||
| ) | ||
| } | ||
| } | ||
| } |
59 changes: 59 additions & 0 deletions
59
alerting/src/main/kotlin/org/opensearch/alerting/resthandler/RestGetWorkflowAction.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,59 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.alerting.resthandler | ||
|
|
||
| import org.apache.logging.log4j.LogManager | ||
| import org.opensearch.alerting.AlertingPlugin | ||
| import org.opensearch.alerting.util.context | ||
| import org.opensearch.client.node.NodeClient | ||
| import org.opensearch.commons.alerting.action.AlertingActions | ||
| import org.opensearch.commons.alerting.action.GetWorkflowRequest | ||
| import org.opensearch.rest.BaseRestHandler | ||
| import org.opensearch.rest.RestHandler | ||
| import org.opensearch.rest.RestRequest | ||
| import org.opensearch.rest.action.RestToXContentListener | ||
| import org.opensearch.search.fetch.subphase.FetchSourceContext | ||
|
|
||
| /** | ||
| * This class consists of the REST handler to retrieve a workflow . | ||
| */ | ||
| class RestGetWorkflowAction : BaseRestHandler() { | ||
|
|
||
| private val log = LogManager.getLogger(javaClass) | ||
|
|
||
| override fun getName(): String { | ||
| return "get_workflow_action" | ||
| } | ||
|
|
||
| override fun routes(): List<RestHandler.Route> { | ||
| return listOf( | ||
| RestHandler.Route( | ||
| RestRequest.Method.GET, | ||
| "${AlertingPlugin.WORKFLOW_BASE_URI}/{workflowID}" | ||
| ) | ||
| ) | ||
| } | ||
|
|
||
| override fun prepareRequest(request: RestRequest, client: NodeClient): RestChannelConsumer { | ||
| log.debug("${request.method()} ${AlertingPlugin.WORKFLOW_BASE_URI}/{workflowID}") | ||
|
|
||
| val workflowId = request.param("workflowID") | ||
| if (workflowId == null || workflowId.isEmpty()) { | ||
| throw IllegalArgumentException("missing id") | ||
| } | ||
|
|
||
| var srcContext = context(request) | ||
| if (request.method() == RestRequest.Method.HEAD) { | ||
| srcContext = FetchSourceContext.DO_NOT_FETCH_SOURCE | ||
| } | ||
| val getWorkflowRequest = | ||
| GetWorkflowRequest(workflowId, request.method()) | ||
| return RestChannelConsumer { | ||
| channel -> | ||
| client.execute(AlertingActions.GET_WORKFLOW_ACTION_TYPE, getWorkflowRequest, RestToXContentListener(channel)) | ||
| } | ||
| } | ||
| } |
99 changes: 99 additions & 0 deletions
99
alerting/src/main/kotlin/org/opensearch/alerting/resthandler/RestIndexWorkflowAction.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,99 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
| package org.opensearch.alerting.resthandler | ||
|
|
||
| import org.opensearch.action.support.WriteRequest | ||
| import org.opensearch.alerting.AlertingPlugin | ||
| import org.opensearch.alerting.util.AlertingException | ||
| 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.client.node.NodeClient | ||
| import org.opensearch.common.xcontent.XContentParserUtils | ||
| import org.opensearch.commons.alerting.action.AlertingActions | ||
| import org.opensearch.commons.alerting.action.IndexWorkflowRequest | ||
| import org.opensearch.commons.alerting.action.IndexWorkflowResponse | ||
| import org.opensearch.commons.alerting.model.Workflow | ||
| import org.opensearch.core.xcontent.ToXContent | ||
| import org.opensearch.core.xcontent.XContentParser | ||
| import org.opensearch.index.seqno.SequenceNumbers | ||
| import org.opensearch.rest.BaseRestHandler | ||
| import org.opensearch.rest.BaseRestHandler.RestChannelConsumer | ||
| import org.opensearch.rest.BytesRestResponse | ||
| import org.opensearch.rest.RestChannel | ||
| import org.opensearch.rest.RestHandler | ||
| import org.opensearch.rest.RestRequest | ||
| import org.opensearch.rest.RestResponse | ||
| import org.opensearch.rest.RestStatus | ||
| import org.opensearch.rest.action.RestResponseListener | ||
| import java.io.IOException | ||
| import java.time.Instant | ||
|
|
||
| /** | ||
| * Rest handlers to create and update workflows. | ||
| */ | ||
| class RestIndexWorkflowAction : BaseRestHandler() { | ||
|
|
||
| override fun getName(): String { | ||
| return "index_workflow_action" | ||
| } | ||
|
|
||
| override fun routes(): List<RestHandler.Route> { | ||
| return listOf( | ||
| RestHandler.Route(RestRequest.Method.POST, AlertingPlugin.WORKFLOW_BASE_URI), | ||
| RestHandler.Route( | ||
| RestRequest.Method.PUT, | ||
| "${AlertingPlugin.WORKFLOW_BASE_URI}/{workflowID}" | ||
| ) | ||
| ) | ||
| } | ||
|
|
||
| @Throws(IOException::class) | ||
| override fun prepareRequest(request: RestRequest, client: NodeClient): RestChannelConsumer { | ||
| val id = request.param("workflowID", Workflow.NO_ID) | ||
| if (request.method() == RestRequest.Method.PUT && Workflow.NO_ID == id) { | ||
| throw AlertingException.wrap(IllegalArgumentException("Missing workflow ID")) | ||
| } | ||
|
|
||
| // Validate request by parsing JSON to Monitor | ||
| val xcp = request.contentParser() | ||
| XContentParserUtils.ensureExpectedToken(XContentParser.Token.START_OBJECT, xcp.nextToken(), xcp) | ||
| val workflow = Workflow.parse(xcp, id).copy(lastUpdateTime = Instant.now()) | ||
| val rbacRoles = request.contentParser().map()["rbac_roles"] as List<String>? | ||
|
|
||
| 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 workflowRequest = | ||
| IndexWorkflowRequest(id, seqNo, primaryTerm, refreshPolicy, request.method(), workflow, rbacRoles) | ||
|
|
||
| return RestChannelConsumer { channel -> | ||
| client.execute(AlertingActions.INDEX_WORKFLOW_ACTION_TYPE, workflowRequest, indexMonitorResponse(channel, request.method())) | ||
| } | ||
| } | ||
|
|
||
| private fun indexMonitorResponse(channel: RestChannel, restMethod: RestRequest.Method): RestResponseListener<IndexWorkflowResponse> { | ||
| return object : RestResponseListener<IndexWorkflowResponse>(channel) { | ||
| @Throws(Exception::class) | ||
| override fun buildResponse(response: IndexWorkflowResponse): RestResponse { | ||
| var returnStatus = RestStatus.CREATED | ||
| if (restMethod == RestRequest.Method.PUT) | ||
| returnStatus = RestStatus.OK | ||
|
|
||
| val restResponse = | ||
| BytesRestResponse(returnStatus, response.toXContent(channel.newBuilder(), ToXContent.EMPTY_PARAMS)) | ||
| if (returnStatus == RestStatus.CREATED) { | ||
| val location = "${AlertingPlugin.WORKFLOW_BASE_URI}/${response.id}" | ||
| restResponse.addHeader("Location", location) | ||
| } | ||
| return restResponse | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.