executor

Documentation

Index

• Variables
• func AllEnvs(ctx context.Context) []string
• func Register(name string, register Creator)
• func SetupCommand(cmd *exec.Cmd)
• func WithEnv(ctx context.Context, e Env) context.Context
• type ChildDAGExecutor
  • func NewChildDAGExecutor(ctx context.Context, childName string) (*ChildDAGExecutor, error)
  • func (e *ChildDAGExecutor) BuildCoordinatorTask(ctx context.Context, runParams RunParams) (*coordinatorv1.Task, error)
  • func (e *ChildDAGExecutor) Cleanup(ctx context.Context) error
  • func (e *ChildDAGExecutor) ExecuteWithResult(ctx context.Context, runParams RunParams, workDir string) (*digraph.RunStatus, error)
  • func (e *ChildDAGExecutor) Kill(sig os.Signal) error
  • func (e *ChildDAGExecutor) ShouldUseDistributedExecution() bool
• type Creator
• type DAGExecutor
• type Env
  • func GetEnv(ctx context.Context) Env
  • func NewEnv(ctx context.Context, step digraph.Step) Env
  • func (e Env) AllEnvs() []string
  • func (e Env) DAGRunRef() digraph.DAGRunRef
  • func (e Env) EvalBool(ctx context.Context, value any) (bool, error)
  • func (e Env) EvalString(ctx context.Context, s string, opts ...cmdutil.EvalOption) (string, error)
  • func (e Env) ForceLoadOutputVariables(vars *SyncMap)
  • func (e Env) LoadOutputVariables(vars *SyncMap)
  • func (e Env) MailerConfig(ctx context.Context) (mailer.Config, error)
  • func (e Env) VariablesMap() map[string]string
  • func (e Env) WithEnv(envs ...string) Env
• type Executor
  • func NewExecutor(ctx context.Context, step digraph.Step) (Executor, error)
• type ExitCoder
• type NodeStatusDeterminer
• type ParallelExecutor
• type PullPolicy
• type RunParams
• type SyncMap
  • func (m *SyncMap) MarshalJSON() ([]byte, error)
  • func (m *SyncMap) MarshalJSONIndent(prefix, indent string) ([]byte, error)
  • func (m *SyncMap) Store(key, value any)
  • func (m *SyncMap) UnmarshalJSON(data []byte) error
  • func (m *SyncMap) Variables() map[string]string

Variables

var (
	ErrWorkingDirNotExist = fmt.Errorf("working directory does not exist")
)

Errors for DAG executor

Functions

func AllEnvs

func AllEnvs(ctx context.Context) []string

AllEnvs returns all environment variables that needs to be passed to the command. Each element is in the form of "key=value".

func Register

func Register(name string, register Creator)

func SetupCommand

func SetupCommand(cmd *exec.Cmd)

SetupCommand configures Unix-specific command attributes

func WithEnv

func WithEnv(ctx context.Context, e Env) context.Context

WithEnv returns a new context with the given execution context.

Types

type ChildDAGExecutor

type ChildDAGExecutor struct {
	// DAG is the child DAG to execute.
	// For local DAGs, this DAG's Location will be set to a temporary file.
	DAG *digraph.DAG
	// contains filtered or unexported fields
}

ChildDAGExecutor is a helper for executing child DAGs. It handles both regular DAGs and local DAGs (defined in the same file).

func NewChildDAGExecutor

func NewChildDAGExecutor(ctx context.Context, childName string) (*ChildDAGExecutor, error)

NewChildDAGExecutor creates a new ChildDAGExecutor. It handles the logic for finding the DAG - either from the database or from local DAGs defined in the parent.

func (*ChildDAGExecutor) BuildCoordinatorTask

func (e *ChildDAGExecutor) BuildCoordinatorTask(
	ctx context.Context,
	runParams RunParams,
) (*coordinatorv1.Task, error)

BuildCoordinatorTask creates a coordinator task for distributed execution

func (*ChildDAGExecutor) Cleanup

func (e *ChildDAGExecutor) Cleanup(ctx context.Context) error

Cleanup removes any temporary files created for local DAGs. This should be called after the child DAG execution is complete.

func (*ChildDAGExecutor) ExecuteWithResult

func (e *ChildDAGExecutor) ExecuteWithResult(ctx context.Context, runParams RunParams, workDir string) (*digraph.RunStatus, error)

ExecuteWithResult executes the child DAG and returns the result. This is useful for parallel execution where results need to be collected.

func (*ChildDAGExecutor) Kill

func (e *ChildDAGExecutor) Kill(sig os.Signal) error

Kill terminates all running child DAG processes (both local and distributed)

func (*ChildDAGExecutor) ShouldUseDistributedExecution

func (e *ChildDAGExecutor) ShouldUseDistributedExecution() bool

ShouldUseDistributedExecution checks if this child DAG should be executed via coordinator

type Creator

type Creator func(ctx context.Context, step digraph.Step) (Executor, error)

Creator is a function type that creates an Executor based on the step configuration.

type DAGExecutor

type DAGExecutor interface {
	Executor

	// SetParams sets the parameters for running a child DAG.
	SetParams(RunParams)
}

DAGExecutor is an interface for child DAG executors.

type Env

type Env struct {
	digraph.Env

	Variables  *SyncMap
	Step       digraph.Step
	Envs       map[string]string
	StepMap    map[string]cmdutil.StepInfo // Map of step ID to step info
	WorkingDir string                      // Working directory for the step
}

Env holds information about the DAG and the current step to execute including the variables (environment variables and DAG variables) that are available to the step.

func GetEnv

func GetEnv(ctx context.Context) Env

GetEnv returns the execution context from the given context.

func NewEnv

func NewEnv(ctx context.Context, step digraph.Step) Env

NewEnv creates a new execution context with the given step.

func (Env) AllEnvs

func (e Env) AllEnvs() []string

AllEnvs returns all environment variables that needs to be passed to the command.

func (Env) DAGRunRef

func (e Env) DAGRunRef() digraph.DAGRunRef

DAGRunRef returns the DAGRunRef for the current execution context.

func (Env) EvalBool

func (e Env) EvalBool(ctx context.Context, value any) (bool, error)

EvalBool evaluates the given value with the variables within the execution context

func (Env) EvalString

func (e Env) EvalString(ctx context.Context, s string, opts ...cmdutil.EvalOption) (string, error)

EvalString evaluates the given string with the variables within the execution context.

func (Env) ForceLoadOutputVariables

func (e Env) ForceLoadOutputVariables(vars *SyncMap)

ForceLoadOutputVariables forces loading of output variables into the execution context. This is the same as LoadOutputVariables, but it does not check if the key already exists.

func (Env) LoadOutputVariables

func (e Env) LoadOutputVariables(vars *SyncMap)

LoadOutputVariables loads the output variables from the given DAG into the

func (Env) MailerConfig

func (e Env) MailerConfig(ctx context.Context) (mailer.Config, error)

func (Env) VariablesMap

func (e Env) VariablesMap() map[string]string

func (Env) WithEnv

func (e Env) WithEnv(envs ...string) Env

WithEnv returns a new execution context with the given environment variable(s).

type Executor

type Executor interface {
	SetStdout(out io.Writer)
	SetStderr(out io.Writer)
	Kill(sig os.Signal) error
	Run(ctx context.Context) error
}

Executor is an interface for executing steps in a DAG.

func NewExecutor

func NewExecutor(ctx context.Context, step digraph.Step) (Executor, error)

type ExitCoder

type ExitCoder interface {
	ExitCode() int
}

ExitCoder is an interface for executors that can return an exit code.

type NodeStatusDeterminer

type NodeStatusDeterminer interface {
	DetermineNodeStatus(ctx context.Context) (status.NodeStatus, error)
}

NodeStatusDeterminer is an interface for executors that can determine the status of a node.

type ParallelExecutor

type ParallelExecutor interface {
	Executor

	// SetParamsList sets the parameters for running multiple child DAGs in parallel.
	SetParamsList([]RunParams)
}

ParallelExecutor is an interface for parallel step executors.

type PullPolicy

type PullPolicy int

const (
	PullPolicyAlways PullPolicy = iota
	PullPolicyNever
	PullPolicyMissing
)

type RunParams

type RunParams struct {
	RunID  string
	Params string
}

RunParams holds the parameters for running a child DAG.

type SyncMap

type SyncMap struct {
	sync.Map
	// contains filtered or unexported fields
}

SyncMap wraps a sync.Map to make it JSON serializable.

func (*SyncMap) MarshalJSON

func (m *SyncMap) MarshalJSON() ([]byte, error)

func (*SyncMap) MarshalJSONIndent

func (m *SyncMap) MarshalJSONIndent(prefix, indent string) ([]byte, error)

func (*SyncMap) Store

func (m *SyncMap) Store(key, value any)

func (*SyncMap) UnmarshalJSON

func (m *SyncMap) UnmarshalJSON(data []byte) error

func (*SyncMap) Variables

func (m *SyncMap) Variables() map[string]string

Variables returns the map of variables. A variable is a string in the form of "key=value".