Repository: argoproj/argo (go)

Issue:
[Bug] Global output parameters validation error with reversed dag.tasks order
## Summary 

What happened/what you expected to happen?

A workflow using global output parameter and DAG could work well with Argo, but if I reverse the dag.tasks list, argo server would return a global output parameters validation error like:
```
{
    "code": 2,
    "message": "templates.dag.tasks.second templates.second: failed to resolve {{workflow.outputs.parameters.first-global-name}}"
}
```

Expection:
Parameter verification should not be related to DAG tasks' order.

## Diagnostics

What Kubernetes provider are you using? 

Bare-metal v1.13.12

What version of Argo Workflows are you running? 

v2.11.3

The workflow works well:
```
POST {argo-server}/api/v1/workflows/test4
{
    "namespace":"test4",
    "workflow":{
        "metadata":{
            "name":"workflow-works-well",
            "namespace":"test4"
        },
        "spec":{
            "templates":[
                {
                    "name":"dag",
                    "dag":{
                        "tasks":[
                            {
                                "name":"first",
                                "template":"first"
                            },
                            {
                                "name":"second",
                                "template":"second",
                                "dependencies":[
                                    "first"
                                ]
                            }
                        ]
                    }
                },
                {
                    "name":"first",
                    "container":{
                        "name":"test",
                        "image":"ubuntu",
                        "command":[
                            "sh",
                            "-c",
                            "echo first > /tmp/empty-dir/output.txt"
                        ],
                        "volumeMounts":[
                            {
                                "mountPath":"/tmp/empty-dir",
                                "name":"output"
                            }
                        ]
                    },
                    "outputs":{
                        "parameters":[
                            {
                                "name":"first-output",
                                "valueFrom":{
                                    "path":"/tmp/empty-dir/output.txt"
                                },
                                "globalName":"first-global-name"
                            }
                        ]
                    },
                    "volumes":[
                        {
                            "name":"output",
                            "emptyDir":{

                            }
                        }
                    ]
                },
                {
                    "name":"second",
                    "container":{
                        "name":"test",
                        "image":"ubuntu",
                        "command":[
                            "sh",
                            "-c",
                            "echo {{workflow.outputs.parameters.first-global-name}}"
                        ]
                    }
                }
            ],
            "entrypoint":"dag",
            "serviceAccountName":"workflow"
        }
    }
}
```

The workflow suffers from a validation error:
```
POST {argo-server}/api/v1/workflows/test4
{
    "namespace":"test4",
    "workflow":{
        "metadata":{
            "name":"workflow-error",
            "namespace":"test4"
        },
        "spec":{
            "templates":[
                {
                    "name":"dag",
                    "dag":{
                        "tasks":[
                            {
                                "name":"second",
                                "template":"second",
                                "dependencies":[
                                    "first"
                                ]
                            },
                            {
                                "name":"first",
                                "template":"first"
                            }
                        ]
                    }
                },
                {
                    "name":"first",
                    "container":{
                        "name":"test",
                        "image":"ubuntu",
                        "command":[
                            "sh",
                            "-c",
                            "echo first > /tmp/empty-dir/output.txt"
                        ],
                        "volumeMounts":[
                            {
                                "mountPath":"/tmp/empty-dir",
                                "name":"output"
                            }
                        ]
                    },
                    "outputs":{
                        "parameters":[
                            {
                                "name":"first-output",
                                "valueFrom":{
                                    "path":"/tmp/empty-dir/output.txt"
                                },
                                "globalName":"first-global-name"
                            }
                        ]
                    },
                    "volumes":[
                        {
                            "name":"output",
                            "emptyDir":{

                            }
                        }
                    ]
                },
                {
                    "name":"second",
                    "container":{
                        "name":"test",
                        "image":"ubuntu",
                        "command":[
                            "sh",
                            "-c",
                            "echo {{workflow.outputs.parameters.first-global-name}}"
                        ]
                    }
                }
            ],
            "entrypoint":"dag",
            "serviceAccountName":"workflow"
        }
    }
}
```

The only different thing is the order of dag.tasks.


```
Paste the logs from the argo-server:
kubectl logs argo-server-6d5dd8655d-bh9dj -n argo

time="2020-10-14T12:11:04Z" level=error msg="finished unary call with code Unknown" error="templates.dag.tasks.second templates.second: failed to resolve {{workflow.outputs.parameters.first-global-name}}" grpc.code=Unknown grpc.method=CreateWorkflow grpc.service=workflow.WorkflowService grpc.start_time="2020-10-14T12:11:04Z" grpc.time_ms=0.322 span.kind=server system=grpc
```

---
<!-- Issue Author: Don't delete this message to encourage other users to support your issue! -->
**Message from the maintainers**:

Impacted by this bug? Give it a 👍. We prioritise the issues with the most 👍.


Pull request:
fix(argo-server): fix global variable validation error with reversed dag.tasks. Fixes #4273
Checklist:

* [x] Either (a) I've created an [enhancement proposal](https://github.com/argoproj/argo/issues/new/choose) and discussed it with the community, (b) this is a bug fix, or (c) this is a chore.
* [x] The title of the PR is (a) [conventional](https://www.conventionalcommits.org/en/v1.0.0/), (b) states what changed, and (c) suffixes the related issues number. E.g. `"fix(controller): Updates such and such. Fixes #1234"`.  
* [ ] My organization is added to [USERS.md](https://github.com/argoproj/argo/blob/master/USERS.md).
* [x] I've signed the CLA.
* [x] I have written unit and/or e2e tests for my change. PRs without these are unlikely to be merged.
* [x] My builds are green. Try syncing with master if they are not. 

Fixes #4273


Patch:
diff --git a/util/sorting/topological_sorting.go b/util/sorting/topological_sorting.go
new file mode 100644
index 000000000..f0e45e0f2
--- /dev/null
+++ b/util/sorting/topological_sorting.go
@@ -0,0 +1,59 @@
+package sorting
+
+import (
+	"fmt"
+)
+
+type TopologicalSortingNode struct {
+	NodeName     string
+	Dependencies []string
+}
+
+func TopologicalSorting(graph []*TopologicalSortingNode) ([]*TopologicalSortingNode, error) {
+	priorNodeCountMap := make(map[string]int, len(graph))               // nodeName -> priorNodeCount
+	nextNodeMap := make(map[string][]string, len(graph))                // nodeName -> nextNodeList
+	nodeNameMap := make(map[string]*TopologicalSortingNode, len(graph)) // nodeName -> node
+	for _, node := range graph {
+		if _, ok := nodeNameMap[node.NodeName]; ok {
+			return nil, fmt.Errorf("duplicated nodeName %s", node.NodeName)
+		}
+		nodeNameMap[node.NodeName] = node
+		priorNodeCountMap[node.NodeName] = len(node.Dependencies)
+	}
+	for _, node := range graph {
+		for _, dependency := range node.Dependencies {
+			if _, ok := nodeNameMap[dependency]; !ok {
+				return nil, fmt.Errorf("invalid dependency %s", dependency)
+			}
+			nextNodeMap[dependency] = append(nextNodeMap[dependency], node.NodeName)
+		}
+	}
+
+	queue := make([]*TopologicalSortingNode, len(graph))
+	head, tail := 0, 0
+	for nodeName, priorNodeCount := range priorNodeCountMap {
+		if priorNodeCount == 0 {
+			queue[tail] = nodeNameMap[nodeName]
+			tail += 1
+		}
+	}
+
+	for head < len(queue) {
+		curr := queue[head]
+		if curr == nil {
+			return nil, fmt.Errorf("graph with cycle")
+		}
+		for _, next := range nextNodeMap[curr.NodeName] {
+			if priorNodeCountMap[next] > 0 {
+				if priorNodeCountMap[next] == 1 {
+					queue[tail] = nodeNameMap[next]
+					tail += 1
+				}
+				priorNodeCountMap[next] -= 1
+			}
+		}
+		head += 1
+	}
+
+	return queue, nil
+}
diff --git a/workflow/validate/validate.go b/workflow/validate/validate.go
index fc6ac611b..546549983 100644
--- a/workflow/validate/validate.go
+++ b/workflow/validate/validate.go
@@ -23,6 +23,7 @@ import (
 	"github.com/argoproj/argo/util"
 	"github.com/argoproj/argo/util/help"
 	"github.com/argoproj/argo/util/intstr"
+	"github.com/argoproj/argo/util/sorting"
 	"github.com/argoproj/argo/workflow/artifacts/hdfs"
 	"github.com/argoproj/argo/workflow/common"
 	"github.com/argoproj/argo/workflow/metrics"
@@ -1180,6 +1181,12 @@ func (ctx *templateValidationCtx) validateDAG(scope map[string]interface{}, tmpl
 	if len(tmpl.DAG.Tasks) == 0 {
 		return errors.Errorf(errors.CodeBadRequest, "templates.%s must have at least one task", tmpl.Name)
 	}
+
+	err = sortDAGTasks(tmpl)
+	if err != nil {
+		return errors.Errorf(errors.CodeBadRequest, "templates.%s sorting failed: %s", tmpl.Name, err.Error())
+	}
+
 	err = validateWorkflowFieldNames(tmpl.DAG.Tasks)
 	if err != nil {
 		return errors.Errorf(errors.CodeBadRequest, "templates.%s.tasks%s", tmpl.Name, err.Error())
@@ -1347,6 +1354,27 @@ func verifyNoCycles(tmpl *wfv1.Template, ctx *dagValidationContext) error {
 	return nil
 }
 
+func sortDAGTasks(tmpl *wfv1.Template) error {
+	taskMap := make(map[string]*wfv1.DAGTask, len(tmpl.DAG.Tasks))
+	sortingGraph := make([]*sorting.TopologicalSortingNode, len(tmpl.DAG.Tasks))
+	for index := range tmpl.DAG.Tasks {
+		taskMap[tmpl.DAG.Tasks[index].Name] = &tmpl.DAG.Tasks[index]
+		sortingGraph[index] = &sorting.TopologicalSortingNode{
+			NodeName:     tmpl.DAG.Tasks[index].Name,
+			Dependencies: tmpl.DAG.Tasks[index].Dependencies,
+		}
+	}
+	sortingResult, err := sorting.TopologicalSorting(sortingGraph)
+	if err != nil {
+		return err
+	}
+	tmpl.DAG.Tasks = make([]wfv1.DAGTask, len(tmpl.DAG.Tasks))
+	for index, node := range sortingResult {
+		tmpl.DAG.Tasks[index] = *taskMap[node.NodeName]
+	}
+	return nil
+}
+
 var (
 	// paramRegex matches a parameter. e.g. {{inputs.parameters.blah}}
 	paramRegex               = regexp.MustCompile(`{{[-a-zA-Z0-9]+(\.[-a-zA-Z0-9_]+)*}}`)


Test patch:
diff --git a/util/sorting/topological_sorting_test.go b/util/sorting/topological_sorting_test.go
new file mode 100644
index 000000000..872ef94b4
--- /dev/null
+++ b/util/sorting/topological_sorting_test.go
@@ -0,0 +1,202 @@
+package sorting
+
+import (
+	"strings"
+	"testing"
+)
+
+func graphToString(graph []*TopologicalSortingNode) string {
+	var nodeNames []string
+	for _, node := range graph {
+		nodeNames = append(nodeNames, node.NodeName)
+	}
+	return strings.Join(nodeNames, ",")
+}
+
+func TestTopologicalSorting_EmptyInput(t *testing.T) {
+	result, err := TopologicalSorting([]*TopologicalSortingNode{})
+	if err != nil {
+		t.Error(err)
+	}
+	if len(result) != 0 {
+		t.Error("return value not empty", result)
+	}
+}
+
+func TestTopologicalSorting_DuplicatedNode(t *testing.T) {
+	graph := []*TopologicalSortingNode{
+		{
+			NodeName: "a",
+		},
+		{
+			NodeName: "b",
+			Dependencies: []string{
+				"a",
+			},
+		},
+		{
+			NodeName: "a",
+			Dependencies: []string{
+				"b",
+			},
+		},
+	}
+	_, err := TopologicalSorting(graph)
+	if err == nil {
+		t.Error("error missing")
+	}
+}
+
+func TestTopologicalSorting_InvalidDependency(t *testing.T) {
+	graph := []*TopologicalSortingNode{
+		{
+			NodeName: "a",
+		},
+		{
+			NodeName: "b",
+			Dependencies: []string{
+				"a",
+			},
+		},
+		{
+			NodeName: "c",
+			Dependencies: []string{
+				"a",
+				"d",
+			},
+		},
+	}
+	_, err := TopologicalSorting(graph)
+	if err == nil {
+		t.Error("error missing")
+	}
+}
+
+func TestTopologicalSorting_GraphWithCycle(t *testing.T) {
+	graph := []*TopologicalSortingNode{
+		{
+			NodeName: "a",
+			Dependencies: []string{
+				"b",
+			},
+		},
+		{
+			NodeName: "b",
+			Dependencies: []string{
+				"a",
+			},
+		},
+	}
+	_, err := TopologicalSorting(graph)
+	if err == nil {
+		t.Error("error missing")
+	}
+}
+
+func TestTopologicalSorting_GraphWithCycle2(t *testing.T) {
+	graph := []*TopologicalSortingNode{
+		{
+			NodeName: "a",
+		},
+		{
+			NodeName: "b",
+			Dependencies: []string{
+				"a",
+				"c",
+			},
+		},
+		{
+			NodeName: "c",
+			Dependencies: []string{
+				"a",
+				"b",
+			},
+		},
+	}
+	_, err := TopologicalSorting(graph)
+	if err == nil {
+		t.Error("error missing")
+	}
+}
+
+func TestTopologicalSorting_ValidInput(t *testing.T) {
+	graph := []*TopologicalSortingNode{
+		{
+			NodeName: "a",
+		},
+		{
+			NodeName: "b",
+			Dependencies: []string{
+				"a",
+			},
+		},
+		{
+			NodeName: "c",
+			Dependencies: []string{
+				"b",
+			},
+		},
+	}
+	result, err := TopologicalSorting(graph)
+	if err != nil {
+		t.Error(err)
+	}
+	resultStr := graphToString(result)
+	if resultStr != "a,b,c" {
+		t.Error("wrong output", resultStr)
+	}
+}
+
+func TestTopologicalSorting_ValidInput2(t *testing.T) {
+	graph := []*TopologicalSortingNode{
+		{
+			NodeName: "a",
+		},
+		{
+			NodeName: "b",
+			Dependencies: []string{
+				"a",
+			},
+		},
+		{
+			NodeName: "c",
+			Dependencies: []string{
+				"a",
+			},
+		},
+		{
+			NodeName: "d",
+			Dependencies: []string{
+				"b",
+				"c",
+			},
+		},
+	}
+	result, err := TopologicalSorting(graph)
+	if err != nil {
+		t.Error(err)
+	}
+	resultStr := graphToString(result)
+	if resultStr != "a,b,c,d" && resultStr != "a,c,b,d" {
+		t.Error("wrong output", resultStr)
+	}
+}
+
+func TestTopologicalSorting_ValidInput3(t *testing.T) {
+	graph := []*TopologicalSortingNode{
+		{
+			NodeName: "a",
+		},
+		{
+			NodeName: "b",
+		},
+	}
+	result, err := TopologicalSorting(graph)
+	if err != nil {
+		t.Error(err)
+	}
+	resultStr := graphToString(result)
+	if resultStr != "a,b" && resultStr != "b,a" {
+		t.Error("wrong output", resultStr)
+	}
+}
diff --git a/workflow/validate/validate_dag_test.go b/workflow/validate/validate_dag_test.go
index ba4c9fd3f..8313b680e 100644
--- a/workflow/validate/validate_dag_test.go
+++ b/workflow/validate/validate_dag_test.go
@@ -189,6 +189,68 @@ spec:
             value: "{{tasks.B.outputs.parameters.unresolvable}}"
 `
 
+var dagResolvedGlobalVar = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: dag-global-var-
+spec:
+  entrypoint: unresolved
+  templates:
+  - name: first
+    container:
+      image: alpine:3.7
+    outputs:
+      parameters:
+      - name: hosts
+        valueFrom:
+          path: /etc/hosts
+        globalName: global
+  - name: second
+    container:
+      image: alpine:3.7
+      command: [echo, "{{workflow.outputs.parameters.global}}"]
+  - name: unresolved
+    dag:
+      tasks:
+      - name: A
+        template: first
+      - name: B
+        dependencies: [A]
+        template: second
+`
+
+var dagResolvedGlobalVarReversed = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: dag-global-var-
+spec:
+  entrypoint: unresolved
+  templates:
+  - name: first
+    container:
+      image: alpine:3.7
+    outputs:
+      parameters:
+      - name: hosts
+        valueFrom:
+          path: /etc/hosts
+        globalName: global
+  - name: second
+    container:
+      image: alpine:3.7
+      command: [echo, "{{workflow.outputs.parameters.global}}"]
+  - name: unresolved
+    dag:
+      tasks:
+      - name: B
+        dependencies: [A]
+        template: second
+      - name: A
+        template: first
+`
+
 func TestDAGVariableResolution(t *testing.T) {
 	_, err := validate(dagUnresolvedVar)
 	if assert.NotNil(t, err) {
@@ -201,6 +263,11 @@ func TestDAGVariableResolution(t *testing.T) {
 	if assert.NotNil(t, err) {
 		assert.Contains(t, err.Error(), "failed to resolve {{tasks.B.outputs.parameters.unresolvable}}")
 	}
+
+	_, err = validate(dagResolvedGlobalVar)
+	assert.NoError(t, err)
+	_, err = validate(dagResolvedGlobalVarReversed)
+	assert.NoError(t, err)
 }
 
 var dagResolvedArt = `