fix: Differentiate between Fulfilled and Completed

pkg/apis/workflow/v1alpha1/workflow_types.go

@@ -1189,16 +1189,21 @@ func (n Nodes) GetResourcesDuration() ResourcesDuration {
 	return i
 }
 
+// Fulfilled returns whether a phase is fulfilled, i.e. it finished execution or was skipped
+func (phase NodePhase) Fulfilled() bool {
+	return phase.Completed() || phase == NodeSkipped
+}
+
+// Completed returns whether or not a phase completed. Notably, a skipped phase is not considered as having completed
 func (phase NodePhase) Completed() bool {
 	return phase == NodeSucceeded ||
 		phase == NodeFailed ||
-		phase == NodeError ||
-		phase == NodeSkipped
+		phase == NodeError
 }
 
 // Completed returns whether or not the workflow has completed execution
 func (ws WorkflowStatus) Completed() bool {
-	return ws.Phase.Completed()
+	return ws.Phase.Fulfilled()
 }
 
 // Successful return whether or not the workflow has succeeded
@@ -1219,9 +1224,14 @@ func (ws WorkflowStatus) FinishTime() *metav1.Time {
 	return &ws.FinishedAt
 }
 
-// Completed returns whether or not the node has completed execution
+// Fulfilled returns whether a node is fulfilled, i.e. it finished execution, was skipped, or was dameoned successfully
+func (n NodeStatus) Fulfilled() bool {
+	return n.Phase.Fulfilled() || n.IsDaemoned() && n.Phase != NodePending
+}
+
+// Completed returns whether a node completed. Notably, a skipped node is not considered as having completed
 func (n NodeStatus) Completed() bool {
-	return n.Phase.Completed() || n.IsDaemoned() && n.Phase != NodePending
+	return n.Phase.Completed()
 }
 
 func (in *WorkflowStatus) AnyActiveSuspendNode() bool {
@@ -1261,7 +1271,7 @@ func (n NodeStatus) FinishTime() *metav1.Time {
 // CanRetry returns whether the node should be retried or not.
 func (n NodeStatus) CanRetry() bool {
 	// TODO(shri): Check if there are some 'unretryable' errors.
-	return n.Completed() && !n.Successful()
+	return n.Fulfilled() && !n.Successful()
 }
 
 func (n NodeStatus) GetTemplateScope() (ResourceScope, string) {

util/printer/workflow-printer.go

@@ -89,7 +89,7 @@ func countPendingRunningCompleted(wf *wfv1.Workflow) (int, int, int) {
 		if tmpl == nil || !tmpl.IsPodType() {
 			continue
 		}
-		if node.Completed() {
+		if node.Fulfilled() {
 			completed++
 		} else if node.Phase == wfv1.NodeRunning {
 			running++

workflow/controller/dag.go

@@ -169,7 +169,7 @@ func (d *dagContext) assessDAGPhase(targetTasks []string, nodes map[string]wfv1.
 		if node.BoundaryID != d.boundaryID {
 			continue
 		}
-		if !node.Completed() {
+		if !node.Fulfilled() {
 			return wfv1.NodeRunning
 		}
 		if node.Successful() {
@@ -278,7 +278,7 @@ func (woc *wfOperationCtx) executeDAG(nodeName string, tmplCtx *templateresoluti
 	}
 
 	defer func() {
-		if woc.wf.Status.Nodes[node.ID].Completed() {
+		if woc.wf.Status.Nodes[node.ID].Fulfilled() {
 			_ = woc.killDaemonedChildren(node.ID)
 		}
 	}()
@@ -377,12 +377,14 @@ func (woc *wfOperationCtx) executeDAGTask(dagCtx *dagContext, taskName string) {
 
 	node := dagCtx.getTaskNode(taskName)
 	task := dagCtx.GetTask(taskName)
-	if node != nil && node.Completed() {
-		// Run the node's onExit node, if any.
-		hasOnExitNode, onExitNode, err := woc.runOnExitNode(task.Name, task.OnExit, dagCtx.boundaryID, dagCtx.tmplCtx)
-		if hasOnExitNode && (onExitNode == nil || !onExitNode.Completed() || err != nil) {
-			// The onExit node is either not complete or has errored out, return.
-			return
+	if node != nil && node.Fulfilled() {
+		if node.Completed() {
+			// Run the node's onExit node, if any.
+			hasOnExitNode, onExitNode, err := woc.runOnExitNode(task.Name, task.OnExit, dagCtx.boundaryID, dagCtx.tmplCtx)
+			if hasOnExitNode && (onExitNode == nil || !onExitNode.Fulfilled() || err != nil) {
+				// The onExit node is either not complete or has errored out, return.
+				return
+			}
 		}
 		return
 	}
@@ -506,7 +508,7 @@ func (woc *wfOperationCtx) executeDAGTask(dagCtx *dagContext, taskName string) {
 		for _, t := range expandedTasks {
 			// Add the child relationship from our dependency's outbound nodes to this node.
 			node := dagCtx.getTaskNode(t.Name)
-			if node == nil || !node.Completed() {
+			if node == nil || !node.Fulfilled() {
 				return
 			}
 			if !node.Successful() {
@@ -702,7 +704,7 @@ func (d *dagContext) evaluateDependsLogic(taskName string) (bool, bool, error) {
 
 		// If the task is still running, we should not proceed.
 		depNode := d.getTaskNode(taskName)
-		if depNode == nil || !depNode.Completed() {
+		if depNode == nil || !depNode.Fulfilled() {
 			return false, false, nil
 		}
 

workflow/controller/operator.go

@@ -302,7 +302,7 @@ func (woc *wfOperationCtx) operate() {
 		return
 	}
 
-	if node == nil || !node.Completed() {
+	if node == nil || !node.Fulfilled() {
 		// node can be nil if a workflow created immediately in a parallelism == 0 state
 		return
 	}
@@ -347,7 +347,7 @@ func (woc *wfOperationCtx) operate() {
 			woc.log.Errorf("error in exit template execution: %+v", err)
 			return
 		}
-		if onExitNode == nil || !onExitNode.Completed() {
+		if onExitNode == nil || !onExitNode.Fulfilled() {
 			return
 		}
 	}
@@ -622,7 +622,7 @@ func (woc *wfOperationCtx) requeue(afterDuration time.Duration) {
 
 // processNodeRetries updates the retry node state based on the child node state and the retry strategy and returns the node.
 func (woc *wfOperationCtx) processNodeRetries(node *wfv1.NodeStatus, retryStrategy wfv1.RetryStrategy) (*wfv1.NodeStatus, bool, error) {
-	if node.Completed() {
+	if node.Fulfilled() {
 		return node, true, nil
 	}
 	lastChildNode, err := woc.getLastChildNode(node)
@@ -634,7 +634,7 @@ func (woc *wfOperationCtx) processNodeRetries(node *wfv1.NodeStatus, retryStrate
 		return node, true, nil
 	}
 
-	if !lastChildNode.Completed() {
+	if !lastChildNode.Fulfilled() {
 		// last child node is still running.
 		return node, true, nil
 	}
@@ -774,7 +774,7 @@ func (woc *wfOperationCtx) podReconciliation() error {
 				woc.updated = true
 			}
 			node := woc.wf.Status.Nodes[pod.ObjectMeta.Name]
-			if node.Completed() && !node.IsDaemoned() {
+			if node.Fulfilled() && !node.IsDaemoned() {
 				if tmpVal, tmpOk := pod.Labels[common.LabelKeyCompleted]; tmpOk {
 					if tmpVal == "true" {
 						return
@@ -784,7 +784,7 @@ func (woc *wfOperationCtx) podReconciliation() error {
 				if woc.shouldPrintPodSpec(node) {
 					printPodSpecLog(pod, woc.wf.Name)
 				}
-				if !woc.orig.Status.Nodes[node.ID].Completed() {
+				if !woc.orig.Status.Nodes[node.ID].Fulfilled() {
 					woc.onNodeComplete(&node)
 				}
 			}
@@ -818,7 +818,7 @@ func (woc *wfOperationCtx) podReconciliation() error {
 	// It is now impossible to infer pod status. The only thing we can do at this point is to mark
 	// the node with Error.
 	for nodeID, node := range woc.wf.Status.Nodes {
-		if node.Type != wfv1.NodeTypePod || node.Completed() || node.StartedAt.IsZero() {
+		if node.Type != wfv1.NodeTypePod || node.Fulfilled() || node.StartedAt.IsZero() {
 			// node is not a pod, it is already complete, or it can be re-run.
 			continue
 		}
@@ -1059,7 +1059,7 @@ func (woc *wfOperationCtx) assessNodeStatus(pod *apiv1.Pod, node *wfv1.NodeStatu
 		node.Message = message
 	}
 
-	if node.Completed() && node.FinishedAt.IsZero() {
+	if node.Fulfilled() && node.FinishedAt.IsZero() {
 		updated = true
 		if !node.IsDaemoned() {
 			node.FinishedAt = getLatestFinishedAt(pod)
@@ -1402,13 +1402,13 @@ func (woc *wfOperationCtx) executeTemplate(nodeName string, orgTmpl wfv1.Templat
 	}
 
 	if node != nil {
-		if node.Completed() {
+		if node.Fulfilled() {
 			woc.log.Debugf("Node %s already completed", nodeName)
 			if resolvedTmpl.Metrics != nil {
 				// Check if this node completed between executions. If it did, emit metrics. If a node completes within
 				// the same execution, its metrics are emitted below.
 				// We can infer that this node completed during the current operation, emit metrics
-				if prevNodeStatus, ok := woc.preExecutionNodePhases[node.ID]; ok && !prevNodeStatus.Completed() {
+				if prevNodeStatus, ok := woc.preExecutionNodePhases[node.ID]; ok && !prevNodeStatus.Fulfilled() {
 					localScope, realTimeScope := woc.prepareMetricScope(node)
 					woc.computeMetrics(resolvedTmpl.Metrics.Prometheus, localScope, realTimeScope, false)
 				}
@@ -1470,14 +1470,14 @@ func (woc *wfOperationCtx) executeTemplate(nodeName string, orgTmpl wfv1.Templat
 		}
 		retryParentNode = processedRetryParentNode
 		// The retry node might have completed by now.
-		if retryParentNode.Completed() {
+		if retryParentNode.Fulfilled() {
 			return retryParentNode, nil
 		}
 		lastChildNode, err := woc.getLastChildNode(retryParentNode)
 		if err != nil {
 			return woc.markNodeError(retryNodeName, err), err
 		}
-		if lastChildNode != nil && !lastChildNode.Completed() {
+		if lastChildNode != nil && !lastChildNode.Fulfilled() {
 			// Last child node is still running.
 			nodeName = lastChildNode.Name
 			node = lastChildNode
@@ -1547,7 +1547,7 @@ func (woc *wfOperationCtx) executeTemplate(nodeName string, orgTmpl wfv1.Templat
 		// completed before this execution. If it did not exist prior, then we can infer that it was completed during this execution.
 		// The statement "(!ok || prevNodeStatus.Completed())" checks for this behavior and represents the material conditional
 		// "ok -> prevNodeStatus.Completed()" (https://en.wikipedia.org/wiki/Material_conditional)
-		if prevNodeStatus, ok := woc.preExecutionNodePhases[node.ID]; (!ok || prevNodeStatus.Completed()) && node.Completed() {
+		if prevNodeStatus, ok := woc.preExecutionNodePhases[node.ID]; (!ok || prevNodeStatus.Fulfilled()) && node.Fulfilled() {
 			localScope, realTimeScope := woc.prepareMetricScope(node)
 			woc.computeMetrics(resolvedTmpl.Metrics.Prometheus, localScope, realTimeScope, false)
 		}
@@ -1558,7 +1558,7 @@ func (woc *wfOperationCtx) executeTemplate(nodeName string, orgTmpl wfv1.Templat
 	// Swap the node back to retry node
 	if retryNodeName != "" {
 		retryNode := woc.getNodeByName(retryNodeName)
-		if !retryNode.Completed() && node.Completed() { //if the retry child has completed we need to update outself
+		if !retryNode.Fulfilled() && node.Fulfilled() { //if the retry child has completed we need to update outself
 			node, err = woc.executeTemplate(retryNodeName, orgTmpl, tmplCtx, args, opts)
 			if err != nil {
 				return woc.markNodeError(node.Name, err), err
@@ -1714,7 +1714,7 @@ func (woc *wfOperationCtx) initializeNode(nodeName string, nodeType wfv1.NodeTyp
 		node.DisplayName = nodeName
 	}
 
-	if node.Completed() && node.FinishedAt.IsZero() {
+	if node.Fulfilled() && node.FinishedAt.IsZero() {
 		node.FinishedAt = node.StartedAt
 	}
 	var message string
@@ -1746,12 +1746,12 @@ func (woc *wfOperationCtx) markNodePhase(nodeName string, phase wfv1.NodePhase,
 			woc.updated = true
 		}
 	}
-	if node.Completed() && node.FinishedAt.IsZero() {
+	if node.Fulfilled() && node.FinishedAt.IsZero() {
 		node.FinishedAt = metav1.Time{Time: time.Now().UTC()}
 		woc.log.Infof("node %s finished: %s", node.ID, node.FinishedAt)
 		woc.updated = true
 	}
-	if !woc.orig.Status.Nodes[node.ID].Completed() && node.Completed() {
+	if !woc.orig.Status.Nodes[node.ID].Fulfilled() && node.Fulfilled() {
 		woc.onNodeComplete(node)
 	}
 	woc.wf.Status.Nodes[node.ID] = *node