nodes

Documentation

Index

• Constants
• func CreateAliasMap(aliases []v1alpha1.Alias) map[string]string
• func GetParentNodeMaxEndTime(ctx context.Context, dag executors.DAGStructure, nl executors.NodeLookup, ...) (t v1.Time, err error)
• func NewExecutor(ctx context.Context, nodeConfig config.NodeConfig, store *storage.DataStore, ...) (executors.Node, error)
• func ParseVarName(varName string) (idx *int, name string, err error)
• func Resolve(ctx context.Context, outputResolver OutputResolver, nl executors.NodeLookup, ...) (*core.LiteralMap, error)
• func ResolveBindingData(ctx context.Context, outputResolver OutputResolver, nl executors.NodeLookup, ...) (*core.Literal, error)
• func ToK8sTime(t time.Time) v1.Time
• func ToNodeExecEventPhase(p handler.EPhase) core.NodeExecution_Phase
• func ToNodeExecOutput(info *handler.OutputInfo) *event.NodeExecutionEvent_OutputUri
• func ToNodeExecTaskNodeMetadata(info *handler.TaskNodeInfo) *event.NodeExecutionEvent_TaskNodeMetadata
• func ToNodeExecWorkflowNodeMetadata(info *handler.WorkflowNodeInfo) *event.NodeExecutionEvent_WorkflowNodeMetadata
• func ToNodeExecutionEvent(nodeExecID *core.NodeExecutionIdentifier, info handler.PhaseInfo, ...) (*event.NodeExecutionEvent, error)
• func ToNodePhase(p handler.EPhase) (v1alpha1.NodePhase, error)
• func UpdateNodeStatus(np v1alpha1.NodePhase, p handler.PhaseInfo, n *nodeStateManager, ...)
• type HandlerFactory
  • func NewHandlerFactory(ctx context.Context, executor executors.Node, ...) (HandlerFactory, error)
• type OutputResolver
  • func NewRemoteFileOutputResolver(store *storage.DataStore) OutputResolver
• type PredicatePhase
  • func CanExecute(ctx context.Context, dag executors.DAGStructure, nl executors.NodeLookup, ...) (PredicatePhase, error)
  • func (p PredicatePhase) String() string
• type VarName

Constants

const NodeIDLabel = "node-id"

const NodeInterruptibleLabel = "interruptible"

const TaskNameLabel = "task-name"

Functions

func CreateAliasMap

func CreateAliasMap(aliases []v1alpha1.Alias) map[string]string

func GetParentNodeMaxEndTime

func GetParentNodeMaxEndTime(ctx context.Context, dag executors.DAGStructure, nl executors.NodeLookup, node v1alpha1.BaseNode) (t v1.Time, err error)

func NewExecutor

func NewExecutor(ctx context.Context, nodeConfig config.NodeConfig, store *storage.DataStore, enQWorkflow v1alpha1.EnqueueWorkflow, eventSink events.EventSink,
	workflowLauncher launchplan.Executor, launchPlanReader launchplan.Reader, maxDatasetSize int64,
	defaultRawOutputPrefix storage.DataReference, kubeClient executors.Client,
	catalogClient catalog.Client, scope promutils.Scope) (executors.Node, error)

func ParseVarName

func ParseVarName(varName string) (idx *int, name string, err error)

Parses var names

func Resolve

func Resolve(ctx context.Context, outputResolver OutputResolver, nl executors.NodeLookup, nodeID v1alpha1.NodeID, bindings []*v1alpha1.Binding) (*core.LiteralMap, error)

func ResolveBindingData

func ResolveBindingData(ctx context.Context, outputResolver OutputResolver, nl executors.NodeLookup, bindingData *core.BindingData) (*core.Literal, error)

func ToK8sTime

func ToK8sTime(t time.Time) v1.Time

func ToNodeExecEventPhase

func ToNodeExecEventPhase(p handler.EPhase) core.NodeExecution_Phase

func ToNodeExecOutput

func ToNodeExecOutput(info *handler.OutputInfo) *event.NodeExecutionEvent_OutputUri

func ToNodeExecTaskNodeMetadata

func ToNodeExecTaskNodeMetadata(info *handler.TaskNodeInfo) *event.NodeExecutionEvent_TaskNodeMetadata

func ToNodeExecWorkflowNodeMetadata

func ToNodeExecWorkflowNodeMetadata(info *handler.WorkflowNodeInfo) *event.NodeExecutionEvent_WorkflowNodeMetadata

func ToNodeExecutionEvent

func ToNodeExecutionEvent(nodeExecID *core.NodeExecutionIdentifier,
	info handler.PhaseInfo,
	reader io.InputReader,
	status v1alpha1.ExecutableNodeStatus,
	eventVersion v1alpha1.EventVersion,
	parentInfo executors.ImmutableParentInfo,
	node v1alpha1.ExecutableNode) (*event.NodeExecutionEvent, error)

func ToNodePhase

func ToNodePhase(p handler.EPhase) (v1alpha1.NodePhase, error)

func UpdateNodeStatus

func UpdateNodeStatus(np v1alpha1.NodePhase, p handler.PhaseInfo, n *nodeStateManager, s v1alpha1.ExecutableNodeStatus)

Types

type HandlerFactory

type HandlerFactory interface {
	GetHandler(kind v1alpha1.NodeKind) (handler.Node, error)
	Setup(ctx context.Context, setup handler.SetupContext) error
}

func NewHandlerFactory

func NewHandlerFactory(ctx context.Context, executor executors.Node, workflowLauncher launchplan.Executor,
	launchPlanReader launchplan.Reader, kubeClient executors.Client, client catalog.Client, scope promutils.Scope) (HandlerFactory, error)

type OutputResolver

type OutputResolver interface {
	// Extracts a subset of node outputs to literals.
	ExtractOutput(ctx context.Context, nl executors.NodeLookup, n v1alpha1.ExecutableNode,
		bindToVar VarName) (values *core.Literal, err error)
}

func NewRemoteFileOutputResolver

func NewRemoteFileOutputResolver(store *storage.DataStore) OutputResolver

Creates a simple output resolver that expects an outputs.pb at the data directory of the node.

type PredicatePhase

type PredicatePhase int

Special enum to indicate if the node under consideration is ready to be executed or should be skipped

const (
	// Indicates node is not yet ready to be executed
	PredicatePhaseNotReady PredicatePhase = iota
	// Indicates node is ready to be executed - execution should proceed
	PredicatePhaseReady
	// Indicates that the node execution should be skipped as one of its parents was skipped or the branch was not taken
	PredicatePhaseSkip
	// Indicates failure during Predicate check
	PredicatePhaseUndefined
)

func CanExecute

func CanExecute(ctx context.Context, dag executors.DAGStructure, nl executors.NodeLookup, node v1alpha1.BaseNode) (PredicatePhase, error)

func (PredicatePhase) String

func (p PredicatePhase) String() string

type VarName

type VarName = string