Documentation
¶
Overview ¶
Package dispatch launches pod-path task instances: it resolves a task's execution context, mints the agent's identity token, and routes the request to the executor. It implements scheduler.Dispatcher.
Index ¶
- Variables
- type BufferConfig
- type BufferedDispatcher
- type Dispatcher
- func (d *Dispatcher) Dispatch(ctx context.Context, runID, dagID, dagVersionID string, task domain.TaskSpec) (executor.Disposition, error)
- func (d *Dispatcher) SetAgentTLSCAConfigMap(name string)
- func (d *Dispatcher) SetAgentTokenTransport(transport, audience string, expirationSeconds int64)
- func (d *Dispatcher) SetAttemptLifetimeCeiling(ceiling time.Duration)
- func (d *Dispatcher) SetDefaultTaskServiceAccount(name string)
- func (d *Dispatcher) SetPlatformDefaults(p PlatformDefaults)
- func (d *Dispatcher) SetSecretsBackend(class, kwargs string)
- func (d *Dispatcher) SetSourceModeImage(image string)
- func (d *Dispatcher) SetTaskSecret(name, mountPath string)
- func (d *Dispatcher) SetUnitMisfitRecorder(r UnitMisfitRecorder)
- func (d *Dispatcher) SetWarmPlacer(p WarmPlacer)
- func (d *Dispatcher) WarmPolicy() domain.WarmPolicy
- type FailureSink
- type Inner
- type MetricsRecorder
- type PlatformDefaults
- type Resolved
- type Resolver
- type RetrySink
- type TokenIssuer
- type UnitMisfitRecorder
- type WarmPlacer
Constants ¶
This section is empty.
Variables ¶
var ErrAtCapacity = errors.New("dispatch buffer at capacity; will retry next tick")
ErrAtCapacity is returned by BufferedDispatcher.Dispatch when the buffered queue cannot accept another request. It travels with executor.Deferred: the scheduler leaves the TI scheduled so a later tick re-tries, without counting it as a failed dispatch. It is the backpressure signal that bounds tick latency under load (ADR 0031: tick rate decoupled from executor latency).
var ErrBufferWaitExceeded = errors.New("dispatch request waited in the buffer past its bound; re-offering")
ErrBufferWaitExceeded is the cause a worker hands to a RetrySink, with executor.Backpressure, for a request that waited in the buffer longer than BufferConfig.MaxWait. The task is re-offered instead of dispatched (#1607).
var ErrDrainTimeout = errors.New("drain timed out")
ErrDrainTimeout is returned by Close when workers did not finish within the drain timeout. Shutdown continues: blocking forever is worse, because the runtime then kills the process and no drain happens at all (#463).
Functions ¶
This section is empty.
Types ¶
type BufferConfig ¶
type BufferConfig struct {
BufferSize int
Workers int
// DrainTimeout bounds how long Close waits for workers to finish. Zero
// selects defaultDrainTimeout. Without a bound, one worker stuck in a
// remote call that never answers blocks shutdown until the runtime kills
// the process, which loses the drain entirely (#463).
DrainTimeout time.Duration
// MaxWait bounds how long a request may wait in the buffer before a worker
// takes it. The scheduler records the task queued, stamping queued_at, as
// soon as the buffer accepts it, so time spent in the buffer counts toward
// the dispatch-lost threshold; a request older than MaxWait when a worker
// takes it is not dispatched. When the sink is a RetrySink it is handed back
// as executor.Backpressure, which re-offers the task with no dispatch
// attempt charged and a fresh queued_at on its next queued episode (#1607).
// Zero selects half the default dispatch-lost threshold; a negative value
// disables the bound.
MaxWait time.Duration
}
BufferConfig sizes the BufferedDispatcher's worker pool. BufferSize=0 means "passthrough sync" (Lite mode, zero overhead): no goroutines spawned, no channel, the inner dispatcher is called inline. Any BufferSize>0 spawns max(Workers, 1) worker goroutines and a buffered channel of BufferSize slots.
type BufferedDispatcher ¶
type BufferedDispatcher struct {
// contains filtered or unexported fields
}
BufferedDispatcher fronts a synchronous Inner dispatcher with a bounded worker pool, so the scheduler tick is never blocked by a slow remote API call. ADR 0031: two-phase scheduler — planning sync, dispatch async.
func NewBuffered ¶
func NewBuffered(inner Inner, sink FailureSink, logger *slog.Logger, metrics MetricsRecorder, cfg BufferConfig) *BufferedDispatcher
NewBuffered constructs a BufferedDispatcher. BufferSize=0 returns a passthrough that is byte-for-byte equivalent to using the inner dispatcher directly (Lite path). BufferSize>0 spawns the worker pool.
func (*BufferedDispatcher) Close ¶
func (b *BufferedDispatcher) Close() error
Close stops accepting new dispatches, drains the in-flight queue (each buffered request is still processed by the inner dispatcher, or failed or re-offered via the sink, so no TI is left stuck `queued`), and waits for every worker to finish. Under mu so a concurrent Dispatch never sends on the closed channel. Idempotent.
The wait is bounded by cfg.DrainTimeout (#463). A worker sits in the inner dispatcher with a detached context, so a remote API that accepts the connection and never answers would otherwise block shutdown forever — the process then dies by SIGKILL and loses the drain #133 added. On expiry Close returns an error naming how many workers were abandoned; the caller logs it and proceeds with shutdown rather than hanging. The abandoned goroutines die with the process.
func (*BufferedDispatcher) Dispatch ¶
func (b *BufferedDispatcher) Dispatch(ctx context.Context, runID, dagID, dagVersionID string, task domain.TaskSpec) (executor.Disposition, error)
Dispatch hands a task off to the inner dispatcher. In passthrough mode the inner call happens inline. In buffered mode the request is enqueued non- blockingly: success returns (Dispatched, nil) immediately (the scheduler then records the TI as `queued`); a full or closed channel returns (Deferred, ErrAtCapacity). Deferred tells the scheduler nothing was attempted, so the TI stays scheduled for a later tick without touching its dispatch-attempt budget (review item S2: a full buffer is backpressure, not a dispatch failure).
type Dispatcher ¶
type Dispatcher struct {
// contains filtered or unexported fields
}
Dispatcher builds executor requests for queued pod-path tasks and runs them.
func NewDispatcher ¶
func NewDispatcher(exec executor.Executor, resolver Resolver, issuer TokenIssuer, controlAddr string, tokenTTL time.Duration) *Dispatcher
NewDispatcher builds a Dispatcher that launches tasks via exec, resolves their context with resolver, mints tokens with issuer (valid for tokenTTL), and tells the agent to reach the control plane at controlAddr.
func (*Dispatcher) Dispatch ¶
func (d *Dispatcher) Dispatch(ctx context.Context, runID, dagID, dagVersionID string, task domain.TaskSpec) (executor.Disposition, error)
Dispatch resolves the task, mints its agent token, and executes it. The executor classifies its own dispatch outcome and returns it as an executor.Disposition (ADR 0051 Phase 4). A dispatcher-INTERNAL failure that happens BEFORE Execute (task resolve, token mint) is permanent and so returns executor.Rejected — those bare errors classified as permanent before this change, so Rejected preserves the behavior exactly.
func (*Dispatcher) SetAgentTLSCAConfigMap ¶
func (d *Dispatcher) SetAgentTLSCAConfigMap(name string)
SetAgentTLSCAConfigMap configures the CA ConfigMap mounted into task pods so agents verify the control plane's gRPC TLS cert (issue #58). Empty = the agent stays on the insecure channel (dev).
func (*Dispatcher) SetAgentTokenTransport ¶
func (d *Dispatcher) SetAgentTokenTransport(transport, audience string, expirationSeconds int64)
SetAgentTokenTransport selects how the agent's bearer credential reaches the task pod (ADR 0055 Fix #3). transport is "" / "envvar" (the plaintext env-var default) or "exchange" (project a ServiceAccount token the agent exchanges for a task-scoped JWT). audience and expirationSeconds configure the projected token and are read only under the exchange transport. Ignored by the subprocess (Lite) executor, which has no pod.
func (*Dispatcher) SetAttemptLifetimeCeiling ¶
func (d *Dispatcher) SetAttemptLifetimeCeiling(ceiling time.Duration)
SetAttemptLifetimeCeiling wires the operator's attempt credential ceiling (auth.max_attempt_credential_lifetime) into every dispatched request, where the Kubernetes executor floors the pod's ActiveDeadlineSeconds with it when the task declares no execution timeout. A non-positive value is "no ceiling" and applies no floor; the subprocess executor ignores it.
func (*Dispatcher) SetDefaultTaskServiceAccount ¶
func (d *Dispatcher) SetDefaultTaskServiceAccount(name string)
SetDefaultTaskServiceAccount sets the ServiceAccount task pods run as when a DAG's task does not specify execution.service_account. Empty leaves pods on the namespace default SA (today's behavior). Wiring the chart's task ServiceAccount here closes the trap where creating the SA silently had no effect until every DAG also set execution.service_account.
func (*Dispatcher) SetPlatformDefaults ¶
func (d *Dispatcher) SetPlatformDefaults(p PlatformDefaults)
SetPlatformDefaults configures the per-cluster task defaults applied at dispatch to fill gaps the DAG artifact left empty (ADR 0023, layer L0).
func (*Dispatcher) SetSecretsBackend ¶
func (d *Dispatcher) SetSecretsBackend(class, kwargs string)
SetSecretsBackend configures the operator's external secrets backend (ADR 0060): the provider class the in-pod resolver drives and its raw kwargs JSON, delivered to the pod as operator-owned LEOFLOW_SECRETS_* env. Empty leaves external secrets off (the chain stays vault-only).
func (*Dispatcher) SetSourceModeImage ¶ added in v0.5.3
func (d *Dispatcher) SetSourceModeImage(image string)
SetSourceModeImage turns Pro source mode on for versions on image (ADR 0067 §3); "" leaves it off, today's behavior. Lite does not set it: the subprocess executor always runs from the source.
func (*Dispatcher) SetTaskSecret ¶
func (d *Dispatcher) SetTaskSecret(name, mountPath string)
SetTaskSecret configures a Kubernetes Secret mounted read-only into every task pod at mountPath, so tasks can read a credential (e.g. a GCP service-account key referenced by a connection's key_path) from the cluster's secret store rather than from Dexaflow (ADR 0035). Empty name = nothing mounted.
func (*Dispatcher) SetUnitMisfitRecorder ¶ added in v0.5.2
func (d *Dispatcher) SetUnitMisfitRecorder(r UnitMisfitRecorder)
SetUnitMisfitRecorder wires the counter for tolerated unit misfits.
func (*Dispatcher) SetWarmPlacer ¶
func (d *Dispatcher) SetWarmPlacer(p WarmPlacer)
SetWarmPlacer wires the warm-worker placement seam (ADR 0058 N1b1-place). With a placer set, Dispatch tries to place an admitted attempt on a free warm worker of its dag_version and only falls back to a dedicated pod on a warm miss. Leave it unset (nil) — the default — to keep dedicated pod-per-task, today's behavior.
func (*Dispatcher) WarmPolicy ¶ added in v0.5.4
func (d *Dispatcher) WarmPolicy() domain.WarmPolicy
WarmPolicy is the warm placement decision this dispatcher applies (ADR 0058): warm pools are on when a placer is wired, and a warm pod has the default task ServiceAccount, the resource unit or default limits, and no source. The register response reports each task's execution mode from the same policy, built from the same configuration (#1707).
type FailureSink ¶
type FailureSink interface {
MarkTaskDispatchFailed(ctx context.Context, runID, taskID, reason string) error
}
FailureSink lets a worker report that an asynchronously-dispatched task failed inside the inner dispatcher, so the scheduler can fail the TI with a clear reason. Without this callback a `queued` TI whose dispatch failed would sit forever (no reaper targets `queued`; ADR 0031 #128 only targets `running`).
type Inner ¶
type Inner interface {
Dispatch(ctx context.Context, runID, dagID, dagVersionID string, task domain.TaskSpec) (executor.Disposition, error)
}
Inner is the underlying synchronous dispatcher BufferedDispatcher wraps — matches scheduler.Dispatcher exactly so production wires through one type.
type MetricsRecorder ¶
type MetricsRecorder interface {
RecordDispatchQueueDepth(depth int)
RecordDispatchAtCapacity()
RecordDispatchLatencySeconds(seconds float64)
RecordDispatchInnerError()
}
MetricsRecorder records dispatch-pool observability signals.
type PlatformDefaults ¶
type PlatformDefaults struct {
// StagingSize/StagingStorageClass default the per-run staging volume when the
// DAG enabled staging but did not pin them (e.g. the cluster's RWX class).
StagingSize string
StagingStorageClass string
// StagingAccessMode is the PVC access mode for the staging volume (default
// ReadWriteMany; single-node dev uses ReadWriteOnce).
StagingAccessMode string
// Resources defaults a task's requests/limits when neither the task override
// nor the DAG set any.
Resources *domain.Resources
// Unit is the operator resource unit (executor.unit, ADR 0066). When set it
// replaces Resources: every task is sized from its pool_slots, and a task
// that declares more than pool_slots x unit is refused (or, under
// enforce: warn, run and counted). Nil: no unit.
Unit *domain.ResourceUnit
// Limits fills an empty CPU or memory limit on every task pod, after the task
// and L0 resources above are chosen (executor.defaults.limits_*, #537). Nil:
// no default limit. Not applied with a unit, which always sets both limits.
// Without a unit the same limits also size the warm pods (see WarmPolicy).
Limits *domain.DefaultLimits
// PodSecurity carries the task-pod hardening choices. It lives here, not in
// the DAG spec, on purpose: whether untrusted task code may run as root is a
// cluster-operator decision. Exposing it per-DAG would let an author elevate
// their own task, which is the same self-service escalation as picking an
// arbitrary service_account. It is populated from the cluster's configured
// defaults (runAsNonRoot on), which an operator can override.
PodSecurity executor.PodSecurity
// ExecutionPolicy is the operator's allowlist of task execution fields
// (executor.defaults.execution_allowlist, issue #629). Registration refuses a
// DAG that breaks it; dispatch refuses a task of a DAG registered before the
// list changed. Like PodSecurity it is cluster policy, never a DAG field.
// Nil allows everything.
ExecutionPolicy *domain.ExecutionPolicy
}
PlatformDefaults are per-cluster task defaults applied at dispatch to fill gaps the DAG artifact left empty (ADR 0023, layer L0). They are the lowest precedence (task override > DAG default > platform default) and never replace a value baked into dag.json, so the artifact stays portable across clusters.
type Resolved ¶
type Resolved struct {
TaskInstanceID string
TenantID string
Image string
ImagePullPolicy string
TryNumber int
// AttemptEpoch identifies this execution attempt of the row (ADR 0051
// amendment). The resolver claims a fresh value on every dispatch, so two
// dispatches of one try never share it, even with no reset rail between
// them. The agent token (A2) and the pod's attempt-epoch label (A4) are
// minted from it.
AttemptEpoch int
// Staging carries the DAG's opt-in staging-volume config (ADR 0022); nil or
// disabled means no per-run volume.
Staging *domain.StagingConfig
// Source is the dag.py text captured at compile time, threaded to the
// SubprocessExecutor so Lite tasks can importlib their DAG without relying
// on a globally-correct workdir. Empty for Pro (the container image carries
// the source) and ignored by the KubernetesExecutor.
Source string
}
Resolved is the execution context the dispatcher needs to launch a task.
type Resolver ¶
type Resolver interface {
ResolveTask(ctx context.Context, runID, taskID string) (Resolved, error)
}
Resolver loads a task instance's execution context from storage.
type RetrySink ¶ added in v0.5.1
type RetrySink interface {
FailureSink
HandleDispatchFailure(ctx context.Context, runID, taskID string, disp executor.Disposition, cause error) error
}
RetrySink is a FailureSink that can also re-offer a task whose dispatch failed in a worker, given the disposition the inner dispatcher classified the error as. When the sink implements it, a worker failure is handled the way the scheduler handles a synchronous one (scheduler.AsyncDispatchFailures): backpressure and transient errors put the task back to scheduled with a backoff instead of failing it. A plain FailureSink keeps failing the task.
type TokenIssuer ¶
type TokenIssuer interface {
IssueAgentToken(id auth.AgentIdentity, ttl time.Duration) (string, error)
}
TokenIssuer mints a per-task-instance agent token.
type UnitMisfitRecorder ¶ added in v0.5.2
type UnitMisfitRecorder interface {
RecordUnitMisfit(stage string)
}
UnitMisfitRecorder counts a task that does not fit the resource unit but is let through under executor.unit.enforce=warn, by stage ("register" or "dispatch"). observability.Metrics satisfies it.
type WarmPlacer ¶
type WarmPlacer interface {
Assign(dagVersionID string, a *agentv1.WorkAssignment) bool
}
WarmPlacer hands a per-attempt WorkAssignment to a free warm worker of a dag_version, returning false when none is free (ADR 0058 N1b1-place). It is a narrow structural view of the agentrpc worker registry: the executor package must not import agentrpc, so the seam lives here and main.go passes the registry, which satisfies it. A nil WarmPlacer on the Dispatcher means warm pools are off — every task takes the dedicated pod path, today's behavior.
Source Files