Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Improve DAG validation for pipelines with hundreds of tasks #5421

Merged
merged 1 commit into from
Sep 5, 2022
Merged

Improve DAG validation for pipelines with hundreds of tasks #5421

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 11 additions & 3 deletions pkg/apis/pipeline/v1/pipeline_validation_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -569,9 +569,11 @@ func TestPipelineSpec_Validate_Failure_CycleDAG(t *testing.T) {
name := "invalid pipeline spec with DAG having cyclic dependency"
ps := &PipelineSpec{
Tasks: []PipelineTask{{
Name: "foo", TaskRef: &TaskRef{Name: "foo-task"}, RunAfter: []string{"bar"},
Name: "foo", TaskRef: &TaskRef{Name: "foo-task"}, RunAfter: []string{"baz"},
}, {
Name: "bar", TaskRef: &TaskRef{Name: "bar-task"}, RunAfter: []string{"foo"},
}, {
Name: "baz", TaskRef: &TaskRef{Name: "baz-task"}, RunAfter: []string{"bar"},
}},
}
ctx := config.SkipValidationDueToPropagatedParametersAndWorkspaces(context.Background(), false)
Expand Down Expand Up @@ -679,9 +681,15 @@ func TestValidateGraph_Failure(t *testing.T) {
}, {
Name: "bar", TaskRef: &TaskRef{Name: "bar-task"}, RunAfter: []string{"foo"},
}}
if err := validateGraph(tasks); err == nil {
expectedError := apis.FieldError{
Message: `invalid value: cycle detected; task "bar" depends on "foo"`,
Paths: []string{"tasks"},
}
err := validateGraph(tasks)
if err == nil {
t.Error("Pipeline.validateGraph() did not return error for invalid DAG of pipeline tasks:", desc)

} else if d := cmp.Diff(expectedError.Error(), err.Error(), cmpopts.IgnoreUnexported(apis.FieldError{})); d != "" {
t.Errorf("Pipeline.validateGraph() errors diff %s", diff.PrintWantGot(d))
}
}

Expand Down
91 changes: 63 additions & 28 deletions pkg/reconciler/pipeline/dag/dag.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ package dag
import (
"errors"
"fmt"
"sort"
"strings"

"github.com/tektoncd/pipeline/pkg/list"
Expand Down Expand Up @@ -79,6 +80,11 @@ func Build(tasks Tasks, deps map[string][]string) (*Graph, error) {
}
}

// Ensure no cycles in the graph
if err := findCyclesInDependencies(deps); err != nil {
return nil, fmt.Errorf("cycle detected; %w", err)
}

// Process all from and runAfter constraints to add task dependency
for pt, taskDeps := range deps {
for _, previousTask := range taskDeps {
Expand Down Expand Up @@ -120,41 +126,72 @@ func GetCandidateTasks(g *Graph, doneTasks ...string) (sets.String, error) {
return d, nil
}

func linkPipelineTasks(prev *Node, next *Node) error {
// Check for self cycle
if prev.Task.HashKey() == next.Task.HashKey() {
return fmt.Errorf("cycle detected; task %q depends on itself", next.Task.HashKey())
}
// Check if we are adding cycles.
path := []string{next.Task.HashKey(), prev.Task.HashKey()}
if err := lookForNode(prev.Prev, path, next.Task.HashKey()); err != nil {
return fmt.Errorf("cycle detected: %w", err)
}
func linkPipelineTasks(prev *Node, next *Node) {
next.Prev = append(next.Prev, prev)
prev.Next = append(prev.Next, next)
return nil
}

func lookForNode(nodes []*Node, path []string, next string) error {
for _, n := range nodes {
path = append(path, n.Task.HashKey())
if n.Task.HashKey() == next {
return errors.New(getVisitedPath(path))
// use Kahn's algorithm to find cycles in dependencies
func findCyclesInDependencies(deps map[string][]string) error {
rafalbigaj marked this conversation as resolved.
Show resolved Hide resolved
independentTasks := sets.NewString()
dag := make(map[string]sets.String, len(deps))
childMap := make(map[string]sets.String, len(deps))
for task, taskDeps := range deps {
if len(taskDeps) == 0 {
continue
}
if err := lookForNode(n.Prev, path, next); err != nil {
return err
dag[task] = sets.NewString(taskDeps...)
for _, dep := range taskDeps {
if len(deps[dep]) == 0 {
independentTasks.Insert(dep)
}
Comment on lines +145 to +147
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I can see it's an optimization originating from the fact that no topo-sort order is actually built or printed (as "independent" tasks that are also "final" ones --- are not even put into the independentTasks set).

if children, ok := childMap[dep]; ok {
children.Insert(task)
} else {
childMap[dep] = sets.NewString(task)
}
}
}
return nil

for {
parent, ok := independentTasks.PopAny()
if !ok {
break
}
children := childMap[parent]
for {
child, ok := children.PopAny()
if !ok {
break
}
dag[child].Delete(parent)
if dag[child].Len() == 0 {
independentTasks.Insert(child)
delete(dag, child)
}
rafalbigaj marked this conversation as resolved.
Show resolved Hide resolved
}
}

return getInterdependencyError(dag)
}

func getVisitedPath(path []string) string {
// Reverse the path since we traversed the Graph 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]
func getInterdependencyError(dag map[string]sets.String) error {
if len(dag) == 0 {
return nil
}
return strings.Join(path, " -> ")
firstChild := ""
for task := range dag {
if firstChild == "" || firstChild > task {
firstChild = task
}
}
deps := dag[firstChild].List()
depNames := make([]string, 0, len(deps))
sort.Strings(deps)
for _, dep := range deps {
depNames = append(depNames, fmt.Sprintf("%q", dep))
}
return fmt.Errorf("task %q depends on %s", firstChild, strings.Join(depNames, ", "))
}

func addLink(pt string, previousTask string, nodes map[string]*Node) error {
Expand All @@ -163,9 +200,7 @@ func addLink(pt string, previousTask string, nodes map[string]*Node) error {
return fmt.Errorf("task %s depends on %s but %s wasn't present in Pipeline", pt, previousTask, previousTask)
}
next := nodes[pt]
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is holding a huge struct (object) in real world applications. It is set to the entire pipelineTask specification along with a list of params, when expressions, the entire specification when taskSpec is specified, etc. The pipelineTask specification is not required at this point since all the dependencies (runAfter, task results in params and when expressions, and from) are calculated before calling dag.Build. All we need is just the pipelineTask.Name i.e. HashKey(). We have a room of improvement in future to avoid passing the entire blob around.

if err := linkPipelineTasks(prev, next); err != nil {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I had troubleshooted till this point, thanks for taking it further, appreciate your efforts 🙏

return fmt.Errorf("couldn't create link from %s to %s: %w", prev.Task.HashKey(), next.Task.HashKey(), err)
}
linkPipelineTasks(prev, next)
return nil
}

Expand Down
139 changes: 139 additions & 0 deletions pkg/reconciler/pipeline/dag/dag_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ limitations under the License.
package dag_test

import (
"fmt"
"strings"
"testing"

Expand Down Expand Up @@ -549,6 +550,78 @@ func TestBuild_InvalidDAG(t *testing.T) {
}
}

func TestBuildGraphWithHundredsOfTasks_Success(t *testing.T) {
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe some performance check too?

Copy link
Member

@pritidesai pritidesai Sep 2, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes please, so that we do not run into it in future, thanks!

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The problem with performance checks is that they could be inherently flaky because of the variable performance of the test nodes. If we do add performance checks, I would suggest for now to only log the execution time.
We can collect such timings over time for some time and then decide on an acceptable bar for the execution time.

Copy link
Member

@pritidesai pritidesai Sep 2, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That's a great idea @afrittoli, where do we store those timings?

I was running the test you had introduced #3524 (a huge thanks to you for introducing such test):

pipeline/pkg/reconciler/pipelinerun/resources/pipelinerunstate_test.go

Line 1404 in 99404d5

func buildPipelineStateWithLargeDepencyGraph(t *testing.T) PipelineRunState {

I ran it locally yesterday multiple times which took 60 seconds 😲 (without this PR).

At the time introducing this test, it took less than the default timeout 30 seconds based on the PR description (if I am reading it right):

This changes adds a unit test that reproduces the issue in https://github.com/tektoncd/pipeline/issues/3521, which
used to fail (with timeout 30s) and now succeedes for pipelines roughly
up to 120 tasks / 120 links. On my laptop, going beyond 120 tasks/links
takes longer than 30s, so I left the unit test at 80 to avoid
introducing a flaky test in CI. There is still work to do to improve
this further, some profiling / tracing work might help.

There has been many changes introduced since then, we really need a way to flag us (nightly performance tests) when we introduce any delay.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nightly performance test or like you are suggesting to collect timings over time. I am fine with either option.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes. The TestBuildGraphWithHundredsOfTasks_Success in a way is a kind of performance test (like the one I added at the time), because if things slow down significantly, the tests will eventually timeout.
If we collected test execution times and graphed them over time - or if we had some dedicated nightly performance test, we would be able to see a change in execution time sooner than just waiting for the tests to timeout.

That is something that we would need to setup as part of the CI infrastructure. Would you like to create an issue about that?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yup definitely, we had a PR from @guillaumerose #4378 which didn't materialize but at least most of us including @vdemeester and @imjasonh were on board with the idea of running nightly performance tests.

var tasks []v1beta1.PipelineTask
// separate branches with sequential tasks and redundant links (each task explicitly depends on all predecessors)
// b00 - 000 - 001 - ... - 100
// b01 - 000 - 001 - ... - 100
// ..
// b04 - 000 - 001 - ... - 100
nBranches, nTasks := 5, 100
for branchIdx := 0; branchIdx < nBranches; branchIdx++ {
var taskDeps []string
firstTaskName := fmt.Sprintf("b%02d", branchIdx)
firstTask := v1beta1.PipelineTask{
Name: firstTaskName,
TaskRef: &v1beta1.TaskRef{Name: firstTaskName + "-task"},
RunAfter: taskDeps,
}
tasks = append(tasks, firstTask)
taskDeps = append(taskDeps, firstTaskName)
for taskIdx := 0; taskIdx < nTasks; taskIdx++ {
taskName := fmt.Sprintf("%s-%03d", firstTaskName, taskIdx)
task := v1beta1.PipelineTask{
Name: taskName,
TaskRef: &v1beta1.TaskRef{Name: taskName + "-task"},
RunAfter: taskDeps,
}
tasks = append(tasks, task)
taskDeps = append(taskDeps, taskName)
}
}

_, err := dag.Build(v1beta1.PipelineTaskList(tasks), v1beta1.PipelineTaskList(tasks).Deps())
if err != nil {
t.Error(err)
}
}

func TestBuildGraphWithHundredsOfTasks_InvalidDAG(t *testing.T) {
var tasks []v1beta1.PipelineTask
// branches with circular interdependencies
nBranches, nTasks := 5, 100
for branchIdx := 0; branchIdx < nBranches; branchIdx++ {
depBranchIdx := branchIdx + 1
if depBranchIdx == nBranches {
depBranchIdx = 0
}
taskDeps := []string{fmt.Sprintf("b%02d", depBranchIdx)}
firstTaskName := fmt.Sprintf("b%02d", branchIdx)
firstTask := v1beta1.PipelineTask{
Name: firstTaskName,
TaskRef: &v1beta1.TaskRef{Name: firstTaskName + "-task"},
RunAfter: taskDeps,
}
tasks = append(tasks, firstTask)
taskDeps = append(taskDeps, firstTaskName)
for taskIdx := 0; taskIdx < nTasks; taskIdx++ {
taskName := fmt.Sprintf("%s-%03d", firstTaskName, taskIdx)
task := v1beta1.PipelineTask{
Name: taskName,
TaskRef: &v1beta1.TaskRef{Name: taskName + "-task"},
RunAfter: taskDeps,
}
tasks = append(tasks, task)
taskDeps = append(taskDeps, taskName)
}
}

_, err := dag.Build(v1beta1.PipelineTaskList(tasks), v1beta1.PipelineTaskList(tasks).Deps())
if err == nil {
t.Errorf("Pipeline.Validate() did not return error for invalid pipeline with cycles")
}
}

func testGraph(t *testing.T) *dag.Graph {
// b a
// | / \
Expand Down Expand Up @@ -630,3 +703,69 @@ func assertSameDAG(t *testing.T, l, r *dag.Graph) {
}
}
}

func TestFindCyclesInDependencies(t *testing.T) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

❤️

deps := map[string][]string{
"a": {},
"b": {"c", "d"},
"c": {},
"d": {},
}

err := dag.FindCyclesInDependencies(deps)
if err != nil {
t.Error(err)
}

tcs := []struct {
name string
deps map[string][]string
err string
}{{
name: "valid-empty-deps",
deps: map[string][]string{
"a": {},
"b": {"c", "d"},
"c": {},
"d": {},
},
}, {
name: "self-link",
deps: map[string][]string{
"a": {"a"},
},
err: `task "a" depends on "a"`,
}, {
name: "interdependent-tasks",
deps: map[string][]string{
"a": {"b"},
"b": {"a"},
},
err: `task "a" depends on "b"`,
}, {
name: "multiple-cycles",
deps: map[string][]string{
"a": {"b", "c"},
"b": {"a"},
"c": {"d"},
"d": {"a", "b"},
},
err: `task "a" depends on "b", "c"`,
},
}
for _, tc := range tcs {
t.Run(tc.name, func(t *testing.T) {
err := dag.FindCyclesInDependencies(tc.deps)
if tc.err == "" {
if err != nil {
t.Errorf("expected to see no error for valid DAG but had: %v", err)
}
} else {
if err == nil || !strings.Contains(err.Error(), tc.err) {
t.Errorf("expected to see an error: %q for invalid DAG but had: %v", tc.err, err)
}
}
})
}

}
21 changes: 21 additions & 0 deletions pkg/reconciler/pipeline/dag/export_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
/*
Copyright 2022 The Tekton 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 dag

// exports for tests

var FindCyclesInDependencies = findCyclesInDependencies
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We have a policy of testing exported functions only (generally), but I think in this case it makes sense to test findCyclesInDependencies directly!

NIT: I wonder if instead of exporting for tests, we could have the test for findCyclesInDependencies in a dedicated test module in the dag package?
Not asking to change this yet, I'd like to see what others think - and we could also change this in a different PR in case.

@tektoncd/core-maintainers wdyt?