executors

Documentation

Index

• Variables
• func NewBaseContextualWorkflow(baseWorkflow v1alpha1.ExecutableWorkflow) v1alpha1.ExecutableWorkflow
• func NewFallbackClient(cachedClient, rawClient client.Client) client.Client
• func NewSubContextualWorkflow(baseWorkflow v1alpha1.ExecutableWorkflow, subWF v1alpha1.ExecutableSubWorkflow, ...) v1alpha1.ExecutableWorkflow
• type Client
• type ContextualWorkflow
• type Node
• type NodePhase
  • func (p NodePhase) String() string
• type NodeStatus
  • func NodeStatusFailed(err error) NodeStatus
  • func (n *NodeStatus) HasFailed() bool
  • func (n *NodeStatus) HasTimedOut() bool
  • func (n *NodeStatus) IsComplete() bool
  • func (n *NodeStatus) PartiallyComplete() bool
• type Workflow

Variables

var NodeStatusComplete = NodeStatus{NodePhase: NodePhaseComplete}

var NodeStatusPending = NodeStatus{NodePhase: NodePhasePending}

var NodeStatusQueued = NodeStatus{NodePhase: NodePhaseQueued}

var NodeStatusRunning = NodeStatus{NodePhase: NodePhaseRunning}

var NodeStatusSuccess = NodeStatus{NodePhase: NodePhaseSuccess}

var NodeStatusTimedOut = NodeStatus{NodePhase: NodePhaseTimedOut}

var NodeStatusUndefined = NodeStatus{NodePhase: NodePhaseUndefined}

Functions

func NewBaseContextualWorkflow

func NewBaseContextualWorkflow(baseWorkflow v1alpha1.ExecutableWorkflow) v1alpha1.ExecutableWorkflow

func NewFallbackClient

func NewFallbackClient(cachedClient, rawClient client.Client) client.Client

Creates a new k8s client that uses the cached client for reads and falls back to making API calls if it failed. Write calls will always go to raw client directly.

func NewSubContextualWorkflow

func NewSubContextualWorkflow(baseWorkflow v1alpha1.ExecutableWorkflow, subWF v1alpha1.ExecutableSubWorkflow,
	nodeStatus v1alpha1.ExecutableNodeStatus) v1alpha1.ExecutableWorkflow

Creates a contextual workflow using the provided interface implementations.

Types

type Client

type Client interface {
	// GetClient returns a client configured with the Config
	GetClient() client.Client

	// GetCache returns a cache.Cache
	GetCache() cache.Cache
}

A friendly controller-runtime client that gets passed to executors

type ContextualWorkflow

type ContextualWorkflow struct {
	v1alpha1.WorkflowMetaExtended
	v1alpha1.ExecutableSubWorkflow
	v1alpha1.NodeStatusGetter
}

type Node

type Node interface {
	// This method is used specifically to set inputs for start node. This is because start node does not retrieve inputs
	// from predecessors, but the inputs are inputs to the workflow or inputs to the parent container (workflow) node.
	SetInputsForStartNode(ctx context.Context, w v1alpha1.ExecutableWorkflow, inputs *core.LiteralMap) (NodeStatus, error)

	// This is the main entrypoint to execute a node. It recursively depth-first goes through all ready nodes and starts their execution
	// This returns either
	// - 1. It finds a blocking node (not ready, or running)
	// - 2. A node fails and hence the workflow will fail
	// - 3. The final/end node has completed and the workflow should be stopped
	RecursiveNodeHandler(ctx context.Context, w v1alpha1.ExecutableWorkflow, currentNode v1alpha1.ExecutableNode) (NodeStatus, error)

	// This aborts the given node. If the given node is complete then it recursively finds the running nodes and aborts them
	AbortHandler(ctx context.Context, w v1alpha1.ExecutableWorkflow, currentNode v1alpha1.ExecutableNode, reason string) error

	FinalizeHandler(ctx context.Context, w v1alpha1.ExecutableWorkflow, currentNode v1alpha1.ExecutableNode) error

	// This method should be used to initialize Node executor
	Initialize(ctx context.Context) error
}

Core Node Executor that is used to execute a node. This is a recursive node executor and understands node dependencies

type NodePhase

type NodePhase int

p of the node

const (
	// Indicates that the node is not yet ready to be executed and is pending any previous nodes completion
	NodePhasePending NodePhase = iota
	// Indicates that the node was queued and will start running soon
	NodePhaseQueued
	// Indicates that the payload associated with this node is being executed and is not yet done
	NodePhaseRunning
	// Indicates that the nodes payload has been successfully completed, but any downstream nodes from this node may not yet have completed
	// We could make Success = running, but this enables more granular control
	NodePhaseSuccess
	// Complete indicates successful completion of a node. For singular nodes (nodes that have only one execution) success = complete, but, the executor
	// will always signal completion
	NodePhaseComplete
	// Node failed in execution, either this node or anything in the downstream chain
	NodePhaseFailed
	// Internal error observed. This state should always be accompanied with an `error`. if not the behavior is undefined
	NodePhaseUndefined
	// Finalize node failing due to timeout
	NodePhaseTimingOut
	// Node failed because execution timed out
	NodePhaseTimedOut
)

func (NodePhase) String

func (p NodePhase) String() string

type NodeStatus

type NodeStatus struct {
	NodePhase NodePhase
	Err       error
}

Helper struct to allow passing of status between functions

func NodeStatusFailed

func NodeStatusFailed(err error) NodeStatus

func (*NodeStatus) HasFailed

func (n *NodeStatus) HasFailed() bool

func (*NodeStatus) HasTimedOut

func (n *NodeStatus) HasTimedOut() bool

func (*NodeStatus) IsComplete

func (n *NodeStatus) IsComplete() bool

func (*NodeStatus) PartiallyComplete

func (n *NodeStatus) PartiallyComplete() bool

type Workflow

type Workflow interface {
	Initialize(ctx context.Context) error
	HandleFlyteWorkflow(ctx context.Context, w *v1alpha1.FlyteWorkflow) error
	HandleAbortedWorkflow(ctx context.Context, w *v1alpha1.FlyteWorkflow, maxRetries uint32) error
}