wip

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

flytepropeller/pkg/compiler/transformers/k8s/workflow.go

@@ -87,6 +87,16 @@ func buildFlyteWorkflowSpec(wf *core.CompiledWorkflow, tasks []*core.CompiledTas
 
 	var failureN *v1alpha1.NodeSpec
 	if n := wf.Template.GetFailureNode(); n != nil {
+		for _, in := range n.Inputs {
+			in.Binding = &core.BindingData{
+				Value: &core.BindingData_Promise{
+					Promise: &core.OutputReference{
+						Var:    in.Binding.GetPromise().GetVar(),
+						NodeId: v1alpha1.StartNodeID,
+					},
+				},
+			}
+		}
 		nodes, ok := buildNodeSpec(n, tasks, errs.NewScope())
 		if !ok {
 			return nil, errs

flytepropeller/pkg/controller/executors/failure_node_lookup.go

@@ -0,0 +1,34 @@
+package executors
+
+import (
+	"context"
+	"github.com/flyteorg/flyte/flytepropeller/pkg/apis/flyteworkflow/v1alpha1"
+)
+
+type FailureNodeLookup struct {
+	NodeSpec   *v1alpha1.NodeSpec
+	NodeStatus v1alpha1.ExecutableNodeStatus
+}
+
+func (f FailureNodeLookup) GetNode(nodeID v1alpha1.NodeID) (v1alpha1.ExecutableNode, bool) {
+	return f.NodeSpec, true
+}
+
+func (f FailureNodeLookup) GetNodeExecutionStatus(ctx context.Context, id v1alpha1.NodeID) v1alpha1.ExecutableNodeStatus {
+	return f.NodeStatus
+}
+
+func (f FailureNodeLookup) ToNode(id v1alpha1.NodeID) ([]v1alpha1.NodeID, error) {
+	return []v1alpha1.NodeID{v1alpha1.StartNodeID}, nil
+}
+
+func (f FailureNodeLookup) FromNode(id v1alpha1.NodeID) ([]v1alpha1.NodeID, error) {
+	return nil, nil
+}
+
+func NewFailureNodeLookup(nodeSpec *v1alpha1.NodeSpec, nodeStatus v1alpha1.ExecutableNodeStatus) NodeLookup {
+	return FailureNodeLookup{
+		NodeSpec:   nodeSpec,
+		NodeStatus: nodeStatus,
+	}
+}