diff --git a/orca-run-multiple-pipelines/src/main/java/io/armory/plugin/smp/config/UtilityHelper.java b/orca-run-multiple-pipelines/src/main/java/io/armory/plugin/smp/config/UtilityHelper.java index 3a07924..3e66d91 100644 --- a/orca-run-multiple-pipelines/src/main/java/io/armory/plugin/smp/config/UtilityHelper.java +++ b/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 getGraphOfApps(Map mapOfApps, List 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 getGraphOfApps(Map mapOfApps, List in } /** - *

Modifies orderOfExecutions using recursion - *

adds new levels searching through the graph looping on the successors of that particular node - *

before adding checks that his predecessors are already added and himself has not been added + *

Modifies orderOfExecutions, building the remaining levels of the DAG. + *

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). + *

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> orderOfExecutions, MutableGraph graph2, List allAppsInWholeQueue, int i) { - List> listListOfApps = new LinkedList<>(); - for(App app : orderOfExecutions.get(i)) { - if (!allAppsInWholeQueue.contains(app)) { - allAppsInWholeQueue.add(app); - } - } - List 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> orderOfExecutions, MutableGraph graph, Set allAppsInWholeQueue) { + int i = 0; + while (i < orderOfExecutions.size()) { + List currentLevel = orderOfExecutions.get(i); + allAppsInWholeQueue.addAll(currentLevel); + + Set 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 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(", "))); } } diff --git a/orca-run-multiple-pipelines/src/main/java/io/armory/plugin/smp/tasks/ParsePipelinesYamlTask.java b/orca-run-multiple-pipelines/src/main/java/io/armory/plugin/smp/tasks/ParsePipelinesYamlTask.java index f54ede9..fe5a6b2 100644 --- a/orca-run-multiple-pipelines/src/main/java/io/armory/plugin/smp/tasks/ParsePipelinesYamlTask.java +++ b/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> 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); diff --git a/orca-run-multiple-pipelines/src/main/java/io/armory/plugin/smp/tasks/RunMultiplePipelinesTask.java b/orca-run-multiple-pipelines/src/main/java/io/armory/plugin/smp/tasks/RunMultiplePipelinesTask.java index d0b0ed4..5fe7f56 100644 --- a/orca-run-multiple-pipelines/src/main/java/io/armory/plugin/smp/tasks/RunMultiplePipelinesTask.java +++ b/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 pipelineExecutionsIds = Collections.synchronizedList(new ArrayList<>()); ExecutionStatus returnExecutionStatus = ExecutionStatus.SUCCEEDED; - //create a linked list of Threads to join on main thread - List 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> 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())); diff --git a/orca-run-multiple-pipelines/src/test/java/io/armory/plugin/smp/config/UtilityHelperTest.java b/orca-run-multiple-pipelines/src/test/java/io/armory/plugin/smp/config/UtilityHelperTest.java index 653a858..adbaab1 100644 --- a/orca-run-multiple-pipelines/src/test/java/io/armory/plugin/smp/config/UtilityHelperTest.java +++ b/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(); + + 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(); + + 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(); + var graph = utilityHelper.getGraphOfApps(testAppsByYamlId, initialExecutions); + + List> orderOfExecutions = new ArrayList<>(); + orderOfExecutions.add(initialExecutions); // empty: both apps depend on each other + + assertThrows(IllegalStateException.class, + () -> utilityHelper.addLevels(orderOfExecutions, graph, new HashSet<>())); + } }