Create a failure node in subworkflow

Signed-off-by: Kevin Su <[email protected]>

flyte-single-binary-local.yaml

@@ -15,7 +15,7 @@ cluster_resources:
 
 logger:
   show-source: true
-  level: 6
+  level: 4
 
 propeller:
   create-flyteworkflow-crd: true
@@ -83,3 +83,8 @@ storage:
       access_key_id: minio
       secret_key: miniostorage
   container: my-s3-bucket
+
+
+flyte:
+  propeller:
+    disableWebhook: true

flytepropeller/pkg/compiler/workflow_compiler.go

@@ -216,9 +216,6 @@ func (w workflowBuilder) ValidateWorkflow(fg *flyteWorkflow, errs errors.Compile
 		}
 	}
 
-	failureNode := fg.Template.FailureNode
-	v.ValidateNode(&wf, wf.GetOrCreateNodeBuilder(failureNode), false /* validateConditionTypes */, errs.NewScope())
-
 	// Add explicitly and implicitly declared edges
 	for nodeID, n := range wf.Nodes {
 		if nodeID == c.StartNodeID {
@@ -228,7 +225,11 @@ func (w workflowBuilder) ValidateWorkflow(fg *flyteWorkflow, errs errors.Compile
 		wf.AddEdges(n, c.EdgeDirectionBidirectional, errs.NewScope())
 	}
 
-	wf.AddEdges(wf.GetOrCreateNodeBuilder(failureNode), c.EdgeDirectionUpstream, errs.NewScope())
+	if fg.Template.FailureNode != nil {
+		failureNode := fg.Template.FailureNode
+		v.ValidateNode(&wf, wf.GetOrCreateNodeBuilder(failureNode), false /* validateConditionTypes */, errs.NewScope())
+		wf.AddEdges(wf.GetOrCreateNodeBuilder(failureNode), c.EdgeDirectionUpstream, errs.NewScope())
+	}
 
 	// Add execution edges for orphan nodes that don't have any inward/outward edges.
 	for nodeID := range wf.Nodes {

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

@@ -49,12 +49,11 @@ func (w *workflowNodeHandler) Handle(ctx context.Context, nCtx interfaces.NodeEx
 			errors.BadSpecificationError, errMsg, nil)), nil
 	}
 
-	updateNodeStateFn := func(transition handler.Transition, newPhase v1alpha1.WorkflowNodePhase, err error) (handler.Transition, error) {
+	updateNodeStateFn := func(transition handler.Transition, workflowNodeState handler.WorkflowNodeState, err error) (handler.Transition, error) {
 		if err != nil {
 			return transition, err
 		}
 
-		workflowNodeState := handler.WorkflowNodeState{Phase: newPhase}
 		err = nCtx.NodeStateWriter().PutWorkflowNodeState(workflowNodeState)
 		if err != nil {
 			logger.Errorf(ctx, "Failed to store WorkflowNodeState, err :%s", err.Error())
@@ -75,10 +74,10 @@ func (w *workflowNodeHandler) Handle(ctx context.Context, nCtx interfaces.NodeEx
 
 		if wfNode.GetSubWorkflowRef() != nil {
 			trns, err := w.subWfHandler.StartSubWorkflow(ctx, nCtx)
-			return updateNodeStateFn(trns, v1alpha1.WorkflowNodePhaseExecuting, err)
+			return updateNodeStateFn(trns, handler.WorkflowNodeState{Phase: v1alpha1.WorkflowNodePhaseExecuting}, err)
 		} else if wfNode.GetLaunchPlanRefID() != nil {
 			trns, err := w.lpHandler.StartLaunchPlan(ctx, nCtx)
-			return updateNodeStateFn(trns, v1alpha1.WorkflowNodePhaseExecuting, err)
+			return updateNodeStateFn(trns, handler.WorkflowNodeState{Phase: v1alpha1.WorkflowNodePhaseExecuting}, err)
 		}
 
 		return invalidWFNodeError()
@@ -95,8 +94,9 @@ func (w *workflowNodeHandler) Handle(ctx context.Context, nCtx interfaces.NodeEx
 		}
 
 		if wfNode.GetSubWorkflowRef() != nil {
+			originalError := nCtx.NodeStateReader().GetWorkflowNodeState().Error
 			trns, err := w.subWfHandler.HandleFailingSubWorkflow(ctx, nCtx)
-			return updateNodeStateFn(trns, workflowPhase, err)
+			return updateNodeStateFn(trns, handler.WorkflowNodeState{Phase: workflowPhase, Error: originalError}, err)
 		} else if wfNode.GetLaunchPlanRefID() != nil {
 			// There is no failure node for launch plans, terminate immediately.
 			return handler.DoTransition(handler.TransitionTypeEphemeral, handler.PhaseInfoFailureErr(wfNodeState.Error, nil)), nil

flytepropeller/pkg/controller/nodes/subworkflow/subworkflow.go

@@ -160,7 +160,7 @@ func (s *subworkflowHandler) HandleFailureNodeOfSubWorkflow(ctx context.Context,
 			return handler.DoTransition(handler.TransitionTypeEphemeral, handler.PhaseInfoUndefined), err
 		}
 
-		if state.NodePhase == interfaces.NodePhaseRunning {
+		if state.NodePhase == interfaces.NodePhaseQueued || state.NodePhase == interfaces.NodePhaseRunning {
 			return handler.DoTransition(handler.TransitionTypeEphemeral, handler.PhaseInfoRunning(nil)), nil
 		}
 
@@ -175,7 +175,7 @@ func (s *subworkflowHandler) HandleFailureNodeOfSubWorkflow(ctx context.Context,
 				return handler.UnknownTransition, err
 			}
 
-			return handler.DoTransition(handler.TransitionTypeEphemeral, handler.PhaseInfoFailingErr(originalError, nil)), nil
+			return handler.DoTransition(handler.TransitionTypeEphemeral, handler.PhaseInfoRunning(nil)), nil
 		}
 
 		// When handling the failure node succeeds, the final status will still be failure

flytepropeller/pkg/controller/nodes/transformers.go

@@ -216,6 +216,8 @@ func ToNodePhase(p handler.EPhase) (v1alpha1.NodePhase, error) {
 		return v1alpha1.NodePhaseSucceeding, nil
 	case handler.EPhaseFailed:
 		return v1alpha1.NodePhaseFailing, nil
+	case handler.EPhaseFailing:
+		return v1alpha1.NodePhaseFailing, nil
 	case handler.EPhaseTimedout:
 		return v1alpha1.NodePhaseTimingOut, nil
 	case handler.EPhaseRecovered:

flytepropeller/pkg/controller/workflow/executor.go

@@ -145,6 +145,7 @@ func (c *workflowExecutor) handleRunningWorkflow(ctx context.Context, w *v1alpha
 	}
 	execcontext := executors.NewExecutionContext(w, w, w, nil, executors.InitializeControlFlow())
 	state, err := c.nodeExecutor.RecursiveNodeHandler(ctx, execcontext, w, w, startNode)
+
 	if err != nil {
 		return StatusRunning, err
 	}
@@ -293,7 +294,6 @@ func (c *workflowExecutor) TransitionToPhase(ctx context.Context, execID *core.W
 			wStatus.UpdatePhase(v1alpha1.WorkflowPhaseFailing, "", wfEvent.GetError())
 			wfEvent.OccurredAt = utils.GetProtoTime(nil)
 		case v1alpha1.WorkflowPhaseHandlingFailureNode:
-			// TODO: Add core.WorkflowPhaseHandlingFailureNode to idl?
 			wfEvent.Phase = core.WorkflowExecution_FAILING
 			wfEvent.OutputResult = convertToExecutionError(toStatus.Err, previousError)
 			wStatus.UpdatePhase(v1alpha1.WorkflowPhaseHandlingFailureNode, "", wfEvent.GetError())