workflow

package
v0.7.0 Latest Latest
Warning

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

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

Documentation

Overview

Package workflow is the domain model of a flow and its graph.

The domain does not know what YAML is. Translating the file into these structs lives in internal/application/workflow -- so the file format can change without touching the invariants.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Edge

type Edge struct {
	From string
	To   string
}

Edge links two nodes: From runs before To.

type Kind

type Kind string

Kind distinguishes how the graph was declared. `chain` is syntactic sugar: the parser turns it into edges before it gets here, so the execution engine only ever knows a DAG. One engine, two ways of writing.

const (
	KindChain Kind = "chain"
	KindDAG   Kind = "dag"
)

type Node

type Node struct {
	ID     string
	Run    string
	Action string
	With   map[string]any

	// Image overrides the workflow's. Empty inherits.
	Image string

	// Resources sizes this step's pod. The gain from separating per step is
	// concrete: a Go fetcher fits in 64Mi while the dbt next to it asks for
	// 1Gi, and under a single image both would pay the larger of the two.
	Resources Resources

	// Runtime and Tools say what this step runs in. Both empty is the normal
	// case: the engine infers from Run and Image, and only what the author
	// DECLARED lands here.
	//
	// They are additive in the stored document. A Workflow is written whole as
	// JSON into workflows.definicao and runs.definicao with no tags, so the Go
	// field name is the key -- an older engine ignores what it does not know,
	// and this one reading an older document gets the zero value, which means
	// "not declared" and falls through to the inference. Nothing to migrate.
	// See TestTheParamKeysAreTheOnDiskFormat for the case where that is NOT
	// true.
	Runtime string
	Tools   []string

	// Shell decides how the command enters the container. Nil means with a
	// shell, which is what `run:` suggests ("python fetch.py"). False passes the
	// argv directly, for a distroless image -- where `sh -c` would fail with
	// "no such file or directory", an error that says nothing about the
	// cause.
	Shell *bool

	// Env are this step's environment variables, with a literal value in the
	// file. They override the workflow's, name by name.
	//
	//	env:
	//	  BREVIS_LOG_LEVEL: info
	Env map[string]string

	// Secrets are variables whose VALUE never appears in the file. The key is
	// the variable's name; the value is where to find it, as `secret/key`.
	//
	//	secrets:
	//	  GABRIEL_SESSION_COOKIE: gabriel-session/cookie
	//
	// Two keys and not one, on purpose. With a single one, the shortest path to
	// making it work would be pasting the secret into the YAML -- and the YAML
	// is in git. `env:` accepts a literal, `secrets:` does not.
	//
	// Where the coordinate resolves depends on the executor, and the asymmetry
	// is deliberate:
	//
	//   Kubernetes  valueFrom.secretKeyRef{name: gabriel-session, key: cookie}
	//   local       the variable of the same name in the engine's own
	//               environment, and missing is an ERROR -- not an empty string
	//
	// In either case the engine passes it on without reading: the value enters
	// no log, no database and no rendered command.
	Secrets map[string]string
}

Node is a unit of work. Exactly one form of execution must be filled in -- `Run` (a command) or `Action` (a typed action with parameters).

func (Node) UsaShell

func (n Node) UsaShell() bool

UsaShell says whether the command enters through `sh -c`.

type Param

type Param struct {
	Name        string    `json:"Nome"`
	Type        ParamType `json:"Tipo"`
	Default     string    `json:"Default"`
	Description string    `json:"Descricao"`

	// Enum restricts the accepted values. Empty = any that passes the type.
	Enum []string `json:"Enum"`

	// Pattern is a regular expression the value has to match. It exists for the
	// author to WIDEN what the `string` type accepts by default — see
	// safeCharacters.
	Pattern string `json:"Pattern"`
}

Param is a run parameter: what changes between two triggers of the SAME workflow without editing the file.

It was the largest distance between this engine and Kestra/Leoflow. Without params there is no backfill (`load_full=true`) and no window reprocessing, and eight of the data repository's 51 flows could not even be converted — their command carries `{{ inputs.start_date }}` and the like. The tags below are the ON-DISK FORMAT, and they keep the Portuguese field names the Go struct used to have.

A Workflow is stored whole as JSON in `workflows.definicao` and `runs.definicao`, with no tags of its own -- so the Go field NAME was the key. Every workflow published before this rename holds `{"Nome":"date", "Tipo":"string","Descricao":"..."}`, and json.Unmarshal ignores a key it does not know: reading one with the fields renamed and untagged gives a param with no name, no type and no description, with no error and no log. The trigger form would render an empty field and the validation would refuse a value the author declared as valid.

Pinning the keys costs three tags and keeps both directions working: an engine on either side of this commit reads and writes the same document.

func (Param) Accepts added in v0.7.0

func (p Param) Accepts(value string) error

Accepts validates a VALUE against the declaration.

func (Param) Validate added in v0.7.0

func (p Param) Validate() error

Validate checks the param's declaration, not the value.

type ParamType added in v0.7.0

type ParamType string
const (
	ParamString  ParamType = "string"
	ParamBool    ParamType = "boolean"
	ParamInteger ParamType = "integer"
)

type Resources

type Resources struct {
	CPU         string
	Memory      string
	CPULimit    string
	MemoryLimit string
}

Resources are a pod's requests and limits, in Kubernetes' own format ("200m", "1Gi"). Text and not a number on purpose: the format belongs to Kubernetes, and converting to a unit of our own would only create a second vocabulary for the same thing.

func (Resources) Empty added in v0.7.0

func (r Resources) Empty() bool

Empty says whether nothing was declared -- the pod then starts without `resources`, inheriting the namespace's LimitRange.

func (Resources) WithDefaults added in v0.7.0

func (r Resources) WithDefaults(p Resources) Resources

WithDefaults fills what the step did not declare from the workflow's. Inheriting field by field, rather than the whole block, lets a step ask for more memory alone without losing the default CPU.

type Workflow

type Workflow struct {
	Slug     string
	Name     string
	Kind     Kind
	Schedule string // cron; vazio = so disparo manual

	// Tags classify the workflow for search and filtering in the UI. They are
	// free-form labels from the YAML's author, not domain: nothing in the
	// engine depends on them.
	Tags []string

	// Image is the steps' default runtime. In Kubernetes EVERY step becomes a
	// pod, and the image is what defines what that pod knows how to do: a dbt
	// step brings up the dbt image, a Go binary brings up a 10 MB one.
	// Declaring it here avoids repeating the same line in ten steps; a step
	// overrides it when it needs a different runtime.
	Image string

	// Resources is the default CPU and memory request, for the same reason.
	Resources Resources

	// Env are environment variables every step receives. A literal value, which
	// is why they are NOT for secrets: the YAML is in git.
	Env map[string]string

	// Secrets are variables whose value the engine neither sees nor stores. See Node.
	Secrets map[string]string

	// Params are the values that change between two dispatches of the same
	// workflow -- `load_full`, a date window, a limit. See param.go.
	Params []Param

	// MaxActive caps simultaneous runs OF THIS workflow. Zero means no limit.
	//
	// It differs from the global step ceiling: that one protects the CLUSTER,
	// this one protects the DATA. A `*/15` that takes 20 minutes overlaps
	// itself, and two `dbt build` on the same model at once fight over the same
	// table.
	MaxActive int

	Nodes []Node
	Edges []Edge
}

Workflow is a flow's definition. Immutable once published: §22 of the plan requires a Run to keep a snapshot of the version that produced it.

func (Workflow) EnvDe

func (w Workflow) EnvDe(n Node) map[string]string

EnvDe resolves a step's effective literal variables: the workflow's, with the step's on top.

func (Workflow) ImageFor added in v0.7.0

func (w Workflow) ImageFor(n Node) string

ImageFor resolves a step's effective image.

func (Workflow) Resolver

func (w Workflow) Resolver(given map[string]string) (map[string]string, error)

Resolver merges the supplied values with the defaults and validates all of them.

An unknown key is an ERROR, not silence: `--param lod_full=true` with a typo would run the workflow with the default and nobody would notice the backfill did not happen.

func (Workflow) ResourcesFor added in v0.7.0

func (w Workflow) ResourcesFor(n Node) Resources

ResourcesFor resolves a step's effective resources.

func (Workflow) SecretsDe

func (w Workflow) SecretsDe(n Node) map[string]string

SecretsDe resolves a step's effective secrets, by the same rule.

func (Workflow) Validate

func (w Workflow) Validate() error

Validate applies the invariants §5 of the plan requires before saving. Order matters: duplicate IDs and missing dependencies are checked before the cycle, because a graph with a dangling edge cannot be walked.

Jump to

Keyboard shortcuts

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