Improve level scheduling: HashSet-based addLevels, cycle detection, ExecutorService for triggers

orca-run-multiple-pipelines/src/main/java/io/armory/plugin/smp/config/UtilityHelper.java

@@ -24,9 +24,13 @@
 import io.armory.plugin.smp.parseyml.BundleWeb;
 import org.apache.commons.lang3.ObjectUtils;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
 
 public class UtilityHelper {
 
@@ -47,7 +51,14 @@ public MutableGraph<App> getGraphOfApps(Map<String, App> mapOfApps, List<App> in
         mapOfApps.forEach((k, v) -> {
             if (ObjectUtils.isNotEmpty(v.getDependsOn())) {
                 v.getDependsOn().forEach(appName -> {
-                    graph.putEdge(mapOfApps.get(appName), mapOfApps.get(k));
+                    App dependency = mapOfApps.get(appName);
+                    if (dependency == null) {
+                        // previously this threw a bare NullPointerException from graph.putEdge(null, ...)
+                        throw new IllegalArgumentException(
+                                "App '" + k + "' declares depends_on '" + appName
+                                        + "' but no app with that identifier exists in the yaml config");
+                    }
+                    graph.putEdge(dependency, mapOfApps.get(k));
                 });
             } else {
                 //add nodes of not connected components
@@ -59,35 +70,43 @@ public MutableGraph<App> getGraphOfApps(Map<String, App> mapOfApps, List<App> in
     }
 
     /**
-     * <p> Modifies orderOfExecutions using recursion
-     * <p> adds new levels searching through the graph looping on the successors of that particular node
-     * <p> before adding checks that his predecessors are already added and himself has not been added
+     * <p> Modifies orderOfExecutions, building the remaining levels of the DAG.
+     * <p> A node becomes eligible for the next level once every one of its predecessors
+     * has already been scheduled in an earlier level (Kahn's-algorithm style).
+     * <p> Iterative rather than recursive, so a long dependency chain can't cause a
+     * StackOverflowError, and membership is tracked with a Set instead of scanning a
+     * List on every check.
+     *
+     * @throws IllegalStateException if a circular depends_on prevents some apps from
+     *                                ever becoming schedulable
      */
-    public void addLevels(List<List<App>> orderOfExecutions, MutableGraph<App> graph2, List<App> allAppsInWholeQueue, int i) {
-        List<List<App>> listListOfApps = new LinkedList<>();
-        for(App app : orderOfExecutions.get(i)) {
-            if (!allAppsInWholeQueue.contains(app)) {
-                allAppsInWholeQueue.add(app);
-            }
-        }
-        List<App> newLevel = new LinkedList<>();
-        for(App appI : orderOfExecutions.get(i)) {
-            graph2.successors(appI).forEach(app -> {
-                if ( allAppsInWholeQueue.containsAll(graph2.predecessors(app)) ) {
-                    if ( !allAppsInWholeQueue.contains(app) ) {
-                        if (!newLevel.contains(app)) {
-                            newLevel.add(app);
-                        }
+    public void addLevels(List<List<App>> orderOfExecutions, MutableGraph<App> graph, Set<App> allAppsInWholeQueue) {
+        int i = 0;
+        while (i < orderOfExecutions.size()) {
+            List<App> currentLevel = orderOfExecutions.get(i);
+            allAppsInWholeQueue.addAll(currentLevel);
+
+            Set<App> newLevel = new LinkedHashSet<>();
+            for (App appI : currentLevel) {
+                for (App app : graph.successors(appI)) {
+                    if (!allAppsInWholeQueue.contains(app)
+                            && allAppsInWholeQueue.containsAll(graph.predecessors(app))) {
+                        newLevel.add(app);
                     }
                 }
-            });
-        }
-        if (ObjectUtils.isNotEmpty(newLevel)) {
-            listListOfApps.add(newLevel);
+            }
+
+            if (!newLevel.isEmpty()) {
+                orderOfExecutions.add(new ArrayList<>(newLevel));
+            }
+            i++;
         }
-        if (ObjectUtils.isNotEmpty(listListOfApps)) {
-            orderOfExecutions.addAll(listListOfApps);
-            addLevels(orderOfExecutions, graph2, allAppsInWholeQueue, ++i);
+
+        if (allAppsInWholeQueue.size() < graph.nodes().size()) {
+            Set<App> stuck = new HashSet<>(graph.nodes());
+            stuck.removeAll(allAppsInWholeQueue);
+            throw new IllegalStateException("Circular depends_on detected, these apps never became schedulable: "
+                    + stuck.stream().map(App::getYamlIdentifier).collect(Collectors.joining(", ")));
         }
     }
 

orca-run-multiple-pipelines/src/main/java/io/armory/plugin/smp/tasks/ParsePipelinesYamlTask.java

@@ -18,6 +18,7 @@
 import org.springframework.stereotype.Component;
 
 import javax.annotation.Nonnull;
+import java.util.HashSet;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
@@ -59,7 +60,7 @@ public TaskResult execute(@Nonnull StageExecution stage) {
 
         List<List<App>> orderOfExecutions = new LinkedList<>();
         orderOfExecutions.add(initialExecutions);
-        utilityHelper.addLevels(orderOfExecutions, graph, new LinkedList<>(), 0);
+        utilityHelper.addLevels(orderOfExecutions, graph, new HashSet<>());
 
         logger.info("Map of apps, detected returning success " + mapOfApps.size() + " size");
         stage.getContext().put("orderOfExecutions", orderOfExecutions);

orca-run-multiple-pipelines/src/main/java/io/armory/plugin/smp/tasks/RunMultiplePipelinesTask.java

@@ -35,6 +35,10 @@
 
 import javax.annotation.Nonnull;
 import java.util.*;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 
 @ExposeToApp
 @Component
@@ -72,30 +76,39 @@ public TaskResult execute(@Nonnull StageExecution stage) {
         List<String> pipelineExecutionsIds = Collections.synchronizedList(new ArrayList<>());
         ExecutionStatus returnExecutionStatus = ExecutionStatus.SUCCEEDED;
 
-        //create a linked list of Threads to join on main thread
-        List<Thread> threadList = new LinkedList<>();
-
         //Add executionIdentifier parameter for UI
         for (App app : executionForThisLevel) {
             app.getArguments().put("executionIdentifier", app.getYamlIdentifier());
         }
 
-        //trigger pipelines in different Threads
-        for (App app : executionForThisLevel) {
-            Runnable triggerThis = () -> triggerThread(app, stage, pipelines, pipelineExecutions);
-            threadList.add(new Thread(triggerThis));
-        }
-
-        logger.info("Starting " + threadList.size() +  " Threads...");
-        threadList.forEach(Thread::start);
+        //trigger pipelines in different Threads, one per app in this level
+        ExecutorService executor = Executors.newFixedThreadPool(executionForThisLevel.size());
+        List<Future<?>> futures = new LinkedList<>();
+        logger.info("Starting " + executionForThisLevel.size() + " Threads...");
+        try {
+            for (App app : executionForThisLevel) {
+                futures.add(executor.submit(() -> triggerThread(app, stage, pipelines, pipelineExecutions)));
+            }
 
-        threadList.forEach(t -> {
-            try {
-                t.join();
-            } catch (InterruptedException e) {
-                e.printStackTrace();
+            for (Future<?> future : futures) {
+                try {
+                    // blocks like Thread#join(), but also surfaces any exception the
+                    // triggering thread threw instead of it silently vanishing to stderr
+                    future.get();
+                } catch (ExecutionException e) {
+                    logger.error("Error while triggering child pipeline", e.getCause());
+                    String message = e.getCause() != null ? e.getCause().getMessage() : e.getMessage();
+                    if (message != null) {
+                        stage.appendErrorMessage(message);
+                    }
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    logger.error("Interrupted while waiting for child pipeline triggers", e);
+                }
             }
-        });
+        } finally {
+            executor.shutdown();
+        }
         logger.info("Finished calling trigger");
 
         pipelineExecutions.forEach(p -> pipelineExecutionsIds.add(p.getId()));

orca-run-multiple-pipelines/src/test/java/io/armory/plugin/smp/config/UtilityHelperTest.java

@@ -9,9 +9,11 @@
 
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
 @ExtendWith(MockitoExtension.class)
 public class UtilityHelperTest {
@@ -95,7 +97,7 @@ public void shouldSetOrderOfExecutionsCorrectly() {
             add(testApp3);
         }});
 
-        utilityHelper.addLevels(orderOfExecutions, graph, new ArrayList<>(), 0);
+        utilityHelper.addLevels(orderOfExecutions, graph, new HashSet<>());
 
         assertEquals(new ArrayList<>() {{
             add(new ArrayList<>() {{
@@ -117,4 +119,43 @@ public void shouldSetOrderOfExecutionsCorrectly() {
             }});
         }}, orderOfExecutions);
     }
+
+    @Test
+    public void shouldThrowClearErrorWhenDependsOnReferencesUnknownApp() {
+        var testAppsByYamlId = new HashMap<String, App>();
+
+        var testApp1 = new App();
+        testApp1.setChildPipeline("testChildPipeline1");
+        testApp1.setDependsOn(List.of("appThatDoesNotExist"));
+
+        testAppsByYamlId.put("testApp1", testApp1);
+
+        assertThrows(IllegalArgumentException.class,
+                () -> utilityHelper.getGraphOfApps(testAppsByYamlId, new ArrayList<>()));
+    }
+
+    @Test
+    public void shouldThrowWhenCircularDependencyDetected() {
+        var testAppsByYamlId = new HashMap<String, App>();
+
+        var testApp1 = new App();
+        testApp1.setChildPipeline("testChildPipeline1");
+        testApp1.setDependsOn(List.of("testApp2"));
+
+        var testApp2 = new App();
+        testApp2.setChildPipeline("testChildPipeline2");
+        testApp2.setDependsOn(List.of("testApp1"));
+
+        testAppsByYamlId.put("testApp1", testApp1);
+        testAppsByYamlId.put("testApp2", testApp2);
+
+        var initialExecutions = new ArrayList<App>();
+        var graph = utilityHelper.getGraphOfApps(testAppsByYamlId, initialExecutions);
+
+        List<List<App>> orderOfExecutions = new ArrayList<>();
+        orderOfExecutions.add(initialExecutions); // empty: both apps depend on each other
+
+        assertThrows(IllegalStateException.class,
+                () -> utilityHelper.addLevels(orderOfExecutions, graph, new HashSet<>()));
+    }
 }