-
Notifications
You must be signed in to change notification settings - Fork 102
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Added rest layer for the workflow. (#963)
* 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 <buzejic.stevan@gmail.com> Signed-off-by: Surya Sashank Nistala <snistala@amazon.com> Co-authored-by: Stevan Buzejic <buzejic.stevan@gmail.com>
- Loading branch information
Showing
11 changed files
with
2,896 additions
and
8 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.