flow

Documentation

Overview

Package flow provides utilities to construct a directed acyclic computational graph that is then executed and monitored with maximum parallelism.

Index

• Variables
• func Causes(err error) *multierror.Error
• func Errors(err error) *multierror.Error
• func WasCanceled(err error) bool
• type ErrorCleaner
• type Flow
  • func (f *Flow) Len() int
  • func (f *Flow) Name() string
  • func (f *Flow) Run(ctx context.Context, opts Opts) error
• type Graph
  • func NewGraph(name string) *Graph
  • func (g *Graph) Add(task Task) TaskID
  • func (g *Graph) Compile() *Flow
  • func (g *Graph) Name() string
• type Opts
• type ProgressReporter
  • func NewDelayingProgressReporter(clock clock.Clock, reporterFn ProgressReporterFn, period time.Duration) ProgressReporter
  • func NewImmediateProgressReporter(reporterFn ProgressReporterFn) ProgressReporter
• type ProgressReporterFn
• type RecoverFn
• type Stats
  • func InitialStats(flowName string, all TaskIDs) *Stats
  • func (s *Stats) Copy() *Stats
  • func (s *Stats) ProgressPercent() int32
• type Task
  • func (t *Task) Spec() *TaskSpec
• type TaskFn
  • func Parallel(fns ...TaskFn) TaskFn
  • func ParallelExitOnError(fns ...TaskFn) TaskFn
  • func Sequential(fns ...TaskFn) TaskFn
  • func (t TaskFn) DoIf(condition bool) TaskFn
  • func (t TaskFn) Recover(recoverFn RecoverFn) TaskFn
  • func (t TaskFn) RetryUntilTimeout(interval, timeout time.Duration) TaskFn
  • func (t TaskFn) SkipIf(condition bool) TaskFn
  • func (t TaskFn) Timeout(timeout time.Duration) TaskFn
  • func (t TaskFn) ToRecoverFn() RecoverFn
• type TaskID
  • func (t TaskID) TaskIDs() []TaskID
• type TaskIDSlice
  • func (t TaskIDSlice) Len() int
  • func (t TaskIDSlice) Less(i1, i2 int) bool
  • func (t TaskIDSlice) Swap(i1, i2 int)
  • func (t TaskIDSlice) TaskIDs() []TaskID
• type TaskIDer
• type TaskIDs
  • func NewTaskIDs(ids ...TaskIDer) TaskIDs
  • func (t TaskIDs) Copy() TaskIDs
  • func (t TaskIDs) Delete(iders ...TaskIDer) TaskIDs
  • func (t TaskIDs) Has(id TaskID) bool
  • func (t TaskIDs) Insert(iders ...TaskIDer) TaskIDs
  • func (t TaskIDs) InsertIf(condition bool, iders ...TaskIDer) TaskIDs
  • func (t TaskIDs) Len() int
  • func (t TaskIDs) List() TaskIDSlice
  • func (t TaskIDs) StringList() []string
  • func (t TaskIDs) TaskIDs() []TaskID
  • func (t TaskIDs) UnsortedList() TaskIDSlice
  • func (t TaskIDs) UnsortedStringList() []string
• type TaskSpec
• type Tasks

Variables

var (
	// ContextWithTimeout is context.WithTimeout. Exposed for testing.
	ContextWithTimeout = context.WithTimeout
)

Functions

func Causes

func Causes(err error) *multierror.Error

Causes reports the causes of all Task errors of the given Flow error.

func Errors

func Errors(err error) *multierror.Error

Errors reports all wrapped Task errors of the given Flow error.

func WasCanceled

func WasCanceled(err error) bool

WasCanceled determines whether the given flow error was caused by cancellation.

Types

type ErrorCleaner

type ErrorCleaner func(context.Context, string)

ErrorCleaner is called when a task which errored during the previous reconciliation phase completes with success

type Flow

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

Flow is a validated executable Graph.

func (*Flow) Len

func (f *Flow) Len() int

Len retrieves the amount of tasks in a Flow.

func (*Flow) Name

func (f *Flow) Name() string

Name retrieves the name of a flow.

func (*Flow) Run

func (f *Flow) Run(ctx context.Context, opts Opts) error

Run starts an execution of a Flow. It blocks until the Flow has finished and returns the error, if any.

type Graph

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

Graph is a builder for a Flow.

func NewGraph

func NewGraph(name string) *Graph

NewGraph returns a new Graph with the given name.

func (*Graph) Add

func (g *Graph) Add(task Task) TaskID

Add adds the given Task to the graph. This panics if - There is already a Task present with the same name - One of the dependencies of the Task is not present

func (*Graph) Compile

func (g *Graph) Compile() *Flow

Compile compiles the graph into an executable Flow.

func (*Graph) Name

func (g *Graph) Name() string

Name returns the name of a graph.

type Opts

type Opts struct {
	// Logger is used to log any output during flow execution.
	Logger logrus.FieldLogger
	// ProgressReporter is used to report the progress during flow execution.
	ProgressReporter ProgressReporter
	// ErrorCleaner is used to clean up a previously failed task.
	ErrorCleaner func(ctx context.Context, taskID string)
	// ErrorContext is used to store any error related context.
	ErrorContext *utilerrors.ErrorContext
}

Opts are options for a Flow execution. If they are not set, they are left blank and don't affect the Flow.

type ProgressReporter

type ProgressReporter interface {
	// Start starts the progress reporter.
	Start(context.Context) error
	// Stop stops the progress reporter.
	Stop()
	// Report reports the progress using the current statistics.
	Report(context.Context, *Stats)
}

ProgressReporter is used to report the current progress of a flow.

func NewDelayingProgressReporter

func NewDelayingProgressReporter(clock clock.Clock, reporterFn ProgressReporterFn, period time.Duration) ProgressReporter

NewDelayingProgressReporter returns a new progress reporter with the given function and the configured period. A period of `0` will lead to immediate reports as soon as flow tasks are completed.

func NewImmediateProgressReporter

func NewImmediateProgressReporter(reporterFn ProgressReporterFn) ProgressReporter

NewImmediateProgressReporter returns a new progress reporter with the given function.

type ProgressReporterFn

type ProgressReporterFn func(context.Context, *Stats)

ProgressReporterFn is continuously called on progress in a flow.

type RecoverFn

type RecoverFn func(ctx context.Context, err error) error

RecoverFn is a function that can recover an error.

type Stats

type Stats struct {
	FlowName  string
	All       TaskIDs
	Succeeded TaskIDs
	Failed    TaskIDs
	Running   TaskIDs
	Pending   TaskIDs
}

Stats are the statistics of a Flow execution.

func InitialStats

func InitialStats(flowName string, all TaskIDs) *Stats

InitialStats creates a new Stats object with the given set of initial TaskIDs. The initial TaskIDs are added to all TaskIDs as well as to the pending ones.

func (*Stats) Copy

func (s *Stats) Copy() *Stats

Copy deeply copies a Stats object.

func (*Stats) ProgressPercent

func (s *Stats) ProgressPercent() int32

ProgressPercent retrieves the progress of a Flow execution in percent.

type Task

type Task struct {
	Name         string
	Fn           TaskFn
	Dependencies TaskIDs
}

Task is a unit of work. It has a name, a payload function and a set of dependencies. A is only started once all its dependencies have been completed successfully.

func (*Task) Spec

func (t *Task) Spec() *TaskSpec

Spec returns the TaskSpec of a task.

type TaskFn

type TaskFn func(ctx context.Context) error

TaskFn is a payload function of a task.

var EmptyTaskFn TaskFn = func(ctx context.Context) error { return nil }

EmptyTaskFn is a TaskFn that does nothing (returns nil).

func Parallel

func Parallel(fns ...TaskFn) TaskFn

Parallel runs the given TaskFns in parallel, collecting their errors in a multierror.

func ParallelExitOnError

func ParallelExitOnError(fns ...TaskFn) TaskFn

ParallelExitOnError runs the given TaskFns in parallel and stops execution as soon as one TaskFn returns an error.

func Sequential

func Sequential(fns ...TaskFn) TaskFn

Sequential runs the given TaskFns sequentially.

func (TaskFn) DoIf

func (t TaskFn) DoIf(condition bool) TaskFn

DoIf returns a TaskFn that will be executed if the condition is true when it is called. Otherwise, it will do nothing when called.

func (TaskFn) Recover

func (t TaskFn) Recover(recoverFn RecoverFn) TaskFn

Recover creates a new TaskFn that recovers an error with the given RecoverFn.

func (TaskFn) RetryUntilTimeout

func (t TaskFn) RetryUntilTimeout(interval, timeout time.Duration) TaskFn

RetryUntilTimeout returns a TaskFn that is retried until the timeout is reached.

func (TaskFn) SkipIf

func (t TaskFn) SkipIf(condition bool) TaskFn

SkipIf returns a TaskFn that does nothing if the condition is true, otherwise the function will be executed once called.

func (TaskFn) Timeout

func (t TaskFn) Timeout(timeout time.Duration) TaskFn

Timeout returns a TaskFn that is bound to a context which times out.

func (TaskFn) ToRecoverFn

func (t TaskFn) ToRecoverFn() RecoverFn

ToRecoverFn converts the TaskFn to a RecoverFn that ignores the incoming error.

type TaskID

type TaskID string

TaskID is an id of a task.

func (TaskID) TaskIDs

func (t TaskID) TaskIDs() []TaskID

TaskIDs retrieves this TaskID as a singleton slice.

type TaskIDSlice

type TaskIDSlice []TaskID

TaskIDSlice is a slice of TaskIDs.

func (TaskIDSlice) Len

func (t TaskIDSlice) Len() int

func (TaskIDSlice) Less

func (t TaskIDSlice) Less(i1, i2 int) bool

func (TaskIDSlice) Swap

func (t TaskIDSlice) Swap(i1, i2 int)

func (TaskIDSlice) TaskIDs

func (t TaskIDSlice) TaskIDs() []TaskID

TaskIDs returns this as a slice of TaskIDs.

type TaskIDer

type TaskIDer interface {
	// TaskIDs reports all TaskIDs of this TaskIDer.
	TaskIDs() []TaskID
}

TaskIDer can produce a slice of TaskIDs. Default implementations of this are TaskIDs, TaskID and TaskIDSlice

type TaskIDs

type TaskIDs map[TaskID]struct{}

TaskIDs is a set of TaskID.

func NewTaskIDs

func NewTaskIDs(ids ...TaskIDer) TaskIDs

NewTaskIDs returns a new set of TaskIDs initialized to contain all TaskIDs of the given TaskIDers.

func (TaskIDs) Copy

func (t TaskIDs) Copy() TaskIDs

Copy makes a deep copy of this TaskIDs.

func (TaskIDs) Delete

func (t TaskIDs) Delete(iders ...TaskIDer) TaskIDs

Delete deletes the TaskIDs of all TaskIDers from this TaskIDs.

func (TaskIDs) Has

func (t TaskIDs) Has(id TaskID) bool

Has checks if the given TaskID is present in this set.

func (TaskIDs) Insert

func (t TaskIDs) Insert(iders ...TaskIDer) TaskIDs

Insert inserts the TaskIDs of all TaskIDers into this TaskIDs.

func (TaskIDs) InsertIf

func (t TaskIDs) InsertIf(condition bool, iders ...TaskIDer) TaskIDs

InsertIf inserts the TaskIDs of all TaskIDers into this TaskIDs if the given condition evaluates to true.

func (TaskIDs) Len

func (t TaskIDs) Len() int

Len returns the amount of TaskIDs this contains.

func (TaskIDs) List

func (t TaskIDs) List() TaskIDSlice

List returns the elements of this in an ordered slice.

func (TaskIDs) StringList

func (t TaskIDs) StringList() []string

StringList returns the elements of this in an ordered string slice.

func (TaskIDs) TaskIDs

func (t TaskIDs) TaskIDs() []TaskID

TaskIDs retrieves all TaskIDs as an unsorted slice.

func (TaskIDs) UnsortedList

func (t TaskIDs) UnsortedList() TaskIDSlice

UnsortedList returns the elements of this in an unordered slice.

func (TaskIDs) UnsortedStringList

func (t TaskIDs) UnsortedStringList() []string

UnsortedStringList returns the elements of this in an unordered string slice.

type TaskSpec

type TaskSpec struct {
	Fn           TaskFn
	Dependencies TaskIDs
}

TaskSpec is functional body of a Task, consisting only of the payload function and the dependencies of the Task.

type Tasks

type Tasks map[TaskID]*TaskSpec

Tasks is a mapping from TaskID to TaskSpec.