Tidy

flytepropeller/pkg/apis/flyteworkflow/v1alpha1/iface.go

@@ -369,6 +369,7 @@ type ExecutableNodeStatus interface {
 	GetOutputDir() DataReference
 	GetMessage() string
 	GetExecutionError() *core.ExecutionError
+	PopExecutionError() *core.ExecutionError
 	GetAttempts() uint32
 	GetSystemFailures() uint32
 	GetWorkflowNodeStatus() ExecutableWorkflowNodeStatus

flytepropeller/pkg/apis/flyteworkflow/v1alpha1/node_status.go

@@ -477,10 +477,10 @@ func (in *NodeStatus) ClearArrayNodeStatus() {
 	in.SetDirty()
 }
 
-func (in *NodeStatus) ClearErrorMessage() {
-	if in.Error != nil {
-		in.Error.ClearMessage()
-	}
+func (in *NodeStatus) PopExecutionError() *core.ExecutionError {
+	executionError := in.GetExecutionError()
+	in.Error = nil
+	return executionError
 }
 
 func (in *NodeStatus) GetLastUpdatedAt() *metav1.Time {
@@ -640,7 +640,6 @@ func (in *NodeStatus) UpdatePhase(p NodePhase, occurredAt metav1.Time, reason st
 		// Clear most fields after reaching a terminal state. This keeps the CRD state small and avoid etcd size
 		// limits. We keep phase and StoppedAt. StoppedAt is used to calculate transition latency between this
 		// node and any downstream nodes and Phase is required for propeller to continue to downstream nodes.
-		in.ClearErrorMessage()
 		in.QueuedAt = nil
 		in.StartedAt = nil
 		in.LastUpdatedAt = nil

flytepropeller/pkg/controller/nodes/executor.go

@@ -235,13 +235,7 @@ func (c *recursiveNodeExecutor) RecursiveNodeHandler(ctx context.Context, execCo
 		if err != nil {
 			return interfaces.NodeStatusUndefined, err
 		}
-		nodeError := *nodeStatus.GetExecutionError()
-		fmt.Println("before modifying", nodeError)
-		status := interfaces.NodeStatusFailed(&nodeError)
-		nodeStatus.ClearErrorMessage()
-		fmt.Println("modified error", nodeStatus.GetExecutionError())
-		fmt.Println("after modifying", nodeError)
-		return status, nil
+		return interfaces.NodeStatusFailed(nodeStatus.PopExecutionError()), nil
 	} else if nodePhase == v1alpha1.NodePhaseTimedOut {
 		logger.Debugf(currentNodeCtx, "Node has timed out, traversing downstream.")
 		_, err := c.handleDownstream(ctx, execContext, dag, nl, currentNode)

flytepropeller/pkg/controller/nodes/executor_test.go

@@ -619,6 +619,7 @@ func TestNodeExecutor_RecursiveNodeHandler_Recurse(t *testing.T) {
 			mockN0Status.OnGetPhase().Return(n0Phase)
 			mockN0Status.OnGetAttempts().Return(uint32(0))
 			mockN0Status.OnGetExecutionError().Return(nil)
+			mockN0Status.OnPopExecutionError().Return(&core.ExecutionError{Code: "code", Message: "message"})
 
 			mockN0Status.OnIsDirty().Return(false)
 			mockN0Status.OnGetParentTaskID().Return(nil)
@@ -736,6 +737,10 @@ func TestNodeExecutor_RecursiveNodeHandler_Recurse(t *testing.T) {
 				return handler.UnknownTransition, fmt.Errorf("error")
 			}, false, false, false, core.NodeExecution_FAILED},
 
+			{"failing->failed", v1alpha1.NodePhaseFailed, v1alpha1.NodePhaseFailed, interfaces.NodePhaseFailed, func() (handler.Transition, error) {
+				return handler.UnknownTransition, fmt.Errorf("error")
+			}, false, false, false, core.NodeExecution_FAILED},
+
 			{"failing->failed(error)", v1alpha1.NodePhaseFailing, v1alpha1.NodePhaseFailing, interfaces.NodePhaseUndefined, func() (handler.Transition, error) {
 				return handler.UnknownTransition, fmt.Errorf("error")
 			}, true, true, false, core.NodeExecution_FAILING},