execution

package
v0.13.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Sep 11, 2026 License: MIT Imports: 7 Imported by: 0

Documentation

Overview

Package execution defines the contract for executing tasks.

The interface is section 13's, with one difference: `Execute` returns a channel of events instead of blocking. A pod that runs for twenty minutes has to report progress before it ends, and the same holds for a local process writing to stdout.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func Render added in v0.7.0

func Render(command string, params map[string]string) (string, error)

Render substitutes the params into a step's command line.

The stdlib's `text/template`, with `missingkey=error`: a param with a typo in the YAML fails HERE, naming what was missing, instead of becoming an empty string and producing a silently wrong command — `--select ` with no target, or `--date` with no date.

So only the command is rendered. `image:` is deliberately NOT templatable: whoever triggers a run would be choosing the image the pod runs, which is choosing the code that executes.

Types

type Event

type Event struct {
	Kind     EventKind
	NodeID   string
	Message  string
	Stream   string // "stdout" | "stderr"
	ExitCode int
	Err      error
}

Event is one occurrence during the run. `Stream` tells stdout from stderr: merging the two loses the information about where the message came from, and that is exactly what made dbt's final summary show up as an error in Leoflow.

type EventKind

type EventKind string

EventKind classifies what the executor reports.

const (
	EventStarted   EventKind = "started"
	EventLog       EventKind = "log"
	EventSucceeded EventKind = "succeeded"

	// EventContext carries what the step published, in Message.
	//
	// It travels as an event for the same reason the phases do: the runner
	// collects it without knowing whether it came from a file on this disk or
	// from a pod's termination message, so a third executor costs the runner
	// nothing.
	EventContext EventKind = "context"
	EventFailed  EventKind = "failed"
)

type Executor

type Executor interface {
	Name() string
	Execute(ctx context.Context, t TaskExec) (<-chan Event, error)
	Cancel(ctx context.Context, execID string) error
}

Executor runs a task and reports what happens.

type FuncTask

type FuncTask struct {
	TaskName string
	Fn       func(ctx context.Context, in Input) error
}

FuncTask adapts a function to the Task interface, for the cases where a type of its own would be ceremony with no gain.

func (FuncTask) Name

func (f FuncTask) Name() string

func (FuncTask) Run

func (f FuncTask) Run(ctx context.Context, in Input) error

type Input

type Input struct {
	NodeID string
	With   map[string]any

	// Log emits one line into the event stream. It exists so the task can report
	// progress without knowing about channels or the executor.
	Log func(msg string)
}

Input is what the task receives.

func (Input) Text added in v0.7.0

func (i Input) Text(key string) (string, error)

Texto reads a required parameter out of `with`. A convenience with a useful error: a task doing the type assertion by hand repeats the same poor message.

type Registry

type Registry struct {
	// contains filtered or unexported fields
}

Registry holds the available tasks. Safe for concurrent use because the dispatcher queries it from several goroutines.

func NewRegistry

func NewRegistry() *Registry

func (*Registry) Get

func (r *Registry) Get(name string) (Task, bool)

Get looks a task up by name.

func (*Registry) MustRegister

func (r *Registry) MustRegister(t Task)

MustRegister registra e entra em panico se falhar.

For use in `init()` or at boot: an invalid registration is a programming error, and failing at start beats finding out on the first scheduled run.

func (*Registry) Names added in v0.7.0

func (r *Registry) Names() []string

Nomes lists what is registered, sorted. It serves the unknown-task error: saying what does exist saves a trip to the documentation.

func (*Registry) Register

func (r *Registry) Register(t Task) error

Register adds a task.

It refuses a duplicate name rather than overwriting: a silently replaced registration is a bug that only shows up in production, when the wrong task runs.

type Task

type Task interface {
	Name() string
	Run(ctx context.Context, in Input) error
}

Task is a unit of work written in Go, compiled into the binary.

The rule is categorical: "Do not execute arbitrary code received through the API. Local tasks must be compiled and registered in the runtime". The registry exists to make that structural — the YAML can only name something that is already in the binary, never supply the code.

type TaskExec

type TaskExec struct {
	ExecutionID string
	NodeID      string

	// Workflow, RunID and Attempt do not change the run — they identify it.
	// In Kubernetes they become the pod's labels, and they are what makes it
	// possible to find "that run's pods" without searching by name.
	Workflow string
	RunID    string

	// The STEP's attempt, within one execution of the run.
	Attempt int

	// TentativaDoRun is the RUN's, counted by the dispatcher. Both go into the
	// pod's name: without the second, a dispatcher retry recreates the run from
	// scratch (the step at attempt 0 again) and runs into the previous pod.
	RunAttempt int

	Command string // a shell line, for the ProcessExecutor
	Action  string // the name in the registry, for the GoExecutor
	With    map[string]any

	// Image is this step's runtime. Empty in local mode (the command runs on the
	// instance itself); required in Kubernetes, where it IS the pod.
	Image string

	// Shell chooses between `sh -c "line"` and plain argv. It matters for a
	// distroless image, which has no shell at all.
	Shell bool

	// The pod's resources, in Kubernetes's format. Ignored in local mode, where
	// a process's limit is the machine's.
	CPU, Memoria       string
	CPUMax, MemoriaMax string

	WorkDir string
	Env     map[string]string

	// Secrets are variables whose value the engine does NOT carry: the map is
	// variable-name -> `secret-name/key`, and the executor is what resolves it.
	//
	// They are kept apart from Env for exactly that reason. If the value arrived
	// resolved here, it would pass through the dispatcher, through the task
	// assembly log and through any TaskExec dump somebody writes later.
	Secrets map[string]string

	// OutputPath is where the step writes what it publishes for the steps that
	// depend on it. It reaches the step as BREVIS_OUTPUT.
	//
	// The RUNNER picks it and the EXECUTOR may override it, because only the
	// executor knows what a path means in its world: a temporary file in the
	// engine's filesystem is meaningless inside a pod, where the answer is
	// /dev/termination-log -- which the engine already reads to get the exit
	// code.
	OutputPath string

	// A zero Timeout means no limit. A timeout was asked for early on, and
	// leaving the default open is deliberate — imposing an arbitrary limit would
	// kill legitimately long tasks.
	Timeout time.Duration
}

TaskExec is what gets asked to run. Deliberately poor: the executor knows nothing of workflows, dependencies or schedules.

`Command` and `Action` are exclusive: the first goes to the ProcessExecutor, the second resolves in the Go task registry. The runner is what chooses the executor, not the executor itself.

Directories

Path Synopsis
Package kubernetes runs each step of a workflow as a POD of its own.
Package kubernetes runs each step of a workflow as a POD of its own.
Package local implements running processes on the host.
Package local implements running processes on the host.
Package remote runs a step on a host the engine does not manage.
Package remote runs a step on a host the engine does not manage.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL