Updates exception handling with FlowFrameworkException AND adds dryrun param to Create Workflow

src/main/java/org/opensearch/flowframework/common/CommonValue.java

@@ -53,6 +53,8 @@ private CommonValue() {}
     public static final String WORKFLOW_URI = FLOW_FRAMEWORK_BASE_URI + "/workflow";
     /** Field name for workflow Id, the document Id of the indexed use case template */
     public static final String WORKFLOW_ID = "workflow_id";
+    /** Field name for dry run, the flag to indicate if validation is necessary */
+    public static final String DRY_RUN = "dryrun";
     /** The field name for provision workflow within a use case template*/
     public static final String PROVISION_WORKFLOW = "provision";
 

src/main/java/org/opensearch/flowframework/rest/RestCreateWorkflowAction.java

@@ -28,6 +28,7 @@
 import java.util.List;
 import java.util.Locale;
 
+import static org.opensearch.flowframework.common.CommonValue.DRY_RUN;
 import static org.opensearch.flowframework.common.CommonValue.WORKFLOW_ID;
 import static org.opensearch.flowframework.common.CommonValue.WORKFLOW_URI;
 
@@ -62,21 +63,25 @@
     @Override
     protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
         try {
+
             String workflowId = request.param(WORKFLOW_ID);
             Template template = Template.parse(request.content().utf8ToString());
-            WorkflowRequest workflowRequest = new WorkflowRequest(workflowId, template);
+            boolean dryRun = request.paramAsBoolean(DRY_RUN, false);
+
+            WorkflowRequest workflowRequest = new WorkflowRequest(workflowId, template, dryRun);
+
             return channel -> client.execute(CreateWorkflowAction.INSTANCE, workflowRequest, ActionListener.wrap(response -> {
                 XContentBuilder builder = response.toXContent(channel.newBuilder(), ToXContent.EMPTY_PARAMS);
                 channel.sendResponse(new BytesRestResponse(RestStatus.CREATED, builder));
             }, exception -> {
                 try {
                     FlowFrameworkException ex = (FlowFrameworkException) exception;
                     XContentBuilder exceptionBuilder = ex.toXContent(channel.newErrorBuilder(), ToXContent.EMPTY_PARAMS);
                     channel.sendResponse(new BytesRestResponse(ex.getRestStatus(), exceptionBuilder));
                 } catch (IOException e) {
                     logger.error("Failed to send back create workflow exception", e);
                 }
             }));
         } catch (Exception e) {
             FlowFrameworkException ex = new FlowFrameworkException(e.getMessage(), RestStatus.BAD_REQUEST);
             return channel -> channel.sendResponse(

src/main/java/org/opensearch/flowframework/transport/CreateWorkflowTransportAction.java

@@ -24,9 +24,14 @@
 import org.opensearch.flowframework.model.ProvisioningProgress;
 import org.opensearch.flowframework.model.State;
 import org.opensearch.flowframework.model.Template;
+import org.opensearch.flowframework.model.Workflow;
+import org.opensearch.flowframework.workflow.ProcessNode;
+import org.opensearch.flowframework.workflow.WorkflowProcessSorter;
 import org.opensearch.tasks.Task;
 import org.opensearch.transport.TransportService;
 
+import java.util.List;
+
 import static org.opensearch.flowframework.common.CommonValue.PROVISIONING_PROGRESS_FIELD;
 import static org.opensearch.flowframework.common.CommonValue.STATE_FIELD;
 import static org.opensearch.flowframework.common.CommonValue.WORKFLOW_STATE_INDEX;
@@ -39,30 +44,35 @@
 
     private final Logger logger = LogManager.getLogger(CreateWorkflowTransportAction.class);
 
+    private final WorkflowProcessSorter workflowProcessSorter;
     private final FlowFrameworkIndicesHandler flowFrameworkIndicesHandler;
     private final Client client;
 
     /**
      * Intantiates a new CreateWorkflowTransportAction
      * @param transportService the TransportService
      * @param actionFilters action filters
+     * @param workflowProcessSorter the workflow process sorter
      * @param flowFrameworkIndicesHandler The handler for the global context index
      * @param client The client used to make the request to OS
      */
     @Inject
     public CreateWorkflowTransportAction(
         TransportService transportService,
         ActionFilters actionFilters,
+        WorkflowProcessSorter workflowProcessSorter,
         FlowFrameworkIndicesHandler flowFrameworkIndicesHandler,
         Client client
     ) {
         super(CreateWorkflowAction.NAME, transportService, actionFilters, WorkflowRequest::new);
+        this.workflowProcessSorter = workflowProcessSorter;
         this.flowFrameworkIndicesHandler = flowFrameworkIndicesHandler;
         this.client = client;
     }
 
     @Override
     protected void doExecute(Task task, WorkflowRequest request, ActionListener<WorkflowResponse> listener) {
+
         User user = getUserContext(client);
         Template templateWithUser = new Template(
             request.getTemplate().name(),
@@ -73,6 +83,21 @@
             request.getTemplate().workflows(),
             user
         );
+
+        if (request.isDryRun()) {
+            try {
+                validateWorkflows(templateWithUser);
+            } catch (Exception e) {
+                if (e instanceof FlowFrameworkException) {
+                    logger.error("Workflow validation failed for template : " + templateWithUser.name());
+                    listener.onFailure(e);
+                } else {
+                    listener.onFailure(new FlowFrameworkException(e.getMessage(), ExceptionsHelper.status(e)));
+                }
+                return;
+            }
+        }
+
         if (request.getWorkflowId() == null) {
             // Create new global context and state index entries
             flowFrameworkIndicesHandler.putTemplateToGlobalContext(templateWithUser, ActionListener.wrap(globalContextResponse -> {
@@ -85,16 +110,16 @@
                     }, exception -> {
                         logger.error("Failed to save workflow state : {}", exception.getMessage());
                         if (exception instanceof FlowFrameworkException) {
                             listener.onFailure(exception);
                         } else {
                             listener.onFailure(new FlowFrameworkException(exception.getMessage(), RestStatus.BAD_REQUEST));
                         }
                     })
                 );
             }, exception -> {
                 logger.error("Failed to save use case template : {}", exception.getMessage());
                 if (exception instanceof FlowFrameworkException) {
                     listener.onFailure(exception);
                 } else {
                     listener.onFailure(new FlowFrameworkException(exception.getMessage(), ExceptionsHelper.status(exception)));
                 }
@@ -116,16 +141,16 @@
                         }, exception -> {
                             logger.error("Failed to update workflow state : {}", exception.getMessage());
                             if (exception instanceof FlowFrameworkException) {
                                 listener.onFailure(exception);
                             } else {
                                 listener.onFailure(new FlowFrameworkException(exception.getMessage(), ExceptionsHelper.status(exception)));
                             }
                         })
                     );
                 }, exception -> {
                     logger.error("Failed to updated use case template {} : {}", request.getWorkflowId(), exception.getMessage());
                     if (exception instanceof FlowFrameworkException) {
                         listener.onFailure(exception);
                     } else {
                         listener.onFailure(new FlowFrameworkException(exception.getMessage(), ExceptionsHelper.status(exception)));
                     }
@@ -135,4 +160,11 @@
         }
     }
 
+    private void validateWorkflows(Template template) throws Exception {
+        for (Workflow workflow : template.workflows().values()) {
+            List<ProcessNode> sortedNodes = workflowProcessSorter.sortProcessNodes(workflow);
+            workflowProcessSorter.validateGraph(sortedNodes);
+        }
+    }
+
 }

src/main/java/org/opensearch/flowframework/transport/WorkflowRequest.java

@@ -32,15 +32,30 @@ public class WorkflowRequest extends ActionRequest {
      */
     @Nullable
     private Template template;
+    /**
+     * Validation flag
+     */
+    private boolean dryRun;
 
     /**
-     * Instantiates a new WorkflowRequest
+     * Instantiates a new WorkflowRequest and defaults dry run to false
      * @param workflowId the documentId of the workflow
      * @param template the use case template which describes the workflow
      */
     public WorkflowRequest(@Nullable String workflowId, @Nullable Template template) {
+        this(workflowId, template, false);
+    }
+
+    /**
+     * 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
+     */
+    public WorkflowRequest(@Nullable String workflowId, @Nullable Template template, boolean dryRun) {
         this.workflowId = workflowId;
         this.template = template;
+        this.dryRun = dryRun;
     }
 
     /**
@@ -53,6 +68,7 @@ public WorkflowRequest(StreamInput in) throws IOException {
         this.workflowId = in.readOptionalString();
         String templateJson = in.readOptionalString();
         this.template = templateJson == null ? null : Template.parse(templateJson);
+        this.dryRun = in.readBoolean();
     }
 
     /**
@@ -73,11 +89,20 @@ public Template getTemplate() {
         return this.template;
     }
 
+    /**
+     * Gets the dry run validation flag
+     * @return the dry run boolean
+     */
+    public boolean isDryRun() {
+        return this.dryRun;
+    }
+
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         out.writeOptionalString(workflowId);
         out.writeOptionalString(template == null ? null : template.toJson());
+        out.writeBoolean(dryRun);
     }
 
     @Override

src/test/java/org/opensearch/flowframework/transport/CreateWorkflowTransportActionTests.java

@@ -22,6 +22,7 @@
 import org.opensearch.flowframework.model.WorkflowEdge;
 import org.opensearch.flowframework.model.WorkflowNode;
 import org.opensearch.flowframework.util.ParseUtils;
+import org.opensearch.flowframework.workflow.WorkflowProcessSorter;
 import org.opensearch.tasks.Task;
 import org.opensearch.test.OpenSearchTestCase;
 import org.opensearch.threadpool.ThreadPool;
@@ -56,6 +57,7 @@ public void setUp() throws Exception {
         this.createWorkflowTransportAction = new CreateWorkflowTransportAction(
             mock(TransportService.class),
             mock(ActionFilters.class),
+            mock(WorkflowProcessSorter.class),
             flowFrameworkIndicesHandler,
             client
         );