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 MakeDescription(stats *Stats) string
- func RegisterMetrics(r prometheus.Registerer)
- func ReportRetry(ctx context.Context, err error)
- func WasCanceled(err error) bool
- type ErrorCleaner
- type Flow
- type Graph
- type Opts
- type ProgressReporter
- type ProgressReporterFn
- type RecoverFn
- type RetryReporter
- type Stats
- type Task
- type TaskFn
- type TaskGroup
- type TaskID
- type TaskIDSlice
- type TaskIDer
- type 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
Constants ¶
This section is empty.
Variables ¶
var ContextWithTimeout = context.WithTimeout
ContextWithTimeout is context.WithTimeout. Exposed for testing.
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 MakeDescription ¶ added in v1.76.0
MakeDescription returns a description based on the stats.
func RegisterMetrics ¶ added in v1.111.0
func RegisterMetrics(r prometheus.Registerer)
RegisterMetrics registers the metrics for the flow library on the passed registry. This function can only be called once. If this function is not called, no metrics are collected in this package.
func ReportRetry ¶
ReportRetry reports that the current task failed with err and will be retried. It extracts the reporter from the context (injected by the flow engine) and calls it. This is a no-op if no reporter is present in the context.
Call this inside a custom retry loop to surface the last error to the progress reporter. When using TaskFn.RetryUntilTimeout, ReportRetry is called automatically on each failure.
func WasCanceled ¶
WasCanceled determines whether the given flow error was caused by cancellation.
Types ¶
type ErrorCleaner ¶
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.
type Graph ¶
type Graph struct {
// Clock is used to retrieve the current time.
Clock clock.Clock
// contains filtered or unexported fields
}
Graph is a builder for a Flow.
func (*Graph) Add ¶
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) AddGroup ¶
AddGroup adds all tasks of the given `TaskGroup` to the graph as regular tasks and returns their `TaskID`s so callers can wire them as dependencies of later `Add` calls.
The group is a build-time convenience only: the compiled `Flow` has no notion of groups — it only sees individual tasks connected by task-level dependencies. `AddGroup` flattens the group by:
- Attaching every group-level dependency to each task in the group, alongside any dependencies already declared on the task itself.
- Resolving a dependency that names another group (either the group value or its id) to the `TaskID`s of every task in that group.
- OR-ing the group's `SkipIf` condition onto each task's own `Task.SkipIf`.
- Namespacing task names with the group's alias when the group was renamed via `WithID`, so the same group can be added multiple times to the same graph without task-id collisions. Intra-group task-level dependencies are rewritten to the namespaced ids.
Tasks are added in the order they were supplied to the group. Callers must ensure intra-group dependencies are added before their dependents, just like when calling `Graph.Add` directly.
Because resolution happens at `AddGroup` time, groups referenced by id via `WithDependencies` must be added before the groups that depend on them; otherwise the id is treated as a plain task id, which then panics for lack of a matching task.
Like `Add`, this panics on duplicate task ids or unknown task-level dependencies.
type Opts ¶
type Opts struct {
// Log is used to log any output during flow execution.
Log logr.Logger
// 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 *errorsutils.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 NewCommandLineProgressReporter ¶
func NewCommandLineProgressReporter(out io.Writer) ProgressReporter
NewCommandLineProgressReporter returns a new progress reporter that writes the current status of the flow to the given output, when a SIGINFO is received (usually Ctrl+T). Any TaskFn wrapped with .RetryUntilTimeout() will have its retry errors surfaced here.
func NewDelayingProgressReporter ¶ added in v1.10.2
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 ¶ added in v1.10.2
func NewImmediateProgressReporter(reporterFn ProgressReporterFn) ProgressReporter
NewImmediateProgressReporter returns a new progress reporter with the given function.
type ProgressReporterFn ¶ added in v1.10.2
ProgressReporterFn is continuously called on progress in a flow.
type RetryReporter ¶
type RetryReporter interface {
ProgressReporter
// ReportRetry is called each time a task fails and will be retried.
ReportRetry(ctx context.Context, id TaskID, err error)
}
RetryReporter is an optional interface for ProgressReporter implementation to receive per-task retry notifications. If implemented, TaskFn.RetryUntilTimeout (and custom retry loops calling ReportRetry) surface errors
type Stats ¶
type Stats struct {
FlowName string
All TaskIDs
Succeeded TaskIDs
Failed TaskIDs
Running TaskIDs
Skipped TaskIDs
Pending TaskIDs
}
Stats are the statistics of a Flow execution.
func InitialStats ¶
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) ProgressPercent ¶
ProgressPercent retrieves the progress of a Flow execution in percent.
type Task ¶
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.
type TaskFn ¶
TaskFn is a payload function of a task.
func Parallel ¶
Parallel runs the given TaskFns in parallel, collecting their errors in a multierror.
func ParallelExitOnError ¶
ParallelExitOnError runs the given TaskFns in parallel and stops execution as soon as one TaskFn returns an error.
func ParallelN ¶ added in v1.86.0
ParallelN returns a function that runs the given TaskFns in parallel by spawning N workers, collecting their errors in a multierror. If N <= 0, then N will be defaulted to len(fns).
func Sequential ¶
Sequential runs the given TaskFns sequentially.
func (TaskFn) Recover ¶
Recover creates a new TaskFn that recovers an error with the given RecoverFn.
func (TaskFn) RetryUntilTimeout ¶
RetryUntilTimeout returns a TaskFn that is retried until the timeout is reached. On each failed attempt, ReportRetry is called automatically so that progress reporters can surface the last error without any additional wiring.
func (TaskFn) ToRecoverFn ¶
ToRecoverFn converts the TaskFn to a RecoverFn that ignores the incoming error.
type TaskGroup ¶
type TaskGroup struct {
// contains filtered or unexported fields
}
TaskGroup is a logical bundle of tasks that share a common set of dependencies.
A group has its own `id` which acts as an alias: other tasks or groups can depend on the group (via `WithDependencies`), and that dependency is expanded to a dependency on every task in the referenced group when the group is added to a `Graph` via `Graph.AddGroup`.
A `TaskGroup` value is safe to build up incrementally via the fluent `AddAll`, `WithDependencies` and `SkipIf` methods before being added to a `Graph`.
Tasks must be added in an order that respects intra-group task-level dependencies (dependencies before dependents), mirroring `Graph.Add`. `Graph.AddGroup` iterates tasks in the order they were added; an out-of-order dependency triggers the usual `g.Add` panic for the missing task.
func NewTaskGroup ¶
NewTaskGroup returns a new `TaskGroup` with the given id and the given initial tasks.
func (*TaskGroup) Add ¶
Add adds a single task to the group and returns its `TaskID`. It panics if a task with the same id is already present in the group.
func (TaskGroup) AddAll ¶
AddAll adds all given tasks to the group and returns the group for chaining. It panics if any of the tasks has an id that is already present in the group.
func (TaskGroup) SkipIf ¶
SkipIf marks every task in the group as skipped when the given condition evaluates to true. Multiple calls OR the conditions together, and per-task `Task.SkipIf` values are preserved: a task already flagged skipped stays skipped regardless of the group condition.
func (TaskGroup) TaskIDs ¶
TaskIDs returns the `TaskID`s of all tasks contained in the group. It makes `TaskGroup` satisfy the `TaskIDer` interface so it can itself be used as a dependency.
func (TaskGroup) WithDependencies ¶
WithDependencies records the given dependencies on the group and returns the group for chaining. Each dependency may reference either a regular task or another group's id — the actual expansion (group id → all tasks in that group) happens at `Graph.AddGroup` time.
func (TaskGroup) WithID ¶
WithID overrides the group's id and returns the group for chaining. Useful when the same task group is added under a different alias than the one baked in by its constructor, or when the same group is added multiple times to a graph — `Graph.AddGroup` namespaces contained task names with the alias so their ids remain unique in the graph.
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 ¶
NewTaskIDs returns a new set of TaskIDs initialized to contain all TaskIDs of the given TaskIDers.
func (TaskIDs) InsertIf ¶ added in v0.34.0
InsertIf inserts the TaskIDs of all TaskIDers into this TaskIDs if the given condition evaluates to true.
func (TaskIDs) List ¶
func (t TaskIDs) List() TaskIDSlice
List returns the elements of this in an ordered slice.
func (TaskIDs) StringList ¶
StringList returns the elements of this in an ordered string slice.
func (TaskIDs) UnsortedList ¶
func (t TaskIDs) UnsortedList() TaskIDSlice
UnsortedList returns the elements of this in an unordered slice.
func (TaskIDs) UnsortedStringList ¶
UnsortedStringList returns the elements of this in an unordered string slice.
Source Files
Directories