Documentation
¶
Overview ¶
Package pipeline implements stage-based execution for stream pipelines. Operators are grouped into stages that execute as direct function calls; bounded edges between stages provide backpressure: a full edge blocks the upstream stage, and the pressure propagates back to the source.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func SampleEdges ¶
func SampleEdges(stop <-chan struct{}, edges []*Edge)
SampleEdges starts the edge gauge sampler in a goroutine; it stops when stop is closed.
Types ¶
type ChannelStage ¶
type ChannelStage struct {
Op operator.Operator
Label string
StageName string // unique name assigned by the planner
}
ChannelStage runs a single channel-based operator that is neither a SingleProcessor nor part of a keyed stage (e.g. Window or Reduce used without KeyBy). It preserves the old per-operator execution model for that operator alone.
func (*ChannelStage) Name ¶
func (s *ChannelStage) Name() string
type Edge ¶
Edge is a bounded channel connecting two stages. A full edge blocks the upstream stage — that blocking is the backpressure mechanism.
type KeyedStage ¶
type KeyedStage struct {
KeyBy *operator.KeyByOperator
// StageName is the unique name assigned by the planner (metrics label).
StageName string
// contains filtered or unexported fields
}
KeyedStage executes stateful operators (Window, Reduce) with keyed parallelism:
┌── worker 0: Op₀_clone → Op₁_clone ──┐
in ── Router ── worker 1: Op₀_clone → Op₁_clone ──┼── alignedMerge → out
└── worker N: Op₀_clone → Op₁_clone ──┘
Each worker owns a clone of every stateful operator with an isolated state backend. The router hash-dispatches data records (same key → same worker) and broadcasts barriers/watermarks to all workers; the merge emits each marker exactly once after all workers deliver it (C1).
func NewKeyedStage ¶
func NewKeyedStage(kb *operator.KeyByOperator, ops []operator.Operator, hooks StageHooks) (*KeyedStage, error)
NewKeyedStage clones the stateful operator chain once per partition. hooks.OnClone registers every clone for checkpoint restore and returns its global index; hooks.OnSnapshot receives each clone's state captured synchronously when a barrier passes through it — the race-free snapshot point (snapshotting later, at the end of the pipeline, races with the clone processing post-barrier records); hooks.StateBackendFor injects each clone's state backend, keyed by the same "worker-<idx>" owner ID the checkpoint format uses.
func (*KeyedStage) Name ¶
func (s *KeyedStage) Name() string
type PlanConfig ¶
type PlanConfig struct {
Source source.Source
Operators []operator.Operator
Labels []string // metric label per operator, aligned with Operators
Sink sink.Sink
// DrainTimeout bounds the source offset flush on shutdown.
DrainTimeout time.Duration
// StageHooks are forwarded to the stages the planner builds.
StageHooks
}
PlanConfig is the input to BuildPlan.
type SinkStage ¶
type SinkStage struct {
Sink sink.Sink
// OnBarrier, when set (uncoordinated checkpointing), is invoked
// for each barrier only after the sink has drained every record
// ahead of it — so a checkpoint is never durable for records the
// sink never received. Coordinated (exactly-once) pipelines leave
// this nil; the sink itself acknowledges barriers there.
OnBarrier func(checkpointID string)
}
SinkStage wraps a sink.Sink as the last stage of a plan.
func (*SinkStage) Run ¶
func (s *SinkStage) Run(runCtx, hardCtx context.Context, in <-chan types.Record, _ chan<- types.Record) error
Run pumps records from the input edge into the sink. The sink gets hardCtx so it keeps draining through graceful shutdown and is only interrupted by the shutdown timeout (C3).
type SourceStage ¶
type SourceStage struct {
Source source.Source
// DrainTimeout bounds the source's offset flush on shutdown.
DrainTimeout time.Duration
}
SourceStage wraps a source.Source as the first stage of a plan.
func (*SourceStage) Name ¶
func (s *SourceStage) Name() string
func (*SourceStage) Run ¶
func (s *SourceStage) Run(runCtx, hardCtx context.Context, _ <-chan types.Record, out chan<- types.Record) error
Run starts the source and forwards its records to the output edge. Source errors are reported via metrics/log but do not fail the pipeline (parity with previous behavior — the stream simply ends).
type Stage ¶
type Stage interface {
Name() string
Run(runCtx, hardCtx context.Context, in <-chan types.Record, out chan<- types.Record) error
}
Stage is one execution unit of a pipeline. Stages run concurrently, connected by bounded edges.
Contract:
- Run consumes in until it is closed and closes out before returning (C2: the writer owns the close). The wiring never closes edges.
- runCtx is the caller's context: only the source stage reacts to it, by stopping production. Downstream stages drain via the cascading channel closes that follow (C3).
- hardCtx aborts blocked sends/reads; it fires only on shutdown-timeout expiry or a fatal error in another stage.
func BuildPlan ¶
func BuildPlan(cfg PlanConfig) ([]Stage, error)
BuildPlan groups the flat operator list into execution stages:
- consecutive stateless operators (SingleProcessor) with the same parallelism share one StatelessStage
- a KeyBy router starts a KeyedStage that consumes the following Cloneable (stateful) operators
- any other channel-based operator gets its own ChannelStage
- SourceStage and SinkStage bracket the plan
The returned stages are wired with one bounded edge between each consecutive pair.
type StageHooks ¶
type StageHooks struct {
// OnClone is called for every stateful operator clone a keyed
// stage creates, so the caller can register it for checkpointing.
// It returns the clone's global worker index ("worker-<idx>" in
// checkpoint data).
OnClone func(operator.Operator) int
// OnSnapshot receives operator state captured synchronously when
// a barrier passes through a stateful operator (the race-free
// snapshot point). key is the checkpoint-data key ("worker-<idx>"
// or "op-<idx>").
OnSnapshot func(checkpointID, key string, snapshot []byte)
// StateBackendFor creates the state backend for a stateful
// operator instance, keyed by its owner ID ("op-<i>" or
// "worker-<idx>"). Nil means operators keep their self-created
// default (in-memory) backends.
StateBackendFor func(ownerID string) (state.StateBackend, error)
// NativeStateDir returns the directory a Checkpointable backend
// should hard-link its barrier-time checkpoint into, for a given
// (checkpoint ID, owner). Nil when checkpointing is off or the
// storage has no state directory support.
NativeStateDir func(checkpointID, ownerID string) string
}
StageHooks are engine callbacks wired into stateful stages.
type StatelessStage ¶
type StatelessStage struct {
StageName string
Ops []operator.Operator // every op must implement operator.SingleProcessor
Labels []string // metric label per op, aligned with Ops
Parallelism int
}
StatelessStage executes a chain of stateless operators (Map, Filter, FlatMap, Process) as direct function calls — no channels between operators. With Parallelism > 1, N workers share the input via a round-robin dispatcher; barriers and watermarks are broadcast to all workers and re-aligned at the merge (C1). Record order across workers is not preserved when Parallelism > 1.
func (*StatelessStage) Name ¶
func (s *StatelessStage) Name() string
Source Files