[Backport 2.x] Set up Rest Integration Test framework

build.gradle

@@ -204,6 +204,13 @@ integTest {
     systemProperty "user", System.getProperty("user")
     systemProperty "password", System.getProperty("password")
 
+    // Only rest case can run with remote cluster
+    if (System.getProperty("tests.rest.cluster") != null) {
+        filter {
+            includeTestsMatching "org.opensearch.flowframework.rest.*IT"
+        }
+    }
+
 
     // doFirst delays this block until execution time
     doFirst {
@@ -263,6 +270,25 @@ testClusters.integTest {
     }
 }
 
+// Remote Integration Tests
+task integTestRemote(type: RestIntegTestTask) {
+    testClassesDirs = sourceSets.test.output.classesDirs
+    classpath = sourceSets.test.runtimeClasspath
+
+    systemProperty "https", System.getProperty("https")
+    systemProperty "user", System.getProperty("user")
+    systemProperty "password", System.getProperty("password")
+    systemProperty 'cluster.number_of_nodes', "${_numNodes}"
+    systemProperty 'tests.security.manager', 'false'
+
+    // Run tests with remote cluster only if rest case is defined
+    if (System.getProperty("tests.rest.cluster") != null) {
+        filter {
+            includeTestsMatching "org.opensearch.flowframework.rest.*IT"
+        }
+    }
+}
+
 // Automatically sets up the integration test cluster locally
 run {
     useCluster testClusters.integTest

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

@@ -77,8 +77,10 @@ protected void doExecute(Task task, GetWorkflowStateRequest request, ActionListe
                         WorkflowState workflowState = WorkflowState.parse(parser);
                         listener.onResponse(new GetWorkflowStateResponse(workflowState, request.getAll()));
                     } catch (Exception e) {
-                        logger.error("Failed to parse workflowState" + r.getId(), e);
-                        listener.onFailure(new FlowFrameworkException("Failed to parse workflowState" + r.getId(), RestStatus.BAD_REQUEST));
+                        logger.error("Failed to parse workflowState: " + r.getId(), e);
+                        listener.onFailure(
+                            new FlowFrameworkException("Failed to parse workflowState: " + r.getId(), RestStatus.BAD_REQUEST)
+                        );
                     }
                 } else {
                     listener.onFailure(new FlowFrameworkException("Fail to find workflow", RestStatus.NOT_FOUND));

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

@@ -138,11 +138,10 @@ protected void doExecute(Task task, WorkflowRequest request, ActionListener<Work
                     ),
                     ActionListener.wrap(updateResponse -> {
                         logger.info("updated workflow {} state to PROVISIONING", request.getWorkflowId());
+                        listener.onResponse(new WorkflowResponse(workflowId));
                     }, exception -> { logger.error("Failed to update workflow state : {}", exception.getMessage()); })
                 );
 
-                // Respond to rest action then execute provisioning workflow async
-                listener.onResponse(new WorkflowResponse(workflowId));
                 executeWorkflowAsync(workflowId, provisionProcessSequence, listener);
 
             }, exception -> {

src/main/java/org/opensearch/flowframework/workflow/RegisterLocalModelStep.java

@@ -114,14 +114,13 @@ public void onFailure(Exception e) {
             NAME_FIELD,
             VERSION_FIELD,
             MODEL_FORMAT,
-            MODEL_GROUP_ID,
             MODEL_TYPE,
             EMBEDDING_DIMENSION,
             FRAMEWORK_TYPE,
             MODEL_CONTENT_HASH_VALUE,
             URL
         );
-        Set<String> optionalKeys = Set.of(DESCRIPTION_FIELD, ALL_CONFIG);
+        Set<String> optionalKeys = Set.of(DESCRIPTION_FIELD, MODEL_GROUP_ID, ALL_CONFIG);
 
         try {
             Map<String, Object> inputs = ParseUtils.getInputsFromPreviousSteps(

src/main/java/org/opensearch/flowframework/workflow/WorkflowProcessSorter.java

@@ -236,7 +236,10 @@ public void validateGraph(List<ProcessNode> processNodes, WorkflowValidator vali
             if (!allInputs.containsAll(expectedInputs)) {
                 expectedInputs.removeAll(allInputs);
                 throw new FlowFrameworkException(
-                    "Invalid graph, missing the following required inputs : " + expectedInputs.toString(),
+                    "Invalid workflow, node ["
+                        + processNode.id()
+                        + "] missing the following required inputs : "
+                        + expectedInputs.toString(),
                     RestStatus.BAD_REQUEST
                 );
             }

src/main/resources/mappings/workflow-steps.json

@@ -61,7 +61,6 @@
             "name",
             "version",
             "model_format",
-            "model_group_id",
             "model_content_hash_value",
             "model_type",
             "embedding_dimension",