Skip to content

Commit

Permalink
Combine create api with provision api by adding a provision param (op…
Browse files Browse the repository at this point in the history 
…ensearch-project#282)

* replace dryrun parameter with provision in create workflow

Signed-off-by: Jackie Han <jkhanjob@gmail.com>

* test

Signed-off-by: Jackie Han <jkhanjob@gmail.com>

* test

Signed-off-by: Jackie Han <jkhanjob@gmail.com>

* test

Signed-off-by: Jackie Han <jkhanjob@gmail.com>

* Combine create api with provision api by adding a provision param

Signed-off-by: Jackie Han <jkhanjob@gmail.com>

* cleanup

Signed-off-by: Jackie Han <jkhanjob@gmail.com>

* keep dryrun option in create workflow

Signed-off-by: Jackie Han <jkhanjob@gmail.com>

* cleanup

Signed-off-by: Jackie Han <jkhanjob@gmail.com>

* keep both dryRun and provision parameter

Signed-off-by: Jackie Han <jkhanjob@gmail.com>

---------

Signed-off-by: Jackie Han <jkhanjob@gmail.com>
  • Loading branch information
@jackiehanyang @dbwiddis
jackiehanyang authored and dbwiddis committed Dec 15, 2023
1 parent c17fa51 commit 531ef4c
Show file tree
Hide file tree
Showing 4 changed files with 235 additions and 9 deletions.
4 changes: 3 additions & 1 deletion src/main/java/org/opensearch/flowframework/rest/RestCreateWorkflowAction.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
import java.util.Locale;

import static org.opensearch.flowframework.common.CommonValue.DRY_RUN;
import static org.opensearch.flowframework.common.CommonValue.PROVISION_WORKFLOW;
import static org.opensearch.flowframework.common.CommonValue.WORKFLOW_ID;
import static org.opensearch.flowframework.common.CommonValue.WORKFLOW_URI;
import static org.opensearch.flowframework.common.FlowFrameworkSettings.FLOW_FRAMEWORK_ENABLED;
Expand Down Expand Up @@ -91,8 +92,9 @@ protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient cli
String workflowId = request.param(WORKFLOW_ID);
Template template = Template.parse(request.content().utf8ToString());
boolean dryRun = request.paramAsBoolean(DRY_RUN, false);
boolean provision = request.paramAsBoolean(PROVISION_WORKFLOW, false);

WorkflowRequest workflowRequest = new WorkflowRequest(workflowId, template, dryRun, requestTimeout, maxWorkflows);
WorkflowRequest workflowRequest = new WorkflowRequest(workflowId, template, dryRun, provision, requestTimeout, maxWorkflows);

return channel -> client.execute(CreateWorkflowAction.INSTANCE, workflowRequest, ActionListener.wrap(response -> {
XContentBuilder builder = response.toXContent(channel.newBuilder(), ToXContent.EMPTY_PARAMS);
Expand Down
24 changes: 22 additions & 2 deletions src/main/java/org/opensearch/flowframework/transport/CreateWorkflowTransportAction.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -135,7 +135,28 @@ protected void doExecute(Task task, WorkflowRequest request, ActionListener<Work
user,
ActionListener.wrap(stateResponse -> {
logger.info("create state workflow doc");
listener.onResponse(new WorkflowResponse(globalContextResponse.getId()));
if (request.isProvision()) {
logger.info("provision parameter");
WorkflowRequest workflowRequest = new WorkflowRequest(globalContextResponse.getId(), null);
client.execute(
ProvisionWorkflowAction.INSTANCE,
workflowRequest,
ActionListener.wrap(provisionResponse -> {
listener.onResponse(new WorkflowResponse(provisionResponse.getWorkflowId()));
}, exception -> {
if (exception instanceof FlowFrameworkException) {
listener.onFailure(exception);
} else {
listener.onFailure(
new FlowFrameworkException(exception.getMessage(), RestStatus.BAD_REQUEST)
);
}
logger.error("Failed to send back provision workflow exception", exception);
})
);
} else {
listener.onResponse(new WorkflowResponse(globalContextResponse.getId()));
}
}, exception -> {
logger.error("Failed to save workflow state : {}", exception.getMessage());
if (exception instanceof FlowFrameworkException) {
Expand Down Expand Up @@ -246,5 +267,4 @@ private void validateWorkflows(Template template) throws Exception {
workflowProcessSorter.validateGraph(sortedNodes);
}
}

}
22 changes: 20 additions & 2 deletions src/main/java/org/opensearch/flowframework/transport/WorkflowRequest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,11 @@ public class WorkflowRequest extends ActionRequest {
*/
private boolean dryRun;

/**
* Provision flag
*/
private boolean provision;

/**
* Timeout for request
*/
Expand All @@ -54,7 +59,7 @@ public class WorkflowRequest extends ActionRequest {
* @param template the use case template which describes the workflow
*/
public WorkflowRequest(@Nullable String workflowId, @Nullable Template template) {
this(workflowId, template, false, null, null);
this(workflowId, template, false, false, null, null);
}

/**
Expand All @@ -70,27 +75,30 @@ public WorkflowRequest(
@Nullable TimeValue requestTimeout,
@Nullable Integer maxWorkflows
) {
this(workflowId, template, false, requestTimeout, maxWorkflows);
this(workflowId, template, false, false, requestTimeout, maxWorkflows);
}

/**
* Instantiates a new WorkflowRequest
* @param workflowId the documentId of the workflow
* @param template the use case template which describes the workflow
* @param dryRun flag to indicate if validation is necessary
* @param provision flag to indicate if provision is necessary
* @param requestTimeout timeout of the request
* @param maxWorkflows max number of workflows
*/
public WorkflowRequest(
@Nullable String workflowId,
@Nullable Template template,
boolean dryRun,
boolean provision,
@Nullable TimeValue requestTimeout,
@Nullable Integer maxWorkflows
) {
this.workflowId = workflowId;
this.template = template;
this.dryRun = dryRun;
this.provision = provision;
this.requestTimeout = requestTimeout;
this.maxWorkflows = maxWorkflows;
}
Expand All @@ -106,6 +114,7 @@ public WorkflowRequest(StreamInput in) throws IOException {
String templateJson = in.readOptionalString();
this.template = templateJson == null ? null : Template.parse(templateJson);
this.dryRun = in.readBoolean();
this.provision = in.readBoolean();
this.requestTimeout = in.readOptionalTimeValue();
this.maxWorkflows = in.readOptionalInt();
}
Expand Down Expand Up @@ -136,6 +145,14 @@ public boolean isDryRun() {
return this.dryRun;
}

/**
* Gets the provision flag
* @return the provision boolean
*/
public boolean isProvision() {
return this.provision;
}

/**
* Gets the timeout of the request
* @return the requestTimeout
Expand All @@ -158,6 +175,7 @@ public void writeTo(StreamOutput out) throws IOException {
out.writeOptionalString(workflowId);
out.writeOptionalString(template == null ? null : template.toJson());
out.writeBoolean(dryRun);
out.writeBoolean(provision);
out.writeOptionalTimeValue(requestTimeout);
out.writeOptionalInt(maxWorkflows);
}
Expand Down
Loading

0 comments on commit 531ef4c

Please sign in to comment.