workflow

package
v0.6.0 Latest Latest
Warning

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

Go to latest
Published: Sep 6, 2026 License: MIT Imports: 4 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

	// 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 {
	Nome      string
	Tipo      TipoParam
	Padrao    string
	Descricao string

	// Enum restringe os valores aceitos. Vazio = qualquer um que passe no tipo.
	Enum []string

	// Pattern e uma expressao regular que o valor precisa casar. Existe para o
	// autor AMPLIAR o que o tipo `string` aceita por padrao — ver `seguro`.
	Pattern string
}

Param e um parametro de execucao: o que muda entre dois disparos do MESMO workflow sem editar o arquivo.

Era a maior distancia entre este motor e o Kestra/Leoflow. Sem params nao ha backfill (`load_full=true`) nem reprocessamento de janela, e oito dos 51 flows do repositorio de dados sequer podiam ser convertidos — o comando deles carrega `{{ inputs.start_date }}` e afins.

func (Param) Aceita

func (p Param) Aceita(valor string) error

Aceita valida um VALOR contra a declaracao.

func (Param) Validar

func (p Param) Validar() error

Validar confere a declaracao do param, nao o valor.

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) ComPadrao

func (r Resources) ComPadrao(p Resources) Resources

ComPadrao 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.

func (Resources) Vazio

func (r Resources) Vazio() bool

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

type TipoParam

type TipoParam string
const (
	ParamTexto   TipoParam = "string"
	ParamBool    TipoParam = "boolean"
	ParamInteiro TipoParam = "integer"
)

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

	// MaxAtivos 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.
	MaxAtivos 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) ImagemDe

func (w Workflow) ImagemDe(n Node) string

ImagemDe resolve a imagem efetiva de um passo.

func (Workflow) RecursosDe

func (w Workflow) RecursosDe(n Node) Resources

RecursosDe resolve os recursos efetivos de um passo.

func (Workflow) Resolver

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

Resolver mistura os valores informados com os padroes e valida tudo.

Chave desconhecida e ERRO, nao silencio: `--param lod_full=true` com typo rodaria o workflow com o padrao e ninguem perceberia que o backfill nao aconteceu.

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