engine

Documentation

Overview

Package engine orchestrates the execution of a keepup Flow. Groups are scheduled either as ordered waves (step mode) or topologically from the implicit data DAG (dag mode); both share the same Runner, OutputStore, Expander, and Logger interfaces.

Index

• Constants
• type Emitter
• type Engine
  • func New(cfg *config.Config, opts ...Option) *Engine
  • func (e *Engine) Outputs() OutputStore
  • func (e *Engine) RunFlow(ctx context.Context, flowName string) error
• type Event
• type JSONEmitter
  • func NewJSONEmitter(w io.Writer) *JSONEmitter
  • func (e *JSONEmitter) Emit(ev Event)
• type MemoryOutputStore
  • func NewMemoryOutputStore() *MemoryOutputStore
  • func (s *MemoryOutputStore) Get(name string) (result.RunResult, bool)
  • func (s *MemoryOutputStore) Set(name string, r result.RunResult)
  • func (s *MemoryOutputStore) Snapshot() map[string]result.RunResult
• type Option
  • func WithCache(s cache.Store) Option
  • func WithDryRun(dry bool) Option
  • func WithEmitter(em Emitter) Option
  • func WithExpander(x template.Expander) Option
  • func WithLogger(l logger.Logger) Option
  • func WithNoCache(disable bool) Option
  • func WithOutputStore(s OutputStore) Option
  • func WithProber(p Prober) Option
  • func WithRetryBackoff(d time.Duration) Option
  • func WithRunner(r Runner) Option
• type OutputStore
• type Prober
• type Runner
• type ShellProber
  • func (ShellProber) Probe(ctx context.Context, script string, env map[string]string) error
• type ShellRunner
  • func NewShellRunner() *ShellRunner
  • func (r *ShellRunner) Run(ctx context.Context, g *config.Group, params []string, ...) (result.RunResult, error)

Constants

const (
	EventFlowStart  = "flow.start"
	EventFlowEnd    = "flow.end"
	EventGroupStart = "group.start"
	EventGroupEnd   = "group.end"
)

Event names emitted over the run lifecycle.

const (
	StatusOK       = "ok"
	StatusFailed   = "failed"
	StatusSkipped  = "skipped"
	StatusCacheHit = "cache-hit"
	StatusDryRun   = "dry-run"
)

Group end statuses.

const DefaultRetryBackoff = 250 * time.Millisecond

DefaultRetryBackoff is the base delay between retry attempts; the delay for attempt N is DefaultRetryBackoff * N.

Types

type Emitter

type Emitter interface {
	Emit(Event)
}

Emitter receives lifecycle events. Implementations must be safe for concurrent use (groups run in parallel).

type Engine

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

Engine binds a parsed Config to a set of pluggable collaborators.

func New

func New(cfg *config.Config, opts ...Option) *Engine

New constructs an Engine. The config pointer is held by reference; do not mutate it for the lifetime of the Engine.

func (*Engine) Outputs

func (e *Engine) Outputs() OutputStore

Outputs returns the captured outputs (populated after RunFlow completes).

func (*Engine) RunFlow

func (e *Engine) RunFlow(ctx context.Context, flowName string) error

RunFlow executes the named Flow, honoring ctx cancellation.

type Event

type Event struct {
	Event      string    `json:"event"`
	Flow       string    `json:"flow,omitempty"`
	Group      string    `json:"group,omitempty"`
	Mode       string    `json:"mode,omitempty"`
	Status     string    `json:"status,omitempty"`
	DurationMS int64     `json:"durationMs,omitempty"`
	Err        string    `json:"err,omitempty"`
	Reason     string    `json:"reason,omitempty"`
	Time       time.Time `json:"time"`
}

Event is a single structured run event for machine consumption (CI tooling).

type JSONEmitter

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

JSONEmitter writes one JSON object per line to a writer.

func NewJSONEmitter

func NewJSONEmitter(w io.Writer) *JSONEmitter

NewJSONEmitter returns an Emitter that writes newline-delimited JSON to w.

func (*JSONEmitter) Emit

func (e *JSONEmitter) Emit(ev Event)

Emit serializes ev as a single JSON line. Encoding errors are dropped — the event stream is best-effort observability, never load-bearing.

type MemoryOutputStore

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

MemoryOutputStore is the default in-memory implementation.

func NewMemoryOutputStore

func NewMemoryOutputStore() *MemoryOutputStore

NewMemoryOutputStore returns an empty store.

func (*MemoryOutputStore) Get

func (s *MemoryOutputStore) Get(name string) (result.RunResult, bool)

Get returns the stored RunResult for name, if any.

func (*MemoryOutputStore) Set

func (s *MemoryOutputStore) Set(name string, r result.RunResult)

Set stores a RunResult under name.

func (*MemoryOutputStore) Snapshot

func (s *MemoryOutputStore) Snapshot() map[string]result.RunResult

Snapshot returns a copy of the current state for safe iteration.

type Option

type Option func(*Engine)

Option configures an Engine.

func WithCache

func WithCache(s cache.Store) Option

WithCache overrides the default file-backed cache store.

func WithDryRun

func WithDryRun(dry bool) Option

WithDryRun forces dry-run mode regardless of the config flag.

func WithEmitter

func WithEmitter(em Emitter) Option

WithEmitter sets the structured event emitter (default: discard).

func WithExpander

func WithExpander(x template.Expander) Option

WithExpander overrides the default template.Expander.

func WithLogger

func WithLogger(l logger.Logger) Option

WithLogger overrides the default no-op Logger.

func WithNoCache

func WithNoCache(disable bool) Option

WithNoCache disables cache reads/writes for this run.

func WithOutputStore

func WithOutputStore(s OutputStore) Option

WithOutputStore overrides the default in-memory OutputStore.

func WithProber

func WithProber(p Prober) Option

WithProber overrides the default ShellProber used for skip-if/require.

func WithRetryBackoff

func WithRetryBackoff(d time.Duration) Option

WithRetryBackoff overrides the base retry backoff (delay for attempt N is base*N). Primarily useful in tests to avoid real sleeps.

func WithRunner

func WithRunner(r Runner) Option

WithRunner overrides the default ShellRunner.

type OutputStore

type OutputStore interface {
	Get(name string) (result.RunResult, bool)
	Set(name string, r result.RunResult)
	Snapshot() map[string]result.RunResult
}

OutputStore is a goroutine-safe key/value store of structured group results.

type Prober

type Prober interface {
	Probe(ctx context.Context, script string, env map[string]string) error
}

Prober evaluates a gating predicate. A nil error means the predicate succeeded (exit code 0); a non-nil error means it failed.

Predicates are short shell snippets (e.g. "test -f bin/app"), so they always run through a shell.

type Runner

type Runner interface {
	Run(ctx context.Context, g *config.Group, params []string, globalEnv map[string]string) (result.RunResult, error)
}

Runner executes a single group and returns its structured RunResult.

type ShellProber

type ShellProber struct{}

ShellProber runs predicates through the platform shell. Output is discarded; only the exit status matters.

func (ShellProber) Probe

func (ShellProber) Probe(ctx context.Context, script string, env map[string]string) error

Probe runs script via the platform shell and returns its exit status as an error (nil on success).

type ShellRunner

type ShellRunner struct {
	// Stdout and Stderr are forwarded for live output; outputs are also captured
	// into the returned RunResult. If nil, os.Stdout/os.Stderr are used.
	Stdout io.Writer
	Stderr io.Writer
}

ShellRunner executes a group via os/exec, optionally through a system shell.

By default (group.Shell == false) it spawns Command directly with Params as argv — no shell interpretation, no injection surface. When group.Shell is true the runner concatenates command+params into a single string and pipes it through the user's preferred shell; that mode is opt-in.

func NewShellRunner

func NewShellRunner() *ShellRunner

NewShellRunner returns a runner wired to the process stdio.

func (*ShellRunner) Run

func (r *ShellRunner) Run(ctx context.Context, g *config.Group, params []string, globalEnv map[string]string) (result.RunResult, error)

Run honors ctx for cancellation. It captures stdout, stderr, and the chronologically interleaved combined stream into three buffers populated on the returned RunResult; ExitCode and DurationMs are also filled in.

The command and arguments come from user-supplied configuration; that is the point of this tool. gosec G204 is suppressed for the exec call.