Clear past errors from workflow state

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

@@ -346,6 +346,8 @@ type MutableNodeStatus interface {
 	GetArrayNodeStatus() MutableArrayNodeStatus
 	GetOrCreateArrayNodeStatus() MutableArrayNodeStatus
 	ClearArrayNodeStatus()
+
+	ClearExecutionError()
 }
 
 type ExecutionTimeInfo interface {

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

@@ -477,6 +477,10 @@
 	in.SetDirty()
 }
 
+func (in *NodeStatus) ClearExecutionError() {
+	in.Error = nil
+}
+
 func (in *NodeStatus) GetLastUpdatedAt() *metav1.Time {
 	return in.LastUpdatedAt
 }
@@ -632,6 +636,7 @@
 		if p == NodePhaseSucceeded || p == NodePhaseSkipped || !enableCRDebugMetadata {
 			// Clear most status related fields after reaching a terminal state. This keeps the CR state small to avoid
 			// etcd size limits. Importantly we keep Phase, StoppedAt and Error which will be needed further.
+			// Errors will still be needed but it will be cleaned up when possible because they can be very large. 
 			in.Message = ""
 			in.QueuedAt = nil
 			in.StartedAt = nil

flytepropeller/pkg/controller/config/config.go

@@ -214,7 +214,7 @@ type NodeConfig struct {
 	InterruptibleFailureThreshold  int32            `json:"interruptible-failure-threshold" pflag:"1,number of failures for a node to be still considered interruptible. Negative numbers are treated as complementary (ex. -1 means last attempt is non-interruptible).'"`
 	DefaultMaxAttempts             int32            `json:"default-max-attempts" pflag:"3,Default maximum number of attempts for a node"`
 	IgnoreRetryCause               bool             `json:"ignore-retry-cause" pflag:",Ignore retry cause and count all attempts toward a node's max attempts"`
-	EnableCRDebugMetadata          bool             `json:"enable-cr-debug-metadata" pflag:",Collapse node on any terminal state, not just successful terminations. This is useful to reduce the size of workflow state in etcd."`
+	EnableCRDebugMetadata          bool             `json:"enable-cr-debug-metadata" pflag:",By default node state gets cleared after flytepropeller will no longer need it. This is useful to reduce the size of workflow state in etcd. Consider enabling this to keep this state for debugging purposes."`
 }
 
 // DefaultDeadlines contains default values for timeouts

flytepropeller/pkg/controller/nodes/executor.go

@@ -277,6 +277,7 @@
 	partialNodeCompletion := false
 	onFailurePolicy := execContext.GetOnFailurePolicy()
 	stateOnComplete := interfaces.NodeStatusComplete
+	var executableNodeStatusOnComplete v1alpha1.ExecutableNodeStatus
 	for _, downstreamNodeName := range downstreamNodes {
 		downstreamNode, ok := nl.GetNode(downstreamNodeName)
 		if !ok {
@@ -298,6 +299,10 @@
 				// If the failure policy allows other nodes to continue running, do not exit the loop,
 				// Keep track of the last failed state in the loop since it'll be the one to return.
 				// TODO: If multiple nodes fail (which this mode allows), consolidate/summarize failure states in one.
+				if executableNodeStatusOnComplete != nil {
+					c.nodeExecutor.Clear(executableNodeStatusOnComplete)
+				}
+				executableNodeStatusOnComplete = nl.GetNodeExecutionStatus(ctx, downstreamNode.GetID())
 				stateOnComplete = state
 			} else {
 				return state, nil
@@ -863,6 +868,12 @@
 	return phase, nil
 }
 
+func (c *nodeExecutor) Clear(executableNodeStatus v1alpha1.ExecutableNodeStatus) {
+	if !c.enableCRDebugMetadata {
+		executableNodeStatus.ClearExecutionError()
+	}
+}
+
 func (c *nodeExecutor) Abort(ctx context.Context, h interfaces.NodeHandler, nCtx interfaces.NodeExecutionContext, reason string, finalTransition bool) error {
 	logger.Debugf(ctx, "Calling aborting & finalize")
 	if err := h.Abort(ctx, nCtx, reason); err != nil {

flytepropeller/pkg/controller/nodes/interfaces/handler.go

@@ -7,6 +7,7 @@ import (
 	"github.com/flyteorg/flyte/flytepropeller/pkg/controller/executors"
 	"github.com/flyteorg/flyte/flytepropeller/pkg/controller/nodes/handler"
 	"github.com/flyteorg/flyte/flytestdlib/promutils"
+	"github.com/flyteorg/flyte/flytepropeller/pkg/apis/flyteworkflow/v1alpha1"
 )
 
 //go:generate mockery -all -case=underscore
@@ -16,6 +17,7 @@ type NodeExecutor interface {
 	HandleNode(ctx context.Context, dag executors.DAGStructure, nCtx NodeExecutionContext, h NodeHandler) (NodeStatus, error)
 	Abort(ctx context.Context, h NodeHandler, nCtx NodeExecutionContext, reason string, finalTransition bool) error
 	Finalize(ctx context.Context, h NodeHandler, nCtx NodeExecutionContext) error
+	Clear(executableNodeStatus v1alpha1.ExecutableNodeStatus)
 }
 
 // Interface that should be implemented for a node type.

flytepropeller/pkg/controller/workflow/executor_test.go

@@ -72,6 +72,10 @@ type fakeRemoteWritePlugin struct {
 	t             assert.TestingT
 }
 
+type fakeNodeExecContext interface {
+	Node()                v1alpha1.ExecutableNode
+}
+
 func (f fakeRemoteWritePlugin) Handle(ctx context.Context, tCtx pluginCore.TaskExecutionContext) (pluginCore.Transition, error) {
 	logger.Infof(ctx, "----------------------------------------------------------------------------------------------")
 	logger.Infof(ctx, "Handle called for %s", tCtx.TaskExecutionMetadata().GetTaskExecutionID().GetGeneratedName())
@@ -224,6 +228,26 @@ func createTaskExecutorErrorInCheck(t assert.TestingT) pluginCore.PluginEntry {
 	}
 }
 
+func CountFailedNodes(nodeStatuses map[v1alpha1.NodeID]*v1alpha1.NodeStatus) int {
+	count := 0
+	for _, v := range nodeStatuses {
+		if v.Phase == v1alpha1.NodePhaseFailed {
+			count++
+		}
+	}
+	return count
+}
+
+func CountNodesWithErrors(nodeStatuses map[v1alpha1.NodeID]*v1alpha1.NodeStatus) int {
+	count := 0
+	for _, v := range nodeStatuses {
+		if v.Error != nil {
+			count++
+		}
+	}
+	return count
+}
+
 func TestWorkflowExecutor_HandleFlyteWorkflow_Error(t *testing.T) {
 	ctx := context.Background()
 	scope := testScope.NewSubScope("12")
@@ -496,54 +520,84 @@ func TestWorkflowExecutor_HandleFlyteWorkflow_Failing(t *testing.T) {
 
 	h := &nodemocks.NodeHandler{}
 	h.OnAbortMatch(mock.Anything, mock.Anything, mock.Anything).Return(nil)
-	h.OnHandleMatch(mock.Anything, mock.Anything).Return(handler.DoTransition(handler.TransitionTypeEphemeral, handler.PhaseInfoSuccess(nil)), nil)
+
+	// Mock handler marks start-node successfully completed but other nodes as failed
+	startNodeMatcher := mock.MatchedBy(func(nodeExecContext fakeNodeExecContext) bool {
+		return nodeExecContext.Node().IsStartNode()
+	})
+	h.OnHandleMatch(mock.Anything, startNodeMatcher).Return(handler.DoTransition(handler.TransitionTypeEphemeral, handler.PhaseInfoSuccess(nil)), nil)
+	h.OnHandleMatch(mock.Anything, mock.Anything).Return(handler.DoTransition(
+		handler.TransitionTypeEphemeral,
+		handler.PhaseInfoFailureErr(&core.ExecutionError{Code: "code", Message: "message", ErrorUri: "uri"}, nil)), nil,
+	)
+
 	h.OnFinalizeMatch(mock.Anything, mock.Anything).Return(nil)
 	h.OnFinalizeRequired().Return(false)
 
 	handlerFactory := &nodemocks.HandlerFactory{}
 	handlerFactory.OnSetupMatch(mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	handlerFactory.OnGetHandlerMatch(mock.Anything).Return(h, nil)
 
-	nodeExec, err := nodes.NewExecutor(ctx, config.GetConfig().NodeConfig, store, enqueueWorkflow, eventSink, adminClient, adminClient,
-		maxOutputSize, "s3://bucket", fakeKubeClient, catalogClient, recoveryClient, eventConfig, testClusterID, signalClient, handlerFactory, promutils.NewTestScope())
-	assert.NoError(t, err)
-	executor, err := NewExecutor(ctx, store, enqueueWorkflow, eventSink, recorder, "", nodeExec, eventConfig, testClusterID, promutils.NewTestScope())
-	assert.NoError(t, err)
-
-	assert.NoError(t, executor.Initialize(ctx))
+	tests := []struct {
+		name                         string
+		onFailurePolicy              v1alpha1.WorkflowOnFailurePolicy
+		enableCRDebugMetadata        bool
+		expectedRoundsToFail         int
+		expectedNodesWithErrorsCount int
+		expectedFailedNodesCount     int
+	}{
+		{"failImidiately", v1alpha1.WorkflowOnFailurePolicy(core.WorkflowMetadata_FAIL_IMMEDIATELY), false, 6, 1, 1},
+		{"failImidiately enableCRDebugMetadata", v1alpha1.WorkflowOnFailurePolicy(core.WorkflowMetadata_FAIL_IMMEDIATELY), true, 6, 1, 1},
+		{"failAfterExecutableNodesComplete", v1alpha1.WorkflowOnFailurePolicy(core.WorkflowMetadata_FAIL_AFTER_EXECUTABLE_NODES_COMPLETE), false, 12, 1, 2},
+		{"failAfterExecutableNodesComplete enableCRDebugMetadata", v1alpha1.WorkflowOnFailurePolicy(core.WorkflowMetadata_FAIL_AFTER_EXECUTABLE_NODES_COMPLETE), true, 12, 2, 2},
+	}
 
 	wJSON, err := yamlutils.ReadYamlFileAsJSON("testdata/benchmark_wf.yaml")
-	if assert.NoError(t, err) {
-		w := &v1alpha1.FlyteWorkflow{
-			RawOutputDataConfig: v1alpha1.RawOutputDataConfig{RawOutputDataConfig: &admin.RawOutputDataConfig{}},
-		}
-		if assert.NoError(t, json.Unmarshal(wJSON, w)) {
-			// For benchmark workflow, we will run into the first failure on round 6
-
-			roundsToFail := 8
-			for i := 0; i < roundsToFail; i++ {
-				err := executor.HandleFlyteWorkflow(ctx, w)
-				assert.Nil(t, err, "Round [%v]", i)
-				fmt.Printf("Round[%d] Workflow[%v]\n", i, w.Status.Phase.String())
-				walkAndPrint(w.Connections, w.Status.NodeStatus)
-				for _, v := range w.Status.NodeStatus {
-					// Reset dirty manually for tests.
-					v.ResetDirty()
-				}
-				fmt.Printf("\n")
-
-				if i == roundsToFail-1 {
-					assert.Equal(t, v1alpha1.WorkflowPhaseFailed, w.Status.Phase)
-				} else if i == roundsToFail-2 {
-					assert.Equal(t, v1alpha1.WorkflowPhaseHandlingFailureNode, w.Status.Phase)
-				} else {
-					assert.NotEqual(t, v1alpha1.WorkflowPhaseFailed, w.Status.Phase, "For Round [%v] got phase [%v]", i, w.Status.Phase.String())
+	assert.NoError(t, err)
+	for _, test := range tests {
+
+		t.Run(test.name, func(t *testing.T) {
+			nodeConfig := config.GetConfig().NodeConfig
+			nodeConfig.EnableCRDebugMetadata = test.enableCRDebugMetadata
+			nodeExec, err := nodes.NewExecutor(ctx, nodeConfig, store, enqueueWorkflow, eventSink, adminClient, adminClient,
+				maxOutputSize, "s3://bucket", fakeKubeClient, catalogClient, recoveryClient, eventConfig, testClusterID, signalClient, handlerFactory, promutils.NewTestScope())
+			assert.NoError(t, err)
+			executor, err := NewExecutor(ctx, store, enqueueWorkflow, eventSink, recorder, "", nodeExec, eventConfig, testClusterID, promutils.NewTestScope())
+			assert.NoError(t, err)
+			assert.NoError(t, executor.Initialize(ctx))
+
+			w := &v1alpha1.FlyteWorkflow{
+				RawOutputDataConfig: v1alpha1.RawOutputDataConfig{RawOutputDataConfig: &admin.RawOutputDataConfig{}},
+				WorkflowSpec:        &v1alpha1.WorkflowSpec{OnFailurePolicy: test.onFailurePolicy},
+			}
+			if assert.NoError(t, json.Unmarshal(wJSON, w)) {
+
+				for i := 0; i < test.expectedRoundsToFail; i++ {
+					t.Run(fmt.Sprintf("Round[%d]", i), func(t *testing.T) {
+						err := executor.HandleFlyteWorkflow(ctx, w)
+						assert.Nil(t, err, "Round [%v]", i)
+						fmt.Printf("Round[%d] Workflow[%v]\n", i, w.Status.Phase.String())
+						walkAndPrint(w.Connections, w.Status.NodeStatus)
+						for _, v := range w.Status.NodeStatus {
+							// Reset dirty manually for tests.
+							v.ResetDirty()
+						}
+						fmt.Printf("\n")
+
+						if i == test.expectedRoundsToFail-1 {
+							assert.Equal(t, v1alpha1.WorkflowPhaseFailed, w.Status.Phase)
+						} else if i == test.expectedRoundsToFail-2 {
+							assert.Equal(t, v1alpha1.WorkflowPhaseHandlingFailureNode, w.Status.Phase)
+						} else {
+							assert.NotEqual(t, v1alpha1.WorkflowPhaseFailed, w.Status.Phase, "For Round [%v] got phase [%v]", i, w.Status.Phase.String())
+						}
+					})
 				}
-
+				assert.Equal(t, test.expectedFailedNodesCount, CountFailedNodes(w.Status.NodeStatus))
+				assert.Equal(t, test.expectedNodesWithErrorsCount, CountNodesWithErrors(w.Status.NodeStatus))
+				assert.Equal(t, v1alpha1.WorkflowPhaseFailed.String(), w.Status.Phase.String(), "Message: [%v]", w.Status.Message)
 			}
-
-			assert.Equal(t, v1alpha1.WorkflowPhaseFailed.String(), w.Status.Phase.String(), "Message: [%v]", w.Status.Message)
-		}
+		})
 	}
 	assert.True(t, recordedRunning)
 	assert.True(t, recordedFailing)