feat(ui): rework topology to handle subflows

closes #774
kestra-io · Sep 8, 2023 · 11ba4cb · 11ba4cb
1 parent a2c81b2
commit 11ba4cb
Show file tree

Hide file tree

Showing 39 changed files with 1,037 additions and 763 deletions.
diff --git a/cli/src/main/java/io/kestra/cli/commands/flows/FlowDotCommand.java b/cli/src/main/java/io/kestra/cli/commands/flows/FlowDotCommand.java
@@ -5,7 +5,7 @@
 import io.kestra.core.models.hierarchies.GraphCluster;
 import io.kestra.core.serializers.YamlFlowParser;
 import io.kestra.core.services.Graph2DotService;
-import io.kestra.core.services.GraphService;
+import io.kestra.core.utils.GraphUtils;
 import io.micronaut.context.ApplicationContext;
 import jakarta.inject.Inject;
 import lombok.extern.slf4j.Slf4j;
@@ -32,7 +32,7 @@ public Integer call() throws Exception {
         YamlFlowParser parser = applicationContext.getBean(YamlFlowParser.class);
         Flow flow = parser.parse(file.toFile(), Flow.class);
 
-        GraphCluster graph = GraphService.of(flow, null);
+        GraphCluster graph = GraphUtils.of(flow, null);
 
         stdOut(Graph2DotService.dot(graph.getGraph()));
 

diff --git a/core/src/main/java/io/kestra/core/models/hierarchies/AbstractGraph.java b/core/src/main/java/io/kestra/core/models/hierarchies/AbstractGraph.java
@@ -3,7 +3,6 @@
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
-import io.kestra.core.utils.IdUtils;
 import io.micronaut.core.annotation.Introspected;
 import lombok.Getter;
 import lombok.Setter;
@@ -17,9 +16,10 @@ public abstract class AbstractGraph {
     protected String uid;
     @JsonInclude
     protected String type;
+    @Setter
+    protected boolean error;
 
     public AbstractGraph() {
-        this.uid = IdUtils.create();
         this.type = this.getClass().getName();
     }
 
@@ -32,4 +32,8 @@ public AbstractGraph(String uid) {
     public String getLabel() {
         return this.getUid();
     }
+
+    public void setUid(String uid) {
+        this.uid = uid;
+    }
 }
diff --git a/core/src/main/java/io/kestra/core/models/hierarchies/AbstractGraphTask.java b/core/src/main/java/io/kestra/core/models/hierarchies/AbstractGraphTask.java
@@ -8,6 +8,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Objects;
 
 @ToString
 @Getter
@@ -18,24 +19,19 @@ public abstract class AbstractGraphTask extends AbstractGraph {
     private final List<String> values;
     private final RelationType relationType;
 
-    public AbstractGraphTask() {
-        super();
-
-        this.task = null;
-        this.taskRun = null;
-        this.values = null;
-        this.relationType = null;
-    }
-
-    public AbstractGraphTask(Task task, TaskRun taskRun, List<String> values, RelationType relationType) {
-        super();
+    public AbstractGraphTask(String uid, Task task, TaskRun taskRun, List<String> values, RelationType relationType) {
+        super(uid);
 
         this.task = task;
         this.taskRun = taskRun;
         this.values = values;
         this.relationType = relationType;
     }
 
+    public AbstractGraphTask(Task task, TaskRun taskRun, List<String> values, RelationType relationType) {
+        this(task.getId(), task, taskRun, values, relationType);
+    }
+
     @Override
     public String getLabel() {
         return this.getUid() + (this.getTaskRun() != null ? " > " + this.getTaskRun().getValue() + " (" + this.getTaskRun().getId() + ")" : "");
@@ -45,9 +41,7 @@ public String getLabel() {
     public String getUid() {
         List<String> list = new ArrayList<>();
 
-        if (this.task != null) {
-            list.add(this.task.getId());
-        }
+        list.add(this.uid);
 
         if (values != null) {
             list.addAll(values);
@@ -57,18 +51,11 @@ public String getUid() {
     }
 
     @Override
-    public boolean equals(final Object o) {
-        if (o == this) return true;
-
-        if (!(o instanceof AbstractGraphTask)) {
-            return false;
-        }
+    public boolean equals(Object object) {
+        if (this == object) return true;
+        if (object == null || getClass() != object.getClass()) return false;
+        AbstractGraphTask that = (AbstractGraphTask) object;
 
-        return o.hashCode() == this.hashCode();
-    }
-
-    @Override
-    public int hashCode() {
-        return (this.uid + this.getClass().getName()).hashCode();
+        return Objects.equals(this.getUid(), that.getUid());
     }
 }
diff --git a/core/src/main/java/io/kestra/core/models/hierarchies/AbstractGraphTrigger.java b/core/src/main/java/io/kestra/core/models/hierarchies/AbstractGraphTrigger.java
@@ -20,7 +20,7 @@ public AbstractGraphTrigger(AbstractTrigger trigger) {
     @Override
     public String getUid() {
         if (this.trigger != null) {
-            return this.trigger.getId() + "_trigger";
+            return this.trigger.getId();
         }
 
         return this.uid;

diff --git a/core/src/main/java/io/kestra/core/models/hierarchies/FlowGraph.java b/core/src/main/java/io/kestra/core/models/hierarchies/FlowGraph.java
@@ -1,9 +1,7 @@
 package io.kestra.core.models.hierarchies;
 
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
-import io.kestra.core.models.executions.Execution;
-import io.kestra.core.models.flows.Flow;
-import io.kestra.core.services.GraphService;
+import io.kestra.core.utils.GraphUtils;
 import lombok.*;
 
 import java.util.ArrayList;
@@ -18,18 +16,11 @@ public class FlowGraph {
     List<Cluster> clusters;
     List<String> flowables;
 
-    public static FlowGraph of(Flow flow) throws IllegalVariableEvaluationException {
-        return FlowGraph.of(flow, null);
-    }
-
-    public static FlowGraph of(Flow flow, Execution execution) throws IllegalVariableEvaluationException {
-        GraphCluster graph = GraphService.of(flow, execution);
-
+    public static FlowGraph of(GraphCluster graph) throws IllegalVariableEvaluationException {
         return FlowGraph.builder()
-            .flowables(GraphService.flowables(flow))
-            .nodes(GraphService.nodes(graph))
-            .edges(GraphService.edges(graph))
-            .clusters(GraphService.clusters(graph, new ArrayList<>())
+            .nodes(GraphUtils.nodes(graph))
+            .edges(GraphUtils.edges(graph))
+            .clusters(GraphUtils.clusters(graph, new ArrayList<>())
                 .stream()
                 .map(g -> new Cluster(g.getKey(), g.getKey().getGraph()
                     .nodes()

diff --git a/core/src/main/java/io/kestra/core/models/hierarchies/Graph.java b/core/src/main/java/io/kestra/core/models/hierarchies/Graph.java
@@ -4,6 +4,7 @@
 import com.google.common.graph.ValueGraphBuilder;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
+import lombok.Setter;
 
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -27,16 +28,6 @@ public Graph<T, V> addNode(T node) {
     }
 
     public Graph<T, V> addEdge(T previous, T next, V value) {
-        /*
-        if (!this.graph.nodes().contains(previous)) {
-            throw new IllegalArgumentException("Unable to find previous node '" + previous + "'@" + previous.hashCode());
-        }
-
-        if (!this.graph.nodes().contains(next)) {
-            throw new IllegalArgumentException("Unable to find next node '" + next + "'" + next.hashCode());
-        }
-        */
-
         this.graph.putEdgeValue(previous, next, value);
 
         return this;
@@ -62,7 +53,12 @@ public Set<Edge<T, V>> edges() {
             .collect(Collectors.toSet());
     }
 
+    public void removeNode(T node) {
+        this.graph.removeNode(node);
+    }
+
     @Getter
+    @Setter
     @AllArgsConstructor
     public static class Edge<T, V> {
         T source;

diff --git a/core/src/main/java/io/kestra/core/models/hierarchies/GraphCluster.java b/core/src/main/java/io/kestra/core/models/hierarchies/GraphCluster.java
@@ -6,6 +6,7 @@
 import lombok.Getter;
 
 import java.util.List;
+import java.util.Optional;
 
 
 @Getter
@@ -16,20 +17,15 @@ public class GraphCluster extends AbstractGraph {
     protected RelationType relationType;
 
     @JsonIgnore
-    private final GraphClusterRoot root;
+    protected final GraphClusterRoot root;
 
     @JsonIgnore
-    private final GraphClusterEnd end;
+    protected final GraphClusterEnd end;
 
-    public GraphCluster() {
-        super();
+    protected final AbstractGraphTask taskNode;
 
-        this.relationType = null;
-        this.root = new GraphClusterRoot();
-        this.end = new GraphClusterEnd();
-
-        graph.addNode(this.root);
-        graph.addNode(this.end);
+    public GraphCluster() {
+        this("root");
     }
 
 
@@ -39,25 +35,70 @@ public GraphCluster(String uid) {
         this.relationType = null;
         this.root = new GraphClusterRoot();
         this.end = new GraphClusterEnd();
+        this.taskNode = null;
 
-        graph.addNode(this.root);
-        graph.addNode(this.end);
+        this.addNode(this.root);
+        this.addNode(this.end);
     }
 
     public GraphCluster(Task task, TaskRun taskRun, List<String> values, RelationType relationType) {
-        super();
+        this(new GraphTask(task.getId(), task, taskRun, values, relationType), task.getId(), relationType);
 
-        this.uid = "cluster_" + task.getId();
-        this.relationType = relationType;
+        this.addNode(this.taskNode, false);
+        this.addEdge(this.getRoot(), this.taskNode, new Relation());
+    }
+
+    protected GraphCluster(AbstractGraphTask taskNode, String uid, RelationType relationType) {
+        super(uid);
 
+        this.relationType = relationType;
         this.root = new GraphClusterRoot();
         this.end = new GraphClusterEnd();
+        this.taskNode = taskNode;
+
+        this.addNode(this.root);
+        this.addNode(this.end);
+    }
+
+    public void addNode(AbstractGraph node) {
+        this.addNode(node, true);
+    }
+
+    public void addNode(AbstractGraph node, boolean withClusterUidPrefix) {
+        if (withClusterUidPrefix) {
+            node.setUid(prefixedUid(node.uid));
+        }
+        this.getGraph().addNode(node);
+    }
+
+    public void addEdge(AbstractGraph source, AbstractGraph target, Relation relation) {
+        this.getGraph().addEdge(source, target, relation);
+    }
+
+    private String prefixedUid(String uid) {
+        return Optional.ofNullable(this.uid).map(u -> u + "." + uid).orElse(uid);
+    }
+
+    @Override
+    public String getUid() {
+        return "cluster_" + super.getUid();
+    }
+
+    @Override
+    public void setUid(String uid) {
+        graph.nodes().stream().filter(node -> (!(node instanceof GraphClusterRoot) && !(node instanceof GraphClusterEnd))
+                || node.equals(this.root) || node.equals(this.end))
+            .forEach(node -> node.setUid(uid + node.uid.substring(this.uid.length())));
+
+        super.setUid(uid);
+    }
 
-        graph.addNode(this.root);
-        graph.addNode(this.end);
+    @Override
+    public void setError(boolean error) {
+        this.error = error;
 
-        GraphTask flowableGraphTask = new GraphTask(task, taskRun, values, relationType);
-        this.getGraph().addNode(flowableGraphTask);
-        this.getGraph().addEdge(this.getRoot(), flowableGraphTask, new Relation());
+        this.taskNode.error = error;
+        this.root.error = error;
+        this.end.error = error;
     }
-}
+}
diff --git a/core/src/main/java/io/kestra/core/models/hierarchies/GraphClusterEnd.java b/core/src/main/java/io/kestra/core/models/hierarchies/GraphClusterEnd.java
@@ -7,6 +7,6 @@
 @Getter
 public class GraphClusterEnd extends AbstractGraph {
     public GraphClusterEnd() {
-        super(IdUtils.create() + "_end");
+        super(IdUtils.create());
     }
 }
diff --git a/core/src/main/java/io/kestra/core/models/hierarchies/GraphClusterRoot.java b/core/src/main/java/io/kestra/core/models/hierarchies/GraphClusterRoot.java
@@ -6,6 +6,6 @@
 @Getter
 public class GraphClusterRoot extends AbstractGraph {
     public GraphClusterRoot() {
-        super(IdUtils.create() + "_root");
+        super(IdUtils.create());
     }
 }
diff --git a/core/src/main/java/io/kestra/core/models/hierarchies/GraphTask.java b/core/src/main/java/io/kestra/core/models/hierarchies/GraphTask.java
@@ -1,15 +1,15 @@
 package io.kestra.core.models.hierarchies;
 
-import lombok.Getter;
 import io.kestra.core.models.executions.TaskRun;
 import io.kestra.core.models.tasks.Task;
+import lombok.Getter;
 
 import java.util.List;
 
 @Getter
 public class GraphTask extends AbstractGraphTask {
-    public GraphTask() {
-        super();
+    public GraphTask(String uid, Task task, TaskRun taskRun, List<String> values, RelationType relationType) {
+        super(uid, task, taskRun, values, relationType);
     }
 
     public GraphTask(Task task, TaskRun taskRun, List<String> values, RelationType relationType) {

diff --git a/core/src/main/java/io/kestra/core/models/hierarchies/SubflowGraphCluster.java b/core/src/main/java/io/kestra/core/models/hierarchies/SubflowGraphCluster.java
@@ -0,0 +1,14 @@
+package io.kestra.core.models.hierarchies;
+
+import lombok.Getter;
+
+
+@Getter
+public class SubflowGraphCluster extends GraphCluster {
+    public SubflowGraphCluster(String uid, SubflowGraphTask subflowGraphTask) {
+        super(subflowGraphTask, uid, RelationType.SEQUENTIAL);
+
+        this.getGraph().addNode(subflowGraphTask);
+        this.addEdge(this.getRoot(), subflowGraphTask, new Relation());
+    }
+}
diff --git a/core/src/main/java/io/kestra/core/models/hierarchies/SubflowGraphTask.java b/core/src/main/java/io/kestra/core/models/hierarchies/SubflowGraphTask.java
@@ -0,0 +1,19 @@
+package io.kestra.core.models.hierarchies;
+
+import io.kestra.core.models.executions.TaskRun;
+import io.kestra.core.tasks.flows.Flow;
+import lombok.Getter;
+
+import java.util.List;
+
+@Getter
+public class SubflowGraphTask extends AbstractGraphTask {
+    public SubflowGraphTask(Flow task, TaskRun taskRun, List<String> values, RelationType relationType) {
+        super(task, taskRun, values, relationType);
+    }
+
+    @Override
+    public Flow getTask() {
+        return (Flow) super.getTask();
+    }
+}