Add setting to limit max workflow steps

Signed-off-by: Daniel Widdis <widdis@gmail.com>

.github/workflows/publish-snapshots.yml

@@ -20,7 +20,7 @@ jobs:
       contents: write
 
     steps:
-      - uses: actions/setup-java@v4
+      - uses: actions/setup-java@v3
         with:
           distribution: temurin # Temurin is a distribution of adoptium
           java-version: 17

DEVELOPER_GUIDE.md

@@ -36,7 +36,7 @@ This package uses the [Gradle](https://docs.gradle.org/current/userguide/usergui
 #### Building from the command line
 
 1. `./gradlew check` builds and tests.
-2. `./gradlew :run` installs and runs ML-Commons and Flow Framework Plugins into a local cluster
+2. `./gradlew :run` runs the plugin.
 3. `./gradlew spotlessApply` formats code. And/or import formatting rules in [formatterConfig.xml](formatter/formatterConfig.xml) with IDE.
 4. `./gradlew test` to run the complete test suite.
 

gradle/wrapper/gradle-wrapper.properties

@@ -1,8 +1,7 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-8.5-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.4-bin.zip
 networkTimeout=10000
 validateDistributionUrl=true
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
-distributionSha256Sum=9d926787066a081739e8200858338b4a69e837c3a821a33aca9db09dd4a41026

src/main/java/org/opensearch/flowframework/FlowFrameworkPlugin.java

@@ -63,6 +63,7 @@
 import static org.opensearch.flowframework.common.FlowFrameworkSettings.FLOW_FRAMEWORK_ENABLED;
 import static org.opensearch.flowframework.common.FlowFrameworkSettings.MAX_GET_TASK_REQUEST_RETRY;
 import static org.opensearch.flowframework.common.FlowFrameworkSettings.MAX_WORKFLOWS;
+import static org.opensearch.flowframework.common.FlowFrameworkSettings.MAX_WORKFLOW_STEPS;
 import static org.opensearch.flowframework.common.FlowFrameworkSettings.WORKFLOW_REQUEST_TIMEOUT;
 
 /**
@@ -106,7 +107,7 @@ public Collection<Object> createComponents(
             mlClient,
             flowFrameworkIndicesHandler
         );
-        WorkflowProcessSorter workflowProcessSorter = new WorkflowProcessSorter(workflowStepFactory, threadPool);
+        WorkflowProcessSorter workflowProcessSorter = new WorkflowProcessSorter(workflowStepFactory, threadPool, clusterService, settings);
 
         return ImmutableList.of(workflowStepFactory, workflowProcessSorter, encryptorUtils, flowFrameworkIndicesHandler);
     }
@@ -144,6 +145,7 @@ public List<Setting<?>> getSettings() {
         List<Setting<?>> settings = ImmutableList.of(
             FLOW_FRAMEWORK_ENABLED,
             MAX_WORKFLOWS,
+            MAX_WORKFLOW_STEPS,
             WORKFLOW_REQUEST_TIMEOUT,
             MAX_GET_TASK_REQUEST_RETRY
         );

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

@@ -18,6 +18,8 @@ private FlowFrameworkSettings() {}
 
     /** The upper limit of max workflows that can be created  */
     public static final int MAX_WORKFLOWS_LIMIT = 10000;
+    /** The upper limit of max workflow steps that can be in a single workflow  */
+    public static final int MAX_WORKFLOW_STEPS_LIMIT = 500;
 
     /** This setting sets max workflows that can be created  */
     public static final Setting<Integer> MAX_WORKFLOWS = Setting.intSetting(
@@ -29,6 +31,16 @@ private FlowFrameworkSettings() {}
         Setting.Property.Dynamic
     );
 
+    /** This setting sets max workflows that can be created  */
+    public static final Setting<Integer> MAX_WORKFLOW_STEPS = Setting.intSetting(
+        "plugins.flow_framework.max_workflow_steps",
+        50,
+        1,
+        MAX_WORKFLOW_STEPS_LIMIT,
+        Setting.Property.NodeScope,
+        Setting.Property.Dynamic
+    );
+
     /** This setting sets the timeout for the request  */
     public static final Setting<TimeValue> WORKFLOW_REQUEST_TIMEOUT = Setting.positiveTimeSetting(
         "plugins.flow_framework.request_timeout",

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

@@ -10,6 +10,8 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.opensearch.cluster.service.ClusterService;
+import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.core.rest.RestStatus;
 import org.opensearch.flowframework.exception.FlowFrameworkException;
@@ -32,6 +34,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import static org.opensearch.flowframework.common.FlowFrameworkSettings.MAX_WORKFLOW_STEPS;
 import static org.opensearch.flowframework.model.WorkflowNode.NODE_TIMEOUT_DEFAULT_VALUE;
 import static org.opensearch.flowframework.model.WorkflowNode.NODE_TIMEOUT_FIELD;
 import static org.opensearch.flowframework.model.WorkflowNode.USER_INPUTS_FIELD;
@@ -45,16 +48,26 @@ public class WorkflowProcessSorter {
 
     private WorkflowStepFactory workflowStepFactory;
     private ThreadPool threadPool;
+    private Integer maxWorkflowSteps;
 
     /**
      * Instantiate this class.
      *
      * @param workflowStepFactory The factory which matches template step types to instances.
      * @param threadPool The OpenSearch Thread pool to pass to process nodes.
+     * @param clusterService The OpenSearch cluster service.
+     * @param settings OpenSerch settings
      */
-    public WorkflowProcessSorter(WorkflowStepFactory workflowStepFactory, ThreadPool threadPool) {
+    public WorkflowProcessSorter(
+        WorkflowStepFactory workflowStepFactory,
+        ThreadPool threadPool,
+        ClusterService clusterService,
+        Settings settings
+    ) {
         this.workflowStepFactory = workflowStepFactory;
         this.threadPool = threadPool;
+        this.maxWorkflowSteps = MAX_WORKFLOW_STEPS.get(settings);
+        clusterService.getClusterSettings().addSettingsUpdateConsumer(MAX_WORKFLOW_STEPS, it -> maxWorkflowSteps = it);
     }
 
     /**
@@ -64,6 +77,20 @@ public WorkflowProcessSorter(WorkflowStepFactory workflowStepFactory, ThreadPool
      * @return A list of Process Nodes sorted topologically.  All predecessors of any node will occur prior to it in the list.
      */
     public List<ProcessNode> sortProcessNodes(Workflow workflow, String workflowId) {
+        if (workflow.nodes().size() > this.maxWorkflowSteps) {
+            throw new FlowFrameworkException(
+                "Workflow "
+                    + workflowId
+                    + " has "
+                    + workflow.nodes().size()
+                    + " nodes, which exceeds the maximum of "
+                    + this.maxWorkflowSteps
+                    + ". Change the setting ["
+                    + MAX_WORKFLOW_STEPS.getKey()
+                    + "] to increase this.",
+                RestStatus.BAD_REQUEST
+            );
+        }
         List<WorkflowNode> sortedNodes = topologicalSort(workflow.nodes(), workflow.edges());
 
         List<ProcessNode> nodes = new ArrayList<>();

src/test/java/org/opensearch/flowframework/FlowFrameworkPluginTests.java

@@ -29,6 +29,7 @@
 import static org.opensearch.flowframework.common.FlowFrameworkSettings.FLOW_FRAMEWORK_ENABLED;
 import static org.opensearch.flowframework.common.FlowFrameworkSettings.MAX_GET_TASK_REQUEST_RETRY;
 import static org.opensearch.flowframework.common.FlowFrameworkSettings.MAX_WORKFLOWS;
+import static org.opensearch.flowframework.common.FlowFrameworkSettings.MAX_WORKFLOW_STEPS;
 import static org.opensearch.flowframework.common.FlowFrameworkSettings.WORKFLOW_REQUEST_TIMEOUT;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -62,7 +63,7 @@ public void setUp() throws Exception {
 
         final Set<Setting<?>> settingsSet = Stream.concat(
             ClusterSettings.BUILT_IN_CLUSTER_SETTINGS.stream(),
-            Stream.of(FLOW_FRAMEWORK_ENABLED, MAX_WORKFLOWS, WORKFLOW_REQUEST_TIMEOUT, MAX_GET_TASK_REQUEST_RETRY)
+            Stream.of(FLOW_FRAMEWORK_ENABLED, MAX_WORKFLOWS, MAX_WORKFLOW_STEPS, WORKFLOW_REQUEST_TIMEOUT, MAX_GET_TASK_REQUEST_RETRY)
         ).collect(Collectors.toSet());
         clusterSettings = new ClusterSettings(settings, settingsSet);
         clusterService = mock(ClusterService.class);
@@ -84,7 +85,7 @@ public void testPlugin() throws IOException {
             assertEquals(4, ffp.getRestHandlers(settings, null, null, null, null, null, null).size());
             assertEquals(4, ffp.getActions().size());
             assertEquals(1, ffp.getExecutorBuilders(settings).size());
-            assertEquals(4, ffp.getSettings().size());
+            assertEquals(5, ffp.getSettings().size());
         }
     }
 }

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

@@ -13,6 +13,9 @@
 import org.opensearch.action.support.ActionFilters;
 import org.opensearch.action.update.UpdateResponse;
 import org.opensearch.client.Client;
+import org.opensearch.cluster.service.ClusterService;
+import org.opensearch.common.settings.ClusterSettings;
+import org.opensearch.common.settings.Setting;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.common.util.concurrent.ThreadContext;
@@ -34,18 +37,24 @@
 
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import org.mockito.ArgumentCaptor;
 
 import static org.opensearch.action.DocWriteResponse.Result.UPDATED;
 import static org.opensearch.flowframework.common.CommonValue.GLOBAL_CONTEXT_INDEX;
 import static org.opensearch.flowframework.common.CommonValue.WORKFLOW_STATE_INDEX;
+import static org.opensearch.flowframework.common.FlowFrameworkSettings.FLOW_FRAMEWORK_ENABLED;
+import static org.opensearch.flowframework.common.FlowFrameworkSettings.MAX_GET_TASK_REQUEST_RETRY;
 import static org.opensearch.flowframework.common.FlowFrameworkSettings.MAX_WORKFLOWS;
+import static org.opensearch.flowframework.common.FlowFrameworkSettings.MAX_WORKFLOW_STEPS;
 import static org.opensearch.flowframework.common.FlowFrameworkSettings.WORKFLOW_REQUEST_TIMEOUT;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.anyString;
-import static org.mockito.Mockito.anyInt;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
@@ -61,6 +70,8 @@ public class CreateWorkflowTransportActionTests extends OpenSearchTestCase {
     private Template template;
     private Client client = mock(Client.class);
     private ThreadPool threadPool;
+    private ClusterSettings clusterSettings;
+    private ClusterService clusterService;
     private ParseUtils parseUtils;
     private ThreadContext threadContext;
     private Settings settings;
@@ -73,8 +84,15 @@ public void setUp() throws Exception {
             .put("plugins.flow_framework.max_workflows.", 2)
             .put("plugins.flow_framework.request_timeout", TimeValue.timeValueSeconds(10))
             .build();
+        final Set<Setting<?>> settingsSet = Stream.concat(
+            ClusterSettings.BUILT_IN_CLUSTER_SETTINGS.stream(),
+            Stream.of(FLOW_FRAMEWORK_ENABLED, MAX_WORKFLOWS, MAX_WORKFLOW_STEPS, WORKFLOW_REQUEST_TIMEOUT, MAX_GET_TASK_REQUEST_RETRY)
+        ).collect(Collectors.toSet());
+        clusterSettings = new ClusterSettings(settings, settingsSet);
+        clusterService = mock(ClusterService.class);
+        when(clusterService.getClusterSettings()).thenReturn(clusterSettings);
         this.flowFrameworkIndicesHandler = mock(FlowFrameworkIndicesHandler.class);
-        this.workflowProcessSorter = new WorkflowProcessSorter(mock(WorkflowStepFactory.class), threadPool);
+        this.workflowProcessSorter = new WorkflowProcessSorter(mock(WorkflowStepFactory.class), threadPool, clusterService, settings);
         this.createWorkflowTransportAction = spy(
             new CreateWorkflowTransportAction(
                 mock(TransportService.class),