Added integrations tests for checking workflow creation and update sc… #1
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
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 |
|---|---|---|
| @@ -1,4 +1,3 @@ | ||
| package org.opensearch.alerting.model.workflow | ||
|
|
||
| data class WorkflowRunResult(private val someArg: String) | ||
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,145 @@ | ||
| /* | ||
|
There was a problem hiding this comment. We also need to make changes in TransportDeleteMonitorAction to stop a monitor from being deleted if its a part of an existing composite workflow sequence as a delegate monitor. |
||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.alerting.transport | ||
|
|
||
| import kotlinx.coroutines.CoroutineName | ||
| import kotlinx.coroutines.Dispatchers | ||
| import kotlinx.coroutines.GlobalScope | ||
| import kotlinx.coroutines.launch | ||
| import org.opensearch.OpenSearchStatusException | ||
| import org.opensearch.action.ActionListener | ||
| import org.opensearch.action.ActionRequest | ||
| import org.opensearch.action.delete.DeleteRequest | ||
| import org.opensearch.action.delete.DeleteResponse | ||
| 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.opensearchapi.suspendUntil | ||
| import org.opensearch.alerting.settings.AlertingSettings | ||
| import org.opensearch.alerting.util.AlertingException | ||
| import org.opensearch.client.Client | ||
| 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.NamedXContentRegistry | ||
| import org.opensearch.common.xcontent.XContentHelper | ||
| import org.opensearch.common.xcontent.XContentType | ||
| import org.opensearch.commons.alerting.action.AlertingActions | ||
| import org.opensearch.commons.alerting.action.DeleteWorkflowRequest | ||
| import org.opensearch.commons.alerting.action.DeleteWorkflowResponse | ||
| import org.opensearch.commons.alerting.model.ScheduledJob | ||
| import org.opensearch.commons.alerting.model.Workflow | ||
| import org.opensearch.commons.authuser.User | ||
| import org.opensearch.commons.utils.recreateObject | ||
| import org.opensearch.rest.RestStatus | ||
| import org.opensearch.tasks.Task | ||
| import org.opensearch.transport.TransportService | ||
|
|
||
|
There was a problem hiding this comment. let's add java docs for classes and comments for methods having complicated logic There was a problem hiding this comment. I agree. This class is pretty much obvious and the methods inside it are pretty much descriptive. So probably for indexing the workflow |
||
| class TransportDeleteWorkflowAction @Inject constructor( | ||
| transportService: TransportService, | ||
| val client: Client, | ||
| actionFilters: ActionFilters, | ||
| val clusterService: ClusterService, | ||
| settings: Settings, | ||
| val xContentRegistry: NamedXContentRegistry | ||
| ) : HandledTransportAction<ActionRequest, DeleteWorkflowResponse>( | ||
| AlertingActions.DELETE_WORKFLOW_ACTION_NAME, transportService, actionFilters, ::DeleteWorkflowRequest | ||
| ), | ||
| SecureTransportAction { | ||
|
|
||
| @Volatile override var filterByEnabled = AlertingSettings.FILTER_BY_BACKEND_ROLES.get(settings) | ||
|
|
||
| init { | ||
| listenFilterBySettingChange(clusterService) | ||
| } | ||
|
|
||
| override fun doExecute(task: Task, request: ActionRequest, actionListener: ActionListener<DeleteWorkflowResponse>) { | ||
| val transformedRequest = request as? DeleteWorkflowRequest | ||
| ?: recreateObject(request) { DeleteWorkflowRequest(it) } | ||
|
|
||
| val user = readUserFromThreadContext(client) | ||
| val deleteRequest = DeleteRequest(ScheduledJob.SCHEDULED_JOBS_INDEX, transformedRequest.workflowId) | ||
| .setRefreshPolicy(transformedRequest.refreshPolicy) | ||
|
There was a problem hiding this comment. plz remove refresh policy |
||
|
|
||
| if (!validateUserBackendRoles(user, actionListener)) { | ||
| return | ||
| } | ||
|
|
||
| GlobalScope.launch(Dispatchers.IO + CoroutineName("DeleteWorkflowAction")) { | ||
| DeleteWorkflowHandler(client, actionListener, deleteRequest, user, transformedRequest.workflowId).resolveUserAndStart() | ||
| } | ||
| } | ||
|
|
||
| inner class DeleteWorkflowHandler( | ||
| private val client: Client, | ||
| private val actionListener: ActionListener<DeleteWorkflowResponse>, | ||
| private val deleteRequest: DeleteRequest, | ||
| private val user: User?, | ||
| private val workflowId: String | ||
| ) { | ||
| suspend fun resolveUserAndStart() { | ||
| try { | ||
| val workflow = getWorkflow() | ||
|
|
||
| val canDelete = user == null || | ||
| !doFilterForUser(user) || | ||
| checkUserPermissionsWithResource( | ||
| user, | ||
| workflow.user, | ||
| actionListener, | ||
| "workflow", | ||
| workflowId | ||
| ) | ||
|
|
||
| if (canDelete) { | ||
| val deleteResponse = deleteWorkflow(workflow) | ||
| // TODO - uncomment once the workflow metadata is added | ||
|
There was a problem hiding this comment. Well no - because the workflow metadata are not added still |
||
| // deleteMetadata(workflow) | ||
| actionListener.onResponse(DeleteWorkflowResponse(deleteResponse.id, deleteResponse.version)) | ||
| } else { | ||
| actionListener.onFailure( | ||
| AlertingException( | ||
| "Not allowed to delete this workflow!", | ||
| RestStatus.FORBIDDEN, | ||
| IllegalStateException() | ||
| ) | ||
| ) | ||
| } | ||
| } catch (t: Exception) { | ||
| actionListener.onFailure(AlertingException.wrap(t)) | ||
| } | ||
| } | ||
|
|
||
| private suspend fun getWorkflow(): Workflow { | ||
| val getRequest = GetRequest(ScheduledJob.SCHEDULED_JOBS_INDEX, workflowId) | ||
|
|
||
| val getResponse: GetResponse = client.suspendUntil { get(getRequest, it) } | ||
| if (getResponse.isExists == false) { | ||
| actionListener.onFailure( | ||
| AlertingException.wrap( | ||
| OpenSearchStatusException("Workflow with $workflowId is not found", RestStatus.NOT_FOUND) | ||
| ) | ||
| ) | ||
| } | ||
| val xcp = XContentHelper.createParser( | ||
| xContentRegistry, LoggingDeprecationHandler.INSTANCE, | ||
| getResponse.sourceAsBytesRef, XContentType.JSON | ||
| ) | ||
| return ScheduledJob.parse(xcp, getResponse.id, getResponse.version) as Workflow | ||
| } | ||
|
|
||
| private suspend fun deleteWorkflow(workflow: Workflow): DeleteResponse { | ||
| return client.suspendUntil { delete(deleteRequest, it) } | ||
|
There was a problem hiding this comment. how are we handling failure in deletion? There was a problem hiding this comment. add debug logs "deleted monitor $monitorId" There was a problem hiding this comment.
Well we are calling delete(deleteRequest, it) as a lambda. You can see that second parameter of delete function call is it -> which is ActionListener forwarded to a delete function. |
||
| } | ||
|
|
||
| private suspend fun deleteMetadata(workflow: Workflow) { | ||
| val deleteRequest = DeleteRequest(ScheduledJob.SCHEDULED_JOBS_INDEX, "${workflow.id}-metadata") | ||
|
There was a problem hiding this comment. add debug logs "deleted monitor $monitorMetadata" There was a problem hiding this comment. So just FYI - right now there is no existing workflow metadata. This method is not called at all. Once we introduce workflow metadata I will add error handling. How this sounds? |
||
| val deleteResponse: DeleteResponse = client.suspendUntil { delete(deleteRequest, it) } | ||
| } | ||
| } | ||
| } | ||
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.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
we should return list of monitor run results within this class with info from each delegate monitor run.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yeah - I agree. That's in another PR that contains execution of the workflow changes. Check it out here