Documentation
¶
Overview ¶
Package jobs runs queued jobs from a store.JobStore through registered per-kind handlers, on a bounded background worker pool.
Index ¶
- Constants
- Variables
- type Handler
- type JobContext
- type Metrics
- type Reconciler
- type Reconcilers
- type Registry
- type Runner
- func (r *Runner) Cancel(id string) bool
- func (r *Runner) Notify()
- func (r *Runner) SetReconcilers(rec Reconcilers)
- func (r *Runner) SetVolumeTransferPool(kinds []string, workers int)
- func (r *Runner) Start(ctx context.Context)
- func (r *Runner) StartRetention(ctx context.Context, retention time.Duration)
- func (r *Runner) Wait()
Constants ¶
const DefaultWorkers = 8
DefaultWorkers is the worker-pool size when NewRunner is given workers <= 0. Raised from 4 to 8 to give plain jobs headroom: a parent evacuate occupies one worker for its whole fan-out, so a few concurrent evacuates could otherwise starve migrate/other jobs. This raises the starvation threshold; a structural fix (separate orchestration pool) remains a future option (#54).
Variables ¶
var VolumeTransferKinds = []string{"backup", "restore", "pitr-restore", "migrate", "evacuate"}
VolumeTransferKinds are the job kinds whose handlers call VolumeExport/VolumeImport (directly or via CopyVolume) on a worker's own claimed goroutine, and so can legitimately hold that worker for up to -volume-transfer-timeout (2h default): backup and restore stream a volume's contents directly; migrate does so via CopyVolume; evacuate runs its migrate children in-process on the parent's own claimed worker (internal/evacuate/handler.go's runChild), not through the job queue, so it's transitively volume-transfer-heavy for its whole fan-out. Intended for server.go to pass to SetVolumeTransferPool, structurally addressing the worker-pool starvation risk #237's 10min->2h deadline increase created (#238, following up on #54).
Functions ¶
This section is empty.
Types ¶
type Handler ¶
Handler executes one job of a given kind. It should honour ctx for cancellation and report progress via jc.Step. Returning a non-nil error fails the job.
type JobContext ¶
type JobContext struct {
// contains filtered or unexported fields
}
JobContext is the handler's progress channel back to the store.
func NewJobContext ¶
func NewJobContext(js store.JobStore, id string) *JobContext
NewJobContext builds a JobContext for a job id. Exposed so handlers can be exercised in tests without the full runner.
func (*JobContext) Step ¶
func (jc *JobContext) Step(step, detail string)
Step records a progress entry. It is best-effort: a store error is logged, not returned, so progress logging never fails the job.
type Metrics ¶
type Metrics interface {
JobStarted(kind string)
JobFinished(kind, result string)
ObserveDuration(kind string, d time.Duration)
JobEnqueued(kind string)
Rollback(kind string)
ChildFailure(kind string)
}
Metrics records job lifecycle events. Implementations must be safe for concurrent use. Runner.Metrics may be nil; a noop implementation is used in that case.
type Reconciler ¶
type Reconciler interface {
Reconcile(ctx context.Context, job store.Job, jc *JobContext) (state store.JobState, message string, resolved bool, err error)
}
Reconciler drives a job that was interrupted by a daemon restart toward a consistent state. It is registered per kind, parallel to Handler.
The implementation should honour ctx for cancellation so that a daemon shutdown does not hang the reconcile sweep.
resolved reports whether a terminal state was reached. When resolved is true, state must be a terminal state (JobSucceeded or JobFailed). resolved=false means the attempt was inconclusive (e.g. a host was unreachable) and the job should stay reconciling and be retried on the next sweep. A non-nil err is logged and treated as inconclusive.
message is the operator-facing text recorded in the job's error field for a terminal failed outcome; it should be empty for success.
type Reconcilers ¶
type Reconcilers map[string]Reconciler
Reconcilers maps a job kind to its reconciler.
type Runner ¶
type Runner struct {
Metrics Metrics // optional; nil-safe
// contains filtered or unexported fields
}
Runner drains queued jobs and dispatches them to handlers.
func NewRunner ¶
NewRunner builds a runner. workers <= 0 uses DefaultWorkers. The handler registry must not be modified after this call.
func (*Runner) Cancel ¶
Cancel signals an in-flight job to stop. Returns true if the job was found running — its handler context is cancelled and it will finish as canceled; false if no such job is currently running (queued/terminal jobs are handled by the caller via the store).
func (*Runner) Notify ¶
func (r *Runner) Notify()
Notify wakes a worker to check for new work; call after an Enqueue. Non-blocking. One Notify is sufficient for a batch of enqueues — a woken worker drains the queue exhaustively before sleeping again.
It broadcasts to all three poke channels (poke, pokeVolume, pokeGeneral) so that whichever pool configuration is in effect (unsplit, or split via SetVolumeTransferPool) gets a prompt wakeup rather than racing the pollInterval ticker. In the unsplit case pokeVolume/pokeGeneral are never read by any worker; each is buffer-1, so the first Notify() after a worker last drained it fills the buffer and every subsequent Notify() before that is harmless — not a leak, just a channel that stays "primed" and unread.
func (*Runner) SetReconcilers ¶
func (r *Runner) SetReconcilers(rec Reconcilers)
SetReconcilers registers the per-kind reconcilers used for boot recovery. Call before Start; the map must not be modified afterwards. A kind with a registered reconciler is moved to reconciling (and later resolved) on restart instead of being failed.
func (*Runner) SetVolumeTransferPool ¶ added in v1.0.36
SetVolumeTransferPool reserves workers of the runner's total pool for the given kinds, claiming ClaimNextMatching(kinds) instead of the shared ClaimNext. The remaining workers claim every other kind, registered or not (so an unregistered kind's jobs still fail fast in run() instead of being stranded queued forever). Call before Start. No-op if kinds is empty or workers <= 0 (matching SetReconcilers/SetVolumeTransferTimeout's convention) — the caller (server.go) separately validates workers < the runner's total worker count at startup.
#238: DefaultWorkers' single shared pool means a worker occupied by a VolumeExport/VolumeImport-heavy job (backup/restore/pitr-restore/migrate, or evacuate — which runs its migrate children in-process on its own claimed worker) can now hold that worker for up to -volume-transfer-timeout (2h default, was 10min pre-#237), starving every other job kind behind it. Reserving a small sub-pool for those kinds guarantees the general pool always has free workers regardless.
func (*Runner) Start ¶
Start reaps crash-interrupted jobs, then launches the worker pool. It returns immediately; the pool runs until ctx is cancelled. Use Wait to block for exit. It must be called exactly once.
func (*Runner) StartRetention ¶
StartRetention periodically prunes terminal jobs older than retention. It is a no-op when retention <= 0. It sweeps once immediately, then every retentionInterval, until ctx is cancelled. It is tracked by the runner's WaitGroup, so Wait blocks for it too.
Source Files