Repository: argoproj/argo
Pull request #4683 (2020-12-09 16:42:29)

fix(controller): Fix node status when daemon pod deleted but its children nodes are still running
Checklist:

* [ ] My organization is added to [USERS.md](https://github.com/argoproj/argo/blob/master/USERS.md).

<!--

Please leave your PR in draft if you don't need a review yet. 

To fix failing `CI / Codegen`, run `make codegen`. 

-->



Patch:
diff --git a/workflow/controller/dag.go b/workflow/controller/dag.go
index fabb23bff..308ca79c4 100644
--- a/workflow/controller/dag.go
+++ b/workflow/controller/dag.go
@@ -692,6 +692,11 @@ type TaskResults struct {
 // evaluateDependsLogic returns whether a node should execute and proceed. proceed means that all of its dependencies are
 // completed and execute means that given the results of its dependencies, this node should execute.
 func (d *dagContext) evaluateDependsLogic(taskName string) (bool, bool, error) {
+	node := d.getTaskNode(taskName)
+	if node != nil {
+		return true, true, nil
+	}
+
 	evalScope := make(map[string]TaskResults)
 
 	for _, taskName := range d.GetTaskDependencies(taskName) {
diff --git a/workflow/controller/operator.go b/workflow/controller/operator.go
index fa9f0f5c4..9665e0b53 100644
--- a/workflow/controller/operator.go
+++ b/workflow/controller/operator.go
@@ -926,7 +926,7 @@ func (woc *wfOperationCtx) podReconciliation(ctx context.Context) error {
 	// It is now impossible to infer pod status. We can do at this point is to mark the node with Error, or
 	// we can re-submit it.
 	for nodeID, node := range woc.wf.Status.Nodes {
-		if node.Type != wfv1.NodeTypePod || node.Fulfilled() || node.StartedAt.IsZero() {
+		if node.Type != wfv1.NodeTypePod || node.Phase.Fulfilled() || node.StartedAt.IsZero() {
 			// node is not a pod, it is already complete, or it can be re-run.
 			continue
 		}
@@ -950,6 +950,10 @@ func (woc *wfOperationCtx) podReconciliation(ctx context.Context) error {
 				continue
 			}
 
+			if node.Daemoned != nil && *node.Daemoned {
+				node.Daemoned = nil
+				woc.updated = true
+			}
 			woc.markNodePhase(node.Name, wfv1.NodeError, "pod deleted")
 		} else {
 			// At this point we are certain that the pod associated with our node is running or has been run;
@@ -1100,6 +1104,7 @@ func (woc *wfOperationCtx) assessNodeStatus(pod *apiv1.Pod, node *wfv1.NodeStatu
 		if pod.DeletionTimestamp != nil {
 			// pod is being terminated
 			newPhase = wfv1.NodeError
+			newDaemonStatus = pointer.BoolPtr(false)
 			message = "pod deleted during operation"
 			woc.log.WithField("displayName", node.DisplayName).WithField("templateName", node.TemplateName).
 				WithField("pod", pod.Name).Error(message)
diff --git a/workflow/cron/operator.go b/workflow/cron/operator.go
index 03889cc22..f76d45ed8 100644
--- a/workflow/cron/operator.go
+++ b/workflow/cron/operator.go
@@ -90,6 +90,10 @@ func (woc *cronWfOperationCtx) run(ctx context.Context, scheduledRuntime time.Ti
 
 	runWf, err := util.SubmitWorkflow(ctx, woc.wfClient, woc.wfClientset, woc.cronWf.Namespace, wf, &v1alpha1.SubmitOpts{})
 	if err != nil {
+		// If the workflow already exists (i.e. this is a duplicate submission), do not report an error
+		if errors.IsAlreadyExists(err) {
+			return
+		}
 		woc.reportCronWorkflowError(v1alpha1.ConditionTypeSubmissionError, fmt.Sprintf("Failed to submit Workflow: %s", err))
 		return
 	}


Test patch:
diff --git a/workflow/controller/dag_test.go b/workflow/controller/dag_test.go
index 01d5e1f79..350e36b4c 100644
--- a/workflow/controller/dag_test.go
+++ b/workflow/controller/dag_test.go
@@ -407,6 +407,55 @@ func TestEvaluateAnyAllDependsLogic(t *testing.T) {
 
 }
 
+func TestEvaluateDependsLogicWhenDaemonFailed(t *testing.T) {
+	testTasks := []wfv1.DAGTask{
+		{
+			Name: "A",
+		},
+		{
+			Name:    "B",
+			Depends: "A",
+		},
+	}
+
+	d := &dagContext{
+		boundaryName: "test",
+		tasks:        testTasks,
+		wf:           &wfv1.Workflow{ObjectMeta: metav1.ObjectMeta{Name: "test-wf"}},
+		dependencies: make(map[string][]string),
+		dependsLogic: make(map[string]string),
+	}
+
+	// Task A is running
+	daemon := true
+	d.wf = &wfv1.Workflow{
+		ObjectMeta: metav1.ObjectMeta{Name: "test-wf"},
+		Status: wfv1.WorkflowStatus{
+			Nodes: map[string]wfv1.NodeStatus{
+				d.taskNodeID("A"): {Phase: wfv1.NodeRunning, Daemoned: &daemon},
+			},
+		},
+	}
+
+	// Task B should proceed and execute
+	execute, proceed, err := d.evaluateDependsLogic("B")
+	assert.NoError(t, err)
+	assert.True(t, proceed)
+	assert.True(t, execute)
+
+	// Task B running
+	d.wf.Status.Nodes[d.taskNodeID("B")] = wfv1.NodeStatus{Phase: wfv1.NodeRunning}
+
+	// Task A failed or error
+	d.wf.Status.Nodes[d.taskNodeID("A")] = wfv1.NodeStatus{Phase: wfv1.NodeFailed}
+
+	// Task B should proceed and execute
+	execute, proceed, err = d.evaluateDependsLogic("B")
+	assert.NoError(t, err)
+	assert.True(t, proceed)
+	assert.True(t, execute)
+}
+
 func TestAllEvaluateDependsLogic(t *testing.T) {
 	statusMap := map[common.TaskResult]wfv1.NodePhase{
 		common.TaskResultSucceeded: wfv1.NodeSucceeded,