-
Notifications
You must be signed in to change notification settings - Fork 31
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[POC] Adding Orchestrate API and Search Response Processor Step (#619)
* Fixing ingest pipeline integ test (#614) Signed-off-by: Joshua Palis <[email protected]> * Adding initial rest, transport actions for orchestration. Search response processor step Signed-off-by: Joshua Palis <[email protected]> * Fixing transport action Signed-off-by: Joshua Palis <[email protected]> * reverting change to resttestcase Signed-off-by: Joshua Palis <[email protected]> * Adding javadocs Signed-off-by: Joshua Palis <[email protected]> * fixing checkstyle Signed-off-by: Joshua Palis <[email protected]> * removing extra common value Signed-off-by: Joshua Palis <[email protected]> * Fixing errors Signed-off-by: Joshua Palis <[email protected]> --------- Signed-off-by: Joshua Palis <[email protected]>
- Loading branch information
Showing
14 changed files
with
597 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
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
110 changes: 110 additions & 0 deletions
110
src/main/java/org/opensearch/flowframework/rest/RestOrchestrateAction.java
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,110 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * The OpenSearch Contributors require contributions made to | ||
| * this file be licensed under the Apache-2.0 license or a | ||
| * compatible open source license. | ||
| */ | ||
| package org.opensearch.flowframework.rest; | ||
|
|
||
| import org.apache.logging.log4j.LogManager; | ||
| import org.apache.logging.log4j.Logger; | ||
| import org.opensearch.ExceptionsHelper; | ||
| import org.opensearch.client.node.NodeClient; | ||
| import org.opensearch.core.action.ActionListener; | ||
| import org.opensearch.core.rest.RestStatus; | ||
| import org.opensearch.core.xcontent.ToXContent; | ||
| import org.opensearch.core.xcontent.XContentBuilder; | ||
| import org.opensearch.core.xcontent.XContentParser; | ||
| import org.opensearch.flowframework.exception.FlowFrameworkException; | ||
| import org.opensearch.flowframework.transport.OrchestrateAction; | ||
| import org.opensearch.flowframework.transport.OrchestrateRequest; | ||
| import org.opensearch.flowframework.util.ParseUtils; | ||
| import org.opensearch.rest.BaseRestHandler; | ||
| import org.opensearch.rest.BytesRestResponse; | ||
| import org.opensearch.rest.RestRequest; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.Collections; | ||
| import java.util.List; | ||
| import java.util.Locale; | ||
| import java.util.Map; | ||
|
|
||
| import static org.opensearch.core.xcontent.XContentParserUtils.ensureExpectedToken; | ||
| import static org.opensearch.flowframework.common.CommonValue.WORKFLOW_ID; | ||
| import static org.opensearch.flowframework.common.CommonValue.WORKFLOW_URI; | ||
|
|
||
| /** | ||
| * Rest action to orchestate workflows | ||
| */ | ||
| public class RestOrchestrateAction extends BaseRestHandler { | ||
|
|
||
| private static final Logger logger = LogManager.getLogger(RestProvisionWorkflowAction.class); | ||
|
|
||
| private static final String ORCHESTRATE_ACTION = "orchestrate_action"; | ||
|
|
||
| /** | ||
| * Creates a new RestOrchestrateAction instance | ||
| */ | ||
| public RestOrchestrateAction() {} | ||
|
|
||
| @Override | ||
| public String getName() { | ||
| return ORCHESTRATE_ACTION; | ||
| } | ||
|
|
||
| @Override | ||
| public List<Route> routes() { | ||
| return List.of( | ||
| new Route(RestRequest.Method.POST, String.format(Locale.ROOT, "%s/{%s}/%s", WORKFLOW_URI, WORKFLOW_ID, "_orchestrate")) | ||
| ); | ||
| } | ||
|
|
||
| @Override | ||
| protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException { | ||
|
|
||
| // Get workflow ID | ||
| String workflowId = request.param(WORKFLOW_ID); | ||
|
|
||
| try { | ||
|
|
||
| // Validate params | ||
| if (workflowId == null) { | ||
| throw new FlowFrameworkException("workflow_id cannot be null", RestStatus.BAD_REQUEST); | ||
| } | ||
|
|
||
| // Retrieve string to string map from content | ||
| Map<String, String> userInputs = Collections.emptyMap(); | ||
| if (request.hasContent()) { | ||
| XContentParser parser = request.contentParser(); | ||
| ensureExpectedToken(XContentParser.Token.START_OBJECT, parser.nextToken(), parser); | ||
| userInputs = ParseUtils.parseStringToStringMap(parser); | ||
| } | ||
| // Create Request object and execute transport action with client to pass ID and values | ||
| OrchestrateRequest orchestrateRequest = new OrchestrateRequest(workflowId, userInputs); | ||
| return channel -> client.execute(OrchestrateAction.INSTANCE, orchestrateRequest, ActionListener.wrap(response -> { | ||
| XContentBuilder builder = response.toXContent(channel.newBuilder(), ToXContent.EMPTY_PARAMS); | ||
| channel.sendResponse(new BytesRestResponse(RestStatus.OK, builder)); | ||
| }, exception -> { | ||
| try { | ||
| FlowFrameworkException ex = exception instanceof FlowFrameworkException | ||
| ? (FlowFrameworkException) exception | ||
| : new FlowFrameworkException("Failed to get workflow.", ExceptionsHelper.status(exception)); | ||
| XContentBuilder exceptionBuilder = ex.toXContent(channel.newErrorBuilder(), ToXContent.EMPTY_PARAMS); | ||
| channel.sendResponse(new BytesRestResponse(ex.getRestStatus(), exceptionBuilder)); | ||
| } catch (IOException e) { | ||
| String errorMessage = "IOException: Failed to send back orchestrate exception"; | ||
| logger.error(errorMessage, e); | ||
| channel.sendResponse(new BytesRestResponse(ExceptionsHelper.status(e), errorMessage)); | ||
| } | ||
| })); | ||
|
|
||
| } catch (FlowFrameworkException ex) { | ||
| return channel -> channel.sendResponse( | ||
| new BytesRestResponse(ex.getRestStatus(), ex.toXContent(channel.newErrorBuilder(), ToXContent.EMPTY_PARAMS)) | ||
| ); | ||
| } | ||
|
|
||
| } | ||
| } |
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
30 changes: 30 additions & 0 deletions
30
src/main/java/org/opensearch/flowframework/transport/OrchestrateAction.java
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,30 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * The OpenSearch Contributors require contributions made to | ||
| * this file be licensed under the Apache-2.0 license or a | ||
| * compatible open source license. | ||
| */ | ||
| package org.opensearch.flowframework.transport; | ||
|
|
||
| import org.opensearch.action.ActionType; | ||
| import org.opensearch.action.search.SearchResponse; | ||
|
|
||
| import static org.opensearch.flowframework.common.CommonValue.TRANSPORT_ACTION_NAME_PREFIX; | ||
|
|
||
| /** | ||
| * External action for public facing RestOrchestrateAction | ||
| */ | ||
| public class OrchestrateAction extends ActionType<SearchResponse> { | ||
|
|
||
| /** The name of this action */ | ||
| public static final String NAME = TRANSPORT_ACTION_NAME_PREFIX + "workflow/orchestrate"; | ||
|
|
||
| /** An instance of this action */ | ||
| public static final OrchestrateAction INSTANCE = new OrchestrateAction(); | ||
|
|
||
| private OrchestrateAction() { | ||
| super(NAME, SearchResponse::new); | ||
| } | ||
| } |
82 changes: 82 additions & 0 deletions
82
src/main/java/org/opensearch/flowframework/transport/OrchestrateRequest.java
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,82 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * The OpenSearch Contributors require contributions made to | ||
| * this file be licensed under the Apache-2.0 license or a | ||
| * compatible open source license. | ||
| */ | ||
| package org.opensearch.flowframework.transport; | ||
|
|
||
| import org.opensearch.action.ActionRequest; | ||
| import org.opensearch.action.ActionRequestValidationException; | ||
| import org.opensearch.core.common.io.stream.StreamInput; | ||
| import org.opensearch.core.common.io.stream.StreamOutput; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.Map; | ||
|
|
||
| /** | ||
| * Transport request for orchestrate API | ||
| */ | ||
| public class OrchestrateRequest extends ActionRequest { | ||
|
|
||
| /** | ||
| * The documentId of the workflow entry within the Global Context index | ||
| */ | ||
| private String workflowId; | ||
|
|
||
| /** | ||
| * User inputs map | ||
| */ | ||
| private Map<String, String> userInputs; | ||
|
|
||
| /** | ||
| * Creates a new orchestrate request | ||
| * @param workflowId the workflow ID | ||
| * @param userInputs the user inputs to substitute values for in the template | ||
| */ | ||
| public OrchestrateRequest(String workflowId, Map<String, String> userInputs) { | ||
| this.workflowId = workflowId; | ||
| this.userInputs = userInputs; | ||
| } | ||
|
|
||
| /** | ||
| * Creates a new orchestrate request from stream input | ||
| * @param in the stream input | ||
| * @throws IOException on error reading from the stream input | ||
| */ | ||
| public OrchestrateRequest(StreamInput in) throws IOException { | ||
| super(in); | ||
| this.workflowId = in.readString(); | ||
| this.userInputs = in.readMap(StreamInput::readString, StreamInput::readString); | ||
| } | ||
|
|
||
| /** | ||
| * Returns the workflow ID | ||
| * @return the workflow ID | ||
| */ | ||
| public String getWorkflowId() { | ||
| return this.workflowId; | ||
| } | ||
|
|
||
| /** | ||
| * Returns the user inputs map | ||
| * @return the user inputs map | ||
| */ | ||
| public Map<String, String> getUserInputs() { | ||
| return this.userInputs; | ||
| } | ||
|
|
||
| @Override | ||
| public void writeTo(StreamOutput out) throws IOException { | ||
| super.writeTo(out); | ||
| out.writeString(workflowId); | ||
| out.writeMap(userInputs, StreamOutput::writeString, StreamOutput::writeString); | ||
| } | ||
|
|
||
| @Override | ||
| public ActionRequestValidationException validate() { | ||
| return null; | ||
| } | ||
| } |
Oops, something went wrong.