workflow

Documentation

Overview

Package workflow runs DAGs of cron.Jobs.

Index

• Variables
• type Condition
• type ConfigError
  • func (e *ConfigError) Error() string
  • func (e *ConfigError) Unwrap() error
• type Dep
  • func After(name string, when Condition) Dep
• type Execution
  • func (e *Execution) Err() error
• type Result
  • func (r Result) String() string
• type Step
  • func NewStep(name string, job cron.Job, deps ...Dep) Step
• type Workflow
  • func MustNew(steps ...Step) *Workflow
  • func New(steps ...Step) (*Workflow, error)
  • func (w *Workflow) Run(ctx context.Context) error
  • func (w *Workflow) WithOnComplete(cb func(*Execution)) *Workflow

Variables

var (
	ErrDuplicateStep = errors.New("workflow: duplicate step")
	ErrUnknownDep    = errors.New("workflow: unknown dependency")
	ErrCycle         = errors.New("workflow: cycle detected")
)

Types

type Condition

type Condition uint8

Condition selects which upstream outcome triggers a Step.

const (
	OnSuccess  Condition = iota // upstream succeeded
	OnFailure                   // upstream failed
	OnSkipped                   // upstream was skipped
	OnComplete                  // any terminal state
)

type ConfigError

type ConfigError struct {
	Err  error
	Step string
	Dep  string // ErrUnknownDep / ErrCycle only
}

func (*ConfigError) Error

func (e *ConfigError) Error() string

func (*ConfigError) Unwrap

func (e *ConfigError) Unwrap() error

type Dep

type Dep struct {
	Name string
	When Condition
}

Dep is one DAG edge.

func After

func After(name string, when Condition) Dep

type Execution

type Execution struct {
	ID      string
	Results map[string]Result
	Errors  map[string]error
}

Execution is the result of one Workflow run.

func (*Execution) Err

func (e *Execution) Err() error

Err joins recorded Step errors.

type Result

type Result uint8

Result is a Step outcome.

const (
	ResultPending Result = iota // never appears in a finished Execution
	ResultSuccess
	ResultFailure // Job returned an error or panicked
	ResultSkipped // a dep didn't satisfy this step's Condition, or ctx was cancelled
)

func (Result) String

func (r Result) String() string

type Step

type Step struct {
	Name string
	Job  cron.Job
	Deps []Dep
}

Step is one node in the DAG.

func NewStep

func NewStep(name string, job cron.Job, deps ...Dep) Step

type Workflow

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

Workflow is a DAG of Steps.

func MustNew

func MustNew(steps ...Step) *Workflow

MustNew panics on configuration error.

func New

func New(steps ...Step) (*Workflow, error)

New constructs a Workflow. It copies Step.Deps and returns *ConfigError for duplicate steps, unknown dependencies, or cycles.

func (*Workflow) Run

func (w *Workflow) Run(ctx context.Context) error

Run executes the DAG once and returns Execution.Err.

func (*Workflow) WithOnComplete

func (w *Workflow) WithOnComplete(cb func(*Execution)) *Workflow

WithOnComplete returns a shallow copy with cb installed. cb runs before Run returns.