fix: make sure taskresult completed when mark node succeed when it ha…

…s outputs (#12537)

Signed-off-by: shuangkun <tsk2013uestc@163.com>
Signed-off-by: shuangkun tian <72060326+shuangkun@users.noreply.github.com>
Co-authored-by: Julie Vogelman <julievogelman0@gmail.com>
Signed-off-by: Isitha Subasinghe <isubasinghe@student.unimelb.edu.au>

docs/fields.md

@@ -853,7 +853,7 @@ WorkflowStatus contains overall status information about a workflow
 |`storedTemplates`|[`Template`](#template)|StoredTemplates is a mapping between a template ref and the node's status.|
 |`storedWorkflowTemplateSpec`|[`WorkflowSpec`](#workflowspec)|StoredWorkflowSpec stores the WorkflowTemplate spec for future execution.|
 |`synchronization`|[`SynchronizationStatus`](#synchronizationstatus)|Synchronization stores the status of synchronization locks|
-|`taskResultsCompletionStatus`|`Map< boolean , string >`|TaskResultsCompletionStatus tracks task result completion status (mapped by pod name). Used to prevent premature archiving and garbage collection.|
+|`taskResultsCompletionStatus`|`Map< boolean , string >`|TaskResultsCompletionStatus tracks task result completion status (mapped by node ID). Used to prevent premature archiving and garbage collection.|
 
 ## CronWorkflowSpec
 

pkg/apis/workflow/v1alpha1/workflow_types.go

@@ -1940,7 +1940,7 @@ type WorkflowStatus struct {
 	// ArtifactGCStatus maintains the status of Artifact Garbage Collection
 	ArtifactGCStatus *ArtGCStatus `json:"artifactGCStatus,omitempty" protobuf:"bytes,19,opt,name=artifactGCStatus"`
 
-	// TaskResultsCompletionStatus tracks task result completion status (mapped by pod name). Used to prevent premature archiving and garbage collection.
+	// TaskResultsCompletionStatus tracks task result completion status (mapped by node ID). Used to prevent premature archiving and garbage collection.
 	TaskResultsCompletionStatus map[string]bool `json:"taskResultsCompletionStatus,omitempty" protobuf:"bytes,20,opt,name=taskResultsCompletionStatus"`
 }
 
@@ -1967,6 +1967,14 @@ func (ws *WorkflowStatus) TaskResultsInProgress() bool {
 	return false
 }
 
+func (ws *WorkflowStatus) IsTaskResultIncomplete(name string) bool {
+	value, found := ws.TaskResultsCompletionStatus[name]
+	if found {
+		return !value
+	}
+	return true
+}
+
 func (ws *WorkflowStatus) IsOffloadNodeStatus() bool {
 	return ws.OffloadNodeStatusVersion != ""
 }

workflow/controller/controller_test.go

@@ -510,6 +510,10 @@ func makePodsPhase(ctx context.Context, woc *wfOperationCtx, phase apiv1.PodPhas
 			if err != nil {
 				panic(err)
 			}
+			if phase == apiv1.PodSucceeded {
+				nodeID := woc.nodeID(&pod)
+				woc.wf.Status.MarkTaskResultComplete(nodeID)
+			}
 		}
 	}
 }

workflow/controller/exit_handler_test.go

@@ -209,6 +209,7 @@ func TestStepsOnExitTmplWithArt(t *testing.T) {
 				},
 			}
 			woc.wf.Status.Nodes[idx] = node
+			woc.wf.Status.MarkTaskResultComplete(node.ID)
 		}
 	}
 	woc1 := newWorkflowOperationCtx(woc.wf, controller)
@@ -283,6 +284,7 @@ func TestDAGOnExitTmplWithArt(t *testing.T) {
 				},
 			}
 			woc.wf.Status.Nodes[idx] = node
+			woc.wf.Status.MarkTaskResultComplete(node.ID)
 		}
 	}
 	woc1 := newWorkflowOperationCtx(woc.wf, controller)
@@ -383,6 +385,7 @@ func TestStepsTmplOnExit(t *testing.T) {
 				},
 			}
 			woc2.wf.Status.Nodes[idx] = node
+			woc.wf.Status.MarkTaskResultComplete(node.ID)
 		}
 	}
 
@@ -487,6 +490,7 @@ func TestDAGOnExit(t *testing.T) {
 				},
 			}
 			woc2.wf.Status.Nodes[idx] = node
+			woc.wf.Status.MarkTaskResultComplete(node.ID)
 		}
 	}
 	woc3 := newWorkflowOperationCtx(woc2.wf, controller)

workflow/controller/hooks_test.go

@@ -997,7 +997,7 @@ spec:
 	assert.Equal(t, wfv1.NodePending, node.Phase)
 	makePodsPhase(ctx, woc, apiv1.PodFailed)
 	woc = newWorkflowOperationCtx(woc.wf, controller)
-	err := woc.podReconciliation(ctx)
+	err, _ := woc.podReconciliation(ctx)
 	assert.NoError(t, err)
 	node = woc.wf.Status.Nodes.FindByDisplayName("hook-failures.hooks.failure")
 	assert.NotNil(t, node)
@@ -1140,6 +1140,7 @@ spec:
 	pod, _ := podcs.Get(ctx, "hook-running", metav1.GetOptions{})
 	pod.Status.Phase = apiv1.PodSucceeded
 	updatedPod, _ := podcs.Update(ctx, pod, metav1.UpdateOptions{})
+	woc.wf.Status.MarkTaskResultComplete(woc.nodeID(pod))
 	_ = woc.controller.podInformer.GetStore().Update(updatedPod)
 	woc = newWorkflowOperationCtx(woc.wf, controller)
 	woc.operate(ctx)
@@ -1231,6 +1232,7 @@ spec:
 	pod.Status.Phase = apiv1.PodSucceeded
 	updatedPod, _ := podcs.Update(ctx, &pod, metav1.UpdateOptions{})
 	_ = woc.controller.podInformer.GetStore().Update(updatedPod)
+	woc.wf.Status.MarkTaskResultComplete(woc.nodeID(&pod))
 	woc = newWorkflowOperationCtx(woc.wf, controller)
 	woc.operate(ctx)
 	assert.Equal(t, wfv1.Progress("1/2"), woc.wf.Status.Progress)

workflow/controller/operator.go

@@ -307,7 +307,7 @@ func (woc *wfOperationCtx) operate(ctx context.Context) {
 		woc.wf.Status.EstimatedDuration = woc.estimateWorkflowDuration()
 	} else {
 		woc.workflowDeadline = woc.getWorkflowDeadline()
-		err = woc.podReconciliation(ctx)
+		err, podReconciliationCompleted := woc.podReconciliation(ctx)
 		if err == nil {
 			woc.failSuspendedAndPendingNodesAfterDeadlineOrShutdown()
 		}
@@ -318,6 +318,12 @@ func (woc *wfOperationCtx) operate(ctx context.Context) {
 			// TODO: we need to re-add to the workqueue, but should happen in caller
 			return
 		}
+
+		if !podReconciliationCompleted {
+			woc.log.WithField("workflow", woc.wf.ObjectMeta.Name).Info("pod reconciliation didn't complete, will retry")
+			woc.requeue()
+			return
+		}
 	}
 
 	if woc.ShouldSuspend() {
@@ -1088,15 +1094,17 @@ func (woc *wfOperationCtx) processNodeRetries(node *wfv1.NodeStatus, retryStrate
 // pods and update the node state before continuing the evaluation of the workflow.
 // Records all pods which were observed completed, which will be labeled completed=true
 // after successful persist of the workflow.
-func (woc *wfOperationCtx) podReconciliation(ctx context.Context) error {
+// returns whether pod reconciliation successfully completed
+func (woc *wfOperationCtx) podReconciliation(ctx context.Context) (error, bool) {
 	podList, err := woc.getAllWorkflowPods()
 	if err != nil {
-		return err
+		return err, false
 	}
 	seenPods := make(map[string]*apiv1.Pod)
 	seenPodLock := &sync.Mutex{}
 	wfNodesLock := &sync.RWMutex{}
 	podRunningCondition := wfv1.Condition{Type: wfv1.ConditionTypePodRunning, Status: metav1.ConditionFalse}
+	taskResultIncomplete := false
 	performAssessment := func(pod *apiv1.Pod) {
 		if pod == nil {
 			return
@@ -1115,6 +1123,12 @@ func (woc *wfOperationCtx) podReconciliation(ctx context.Context) error {
 		node, err := woc.wf.Status.Nodes.Get(nodeID)
 		if err == nil {
 			if newState := woc.assessNodeStatus(pod, node); newState != nil {
+				// Check whether its taskresult is in an incompleted state.
+				if newState.Succeeded() && woc.wf.Status.IsTaskResultIncomplete(node.ID) {
+					woc.log.WithFields(log.Fields{"nodeID": newState.ID}).Debug("Taskresult of the node not yet completed")
+					taskResultIncomplete = true
+					return
+				}
 				woc.addOutputsToGlobalScope(newState.Outputs)
 				if newState.MemoizationStatus != nil {
 					if newState.Succeeded() {
@@ -1158,6 +1172,12 @@ func (woc *wfOperationCtx) podReconciliation(ctx context.Context) error {
 
 	wg.Wait()
 
+	// If true, it means there are some nodes which have outputs we wanted to be marked succeed, but the node's taskresults didn't completed.
+	// We should make sure the taskresults processing is complete as it will be possible to reference it in the next step.
+	if taskResultIncomplete {
+		return nil, false
+	}
+
 	woc.wf.Status.Conditions.UpsertCondition(podRunningCondition)
 
 	// Now check for deleted pods. Iterate our nodes. If any one of our nodes does not show up in
@@ -1197,7 +1217,7 @@ func (woc *wfOperationCtx) podReconciliation(ctx context.Context) error {
 			woc.markNodePhase(node.Name, wfv1.NodeError, "pod deleted")
 		}
 	}
-	return nil
+	return nil, !taskResultIncomplete
 }
 
 func (woc *wfOperationCtx) nodeID(pod *apiv1.Pod) string {

workflow/controller/operator_concurrency_test.go

@@ -196,7 +196,7 @@ func TestSemaphoreTmplLevel(t *testing.T) {
 		woc_two.operate(ctx)
 
 		// Check Node status
-		err = woc_two.podReconciliation(ctx)
+		err, _ = woc_two.podReconciliation(ctx)
 		assert.NoError(t, err)
 		for _, node := range woc_two.wf.Status.Nodes {
 			assert.Equal(t, wfv1.NodePending, node.Phase)
@@ -257,7 +257,7 @@ func TestSemaphoreScriptTmplLevel(t *testing.T) {
 		woc_two.operate(ctx)
 
 		// Check Node status
-		err = woc_two.podReconciliation(ctx)
+		err, _ = woc_two.podReconciliation(ctx)
 		assert.NoError(t, err)
 		for _, node := range woc_two.wf.Status.Nodes {
 			assert.Equal(t, wfv1.NodePending, node.Phase)
@@ -319,7 +319,7 @@ func TestSemaphoreScriptConfigMapInDifferentNamespace(t *testing.T) {
 		woc_two.operate(ctx)
 
 		// Check Node status
-		err = woc_two.podReconciliation(ctx)
+		err, _ = woc_two.podReconciliation(ctx)
 		assert.NoError(t, err)
 		for _, node := range woc_two.wf.Status.Nodes {
 			assert.Equal(t, wfv1.NodePending, node.Phase)
@@ -379,7 +379,7 @@ func TestSemaphoreResourceTmplLevel(t *testing.T) {
 		woc_two.operate(ctx)
 
 		// Check Node status
-		err = woc_two.podReconciliation(ctx)
+		err, _ = woc_two.podReconciliation(ctx)
 		assert.NoError(t, err)
 		for _, node := range woc_two.wf.Status.Nodes {
 			assert.Equal(t, wfv1.NodePending, node.Phase)
@@ -416,7 +416,7 @@ func TestSemaphoreWithOutConfigMap(t *testing.T) {
 		wf, err := controller.wfclientset.ArgoprojV1alpha1().Workflows(wf.Namespace).Create(ctx, wf, metav1.CreateOptions{})
 		assert.NoError(t, err)
 		woc := newWorkflowOperationCtx(wf, controller)
-		err = woc.podReconciliation(ctx)
+		err, _ = woc.podReconciliation(ctx)
 		assert.NoError(t, err)
 		for _, node := range woc.wf.Status.Nodes {
 			assert.Equal(t, wfv1.NodePending, node.Phase)