dag: validate the DAG when validating the pipeline spec 🚒

- Migrate the `DAG` struct and how it's build to the API
- Update `dag` package to take `v1alpha1.DAG` as argument for
  Schedulable, …
- Validate the DAG after `validateFrom`, and update `validateFrom`
  with no linear assumption

Signed-off-by: Vincent Demeester <[email protected]>

pkg/apis/pipeline/v1alpha1/dag.go

@@ -0,0 +1,135 @@
+/*
+Copyright 2018 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package v1alpha1
+
+import (
+	"fmt"
+	"strings"
+)
+
+// Node represents a Task in a pipeline.
+type Node struct {
+	// Task represent the PipelineTask in Pipeline
+	Task PipelineTask
+	// Prev represent all the Previous task Nodes for the current Task
+	Prev []*Node
+	// Next represent all the Next task Nodes for the current Task
+	Next []*Node
+}
+
+// DAG represents the Pipeline DAG
+type DAG struct {
+	//Nodes represent map of PipelineTask name to Node in Pipeline DAG
+	Nodes map[string]*Node
+}
+
+// Returns an empty Pipeline DAG
+func newDAG() *DAG {
+	return &DAG{Nodes: map[string]*Node{}}
+}
+
+func (g *DAG) addPipelineTask(t PipelineTask) (*Node, error) {
+	if _, ok := g.Nodes[t.Name]; ok {
+		return nil, fmt.Errorf("duplicate pipeline taks")
+	}
+	newNode := &Node{
+		Task: t,
+	}
+	g.Nodes[t.Name] = newNode
+	return newNode, nil
+}
+
+func linkPipelineTasks(prev *Node, next *Node) error {
+	// Check for self cycle
+	if prev.Task.Name == next.Task.Name {
+		return fmt.Errorf("cycle detected; task %q depends on itself", next.Task.Name)
+	}
+	// Check if we are adding cycles.
+	visited := map[string]bool{prev.Task.Name: true, next.Task.Name: true}
+	path := []string{next.Task.Name, prev.Task.Name}
+	if err := visit(next.Task.Name, prev.Prev, path, visited); err != nil {
+		return fmt.Errorf("cycle detected: %v", err)
+	}
+	next.Prev = append(next.Prev, prev)
+	prev.Next = append(prev.Next, next)
+	return nil
+}
+
+func visit(currentName string, nodes []*Node, path []string, visited map[string]bool) error {
+	for _, n := range nodes {
+		path = append(path, n.Task.Name)
+		if _, ok := visited[n.Task.Name]; ok {
+			return fmt.Errorf(getVisitedPath(path))
+		}
+		visited[currentName+"."+n.Task.Name] = true
+		if err := visit(n.Task.Name, n.Prev, path, visited); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func getVisitedPath(path []string) string {
+	// Reverse the path since we traversed the DAG using prev pointers.
+	for i := len(path)/2 - 1; i >= 0; i-- {
+		opp := len(path) - 1 - i
+		path[i], path[opp] = path[opp], path[i]
+	}
+	return strings.Join(path, " -> ")
+}
+
+func addLink(pt PipelineTask, previousTask string, nodes map[string]*Node) error {
+	prev, ok := nodes[previousTask]
+	if !ok {
+		return fmt.Errorf("Task %s depends on %s but %s wasn't present in Pipeline", pt.Name, previousTask, previousTask)
+	}
+	next, _ := nodes[pt.Name]
+	if err := linkPipelineTasks(prev, next); err != nil {
+		return fmt.Errorf("Couldn't create link from %s to %s: %v", prev.Task.Name, next.Task.Name, err)
+	}
+	return nil
+}
+
+// BuildDAG returns a valid pipeline DAG. Returns error if the pipeline is invalid
+func BuildDAG(tasks []PipelineTask) (*DAG, error) {
+	d := newDAG()
+
+	// Add all Tasks mentioned in the `PipelineSpec`
+	for _, pt := range tasks {
+		if _, err := d.addPipelineTask(pt); err != nil {
+			return nil, fmt.Errorf("task %s is already present in DAG, can't add it again: %v", pt.Name, err)
+		}
+	}
+	// Process all from and runAfter constraints to add task dependency
+	for _, pt := range tasks {
+		for _, previousTask := range pt.RunAfter {
+			if err := addLink(pt, previousTask, d.Nodes); err != nil {
+				return nil, fmt.Errorf("couldn't add link between %s and %s: %v", pt.Name, previousTask, err)
+			}
+		}
+		if pt.Resources != nil {
+			for _, rd := range pt.Resources.Inputs {
+				for _, previousTask := range rd.From {
+					if err := addLink(pt, previousTask, d.Nodes); err != nil {
+						return nil, fmt.Errorf("couldn't add link between %s and %s: %v", pt.Name, previousTask, err)
+					}
+				}
+			}
+		}
+	}
+	return d, nil
+}