Documentation
¶
Overview ¶
Package execution (application) percorre o grafo e executa os nos.
Esta e a versao LOCAL: sem fila, sem persistencia, sem scheduler — essas pecas tem fase propria no plano (§37, fases 2 e 4). O que existe aqui e o suficiente para `brevis run arquivo.yaml` rodar na propria instancia, que foi o pedido.
Index ¶
Constants ¶
const TetoDoLog = 128 << 10
TetoDoLog e quanto da saida de um passo vai para o banco.
128 KB cobre com folga uma run de dbt com 60 nos (~25 KB de texto) e ainda segura um backfill barulhento. O teto existe porque um `while true; do echo` num workflow qualquer nao pode encher o disco do Postgres.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type ErroDePasso ¶
type ErroDePasso struct {
NodeID string
ExitCode int
Mensagem string
// Saida sao as ultimas linhas de stderr. Guardar so as ultimas, e nao tudo,
// porque um processo verboso encheria a coluna de erro do banco — e a causa
// quase sempre esta no fim.
Saida []string
}
ErroDePasso e a falha de um passo, com o contexto necessario para entende-la sem abrir log nenhum: o codigo de saida, o que ele significa, e as ultimas linhas que o processo escreveu em stderr.
Antes so sobrava "saiu com codigo 127" — tecnicamente correto e inutil. A causa (`/bin/sh: python: not found`) passava pelos eventos como log e era descartada ali mesmo, entao a tela mostrava o sintoma sem a explicacao.
func (*ErroDePasso) Error ¶
func (e *ErroDePasso) Error() string
type Etapa ¶
type Etapa struct {
Nome string `json:"nome"`
Estado string `json:"estado"`
Ms *int64 `json:"ms,omitempty"`
Em string `json:"em"`
Numeros map[string]any `json:"numeros,omitempty"`
}
Etapa e uma fase de um passo do SDK, como ela esta agora.
EtapaGravada e o mesmo tipo sob o nome pelo qual ele atravessa o banco: e o que um teste de fora do pacote precisa para conferir o que foi gravado.
type EtapaGravada ¶
type EtapaGravada = Etapa
etapasConhecidas e uma lista fechada de proposito: uma etapa que este motor nao conhece e ignorada, em vez de virar um bloco sem sentido na tela. EtapaGravada e o formato em que uma Etapa vai para o JSONB.
type Historico ¶
type Historico interface {
PassoJaTeveSucesso(ctx context.Context, workflowSlug, nodeID string, exceto uuid.UUID) (bool, error)
}
Persistidor grava o estado de cada passo. Opcional: o `brevis run` local nao tem banco, e exigi-lo tornaria a execucao ad-hoc dependente de infraestrutura. Historico responde se um passo ja teve sucesso antes. E o que decide se esta e a PRIMEIRA execucao dele — informacao que o passo nao tem e que so o engine possui.
Interface pequena e declarada aqui, no consumidor, e nao no pacote que a implementa.
A pergunta e por (workflow, passo), nao por workflow: um workflow com tres fetchers escrevendo em tres tabelas criaria apenas a do primeiro passo se a resposta fosse do workflow inteiro, e as outras duas falhariam em silencio.
type Persistidor ¶
type Persistidor interface {
IniciarTask(ctx context.Context, runID uuid.UUID, nodeID string, tentativa int) error
TerminarTask(ctx context.Context, runID uuid.UUID, nodeID string, tentativa int,
status run.Status, exit *int, erro string, log string) error
// RegistrarEtapas grava o estado das etapas de um passo do SDK, enquanto
// ele roda. E o que faz a tela avancar antes de o passo terminar.
RegistrarEtapas(ctx context.Context, runID uuid.UUID, nodeID string, tentativa int,
sdkVersao string, etapas json.RawMessage) error
}
type Reporter ¶
Reporter recebe os eventos da execucao. Interface pequena para que a CLI, os testes e o persistidor possam observar o mesmo fluxo.
type Runner ¶
type Runner struct {
Processo execution.Executor // atende `run:`; pode ser nil se so houver tasks Go
Go execution.Executor // atende `action:`; pode ser nil
WorkDir string
Env map[string]string
Report Reporter
// Timeout por no. Zero = sem limite.
Timeout time.Duration
// MaxTentativas por no. Zero ou 1 = tentativa unica.
MaxTentativas int
BackoffBase time.Duration
// Persist e RunID sao usados juntos: sem os dois, o estado por passo nao e
// gravado e a DAG na UI aparece sem estado de execucao.
Persist Persistidor
RunID uuid.UUID
// Params sao os valores desta execucao. Entram no comando do passo por
// template (ver execution.Renderizar) e no ambiente do passo, para que um
// fetcher que use o SDK os enxergue sem receber nada por argumento.
Params map[string]string
// Trigger diz por que este Run existe: schedule, manual ou backfill.
Trigger string
// LogicalDate e o slot que este Run representa. Nulo em disparo manual.
LogicalDate *time.Time
// Historico decide se um passo esta rodando pela primeira vez. Nulo
// significa que nao da para saber — e nesse caso o passo recebe
// first=false, porque criar tabela sem certeza e pior que nao criar.
Historico Historico
// Vagas limita quantos PASSOS correm ao mesmo tempo — em Kubernetes, quantos
// pods existem simultaneamente. Nulo = sem limite.
//
// Precisa ser compartilhado entre todos os Runners do processo, e por isso e
// injetado em vez de criado aqui: o teto e do CLUSTER, nao de um workflow.
// Sem ele, o limite de concorrencia do dispatcher contava RUNS — cinco runs
// com tres passos paralelos cada davam quinze pods, nao cinco.
Vagas chan struct{}
// TentativaDoRun e a tentativa deste RUN, contada pelo dispatcher. Entra no
// nome do pod para que um retry nao reencontre o pod da tentativa anterior.
TentativaDoRun int
// Pods executa passos como pod no Kubernetes. Quando presente, ele atende
// todo passo que declara `image:` — e a mesma DAG roda em pod no cluster e
// em processo na maquina, sem alterar o YAML.
Pods execution.Executor
}
Runner executa um workflow inteiro.
Guarda DOIS executores e escolhe por no: `run:` vai para o de processo, `action:` resolve no registry Go. A escolha e do runner, e nao do executor, para que cada executor continue ignorando a existencia do outro.
func (Runner) Run ¶
Run percorre o grafo por niveis: tudo dentro de um nivel roda em paralelo, e o nivel seguinte so comeca quando o anterior fecha inteiro.
Para na PRIMEIRA falha do nivel, sem iniciar o proximo. Continuar depois de um erro produziria resultado parcial que parece completo — foi assim que uma pipeline ficou 28 dias atrasada sem ninguem ver, no sistema que este substitui.