daemon/internal/jobs: Add the jobs manager and scheduler
Add the lifecycle core of the jobs extension. Registration is idempotent on a canonical spec hash, so formatting and defaults do not change a job's identity, and a spec from a newer contract fails validation on an older daemon. Create-and-run fires the first run before anything else can reach the new job. Container specs with a negative StopTimeout are rejected, as runs must be stoppable, and so is the host-dependent Local timezone. Every fire goes through one chokepoint serialized per job, which applies the concurrency policy and persists the run record before the container is created, so a crash always leaves a discoverable run. A fire is detached from its caller, as the daemon completes a container start even when the request goes away. A run whose start is unknown, because the start request failed or its record could not be written, has its container stopped and keeps the job occupied until the container exits. Outcomes are persisted before auto-removal, and failed terminal or idle writes are retried instead of leaving the job stuck; only the job's latest run returns it to idle. Cancels and timeouts always stop the container, even after the other decided the outcome, so a failed stop is not final. The scheduler is one goroutine over a min-heap, dispatching through the same chokepoint. On start, missed occurrences follow the missed-fires policy with at most one catch-up, and an occurrence moved by a reschedule is kept. A reschedule replaces exactly one occurrence, including one a tick already consumed but has not fired yet. A dispatched fire is dropped if its job was paused, resumed or removed since, and the next-fire time is persisted from the scheduler's current state after the fire is recorded, so a delayed fire cannot undo a newer change and a crash cannot lose the occurrence. A fire deferred by the queue policy keeps its occurrence recorded until it runs, so a restart in between catches it up. Removing a job keeps its run history by default, reachable by the removed job's ID, and run containers are removed with their records. Stored specs are compared as the running daemon canonicalizes them, so jobs survive upgrades that change the container API types. Co-authored-by: Paweł Gronowski <pawel.gronowski@docker.com> Signed-off-by: Guillaume Lours <glours@users.noreply.github.com> Signed-off-by: Paweł Gronowski <pawel.gronowski@docker.com>
This commit is contained in:
Guillaume LoursandPaweł Gronowski authored and Paweł Gronowski committed
1 parent
093882dc71
commit
333776c7cf
6 files changed
+3886
No files matched your search
@@ -0,0 +1,722 @@
|
||||
package jobs
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"math"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
cerrdefs "github.com/containerd/errdefs"
|
||||
"github.com/containerd/log"
|
||||
"github.com/moby/locker"
|
||||
"github.com/moby/moby/api/types/container"
|
||||
"github.com/moby/moby/api/types/network"
|
||||
"github.com/moby/moby/v2/daemon/internal/stringid"
|
||||
"github.com/moby/moby/v2/daemon/names"
|
||||
"github.com/moby/moby/v2/errdefs"
|
||||
jobsv0 "github.com/moby/moby/v2/extpoints/jobs/api/v0"
|
||||
)
|
||||
|
||||
// Backend is the container surface the manager drives runs with.
|
||||
type Backend interface {
|
||||
ContainerCreate(ctx context.Context, req ContainerCreateRequest) (container.CreateResponse, error)
|
||||
ContainerStart(ctx context.Context, id string) error
|
||||
// ContainerStop stops a container using its configured stop signal and
|
||||
// timeout.
|
||||
ContainerStop(ctx context.Context, id string) error
|
||||
ContainerRm(ctx context.Context, id string) error
|
||||
// ContainerWait resolves on the container's final exit, after its
|
||||
// restart policy is exhausted. It fails only when the container does not
|
||||
// exist; other failures to observe it are retried until ctx is done.
|
||||
// The returned channel delivers exactly one status unless ctx is done
|
||||
// first; it must never be closed without delivering it.
|
||||
// If the container has already exited, the status is ready on return.
|
||||
ContainerWait(ctx context.Context, id string) (<-chan StateStatus, error)
|
||||
}
|
||||
|
||||
// ContainerCreateRequest describes a run container to create.
|
||||
type ContainerCreateRequest struct {
|
||||
Name string
|
||||
Config *container.Config
|
||||
HostConfig *container.HostConfig
|
||||
NetworkingConfig *network.NetworkingConfig
|
||||
}
|
||||
|
||||
// StateStatus reports a container's final exit, as delivered by
|
||||
// [Backend.ContainerWait].
|
||||
type StateStatus interface {
|
||||
ExitCode() int
|
||||
Err() error
|
||||
}
|
||||
|
||||
// NameGenerator produces a default job name. Retry counts the prior
|
||||
// generated names that collided, so implementations can vary their answer.
|
||||
type NameGenerator func(ctx context.Context, retry int64) (string, error)
|
||||
|
||||
// Reserved labels applied to every run container, correlating it with its
|
||||
// job and run records across the container API.
|
||||
const (
|
||||
// LabelJobID is the ID of the job that owns the container.
|
||||
LabelJobID = "com.docker.job.id"
|
||||
// LabelRunID is the ID of the run the container backs.
|
||||
LabelRunID = "com.docker.job.run-id"
|
||||
)
|
||||
|
||||
// nameGenerationRetries bounds the attempts at generating an unused random
|
||||
// job name, mirroring the daemon's own retry bound for container names.
|
||||
const nameGenerationRetries = 6
|
||||
|
||||
// Manager owns the job lifecycle: registration with spec-hash idempotency,
|
||||
// run execution through the single tryFireLocked chokepoint, and the queries
|
||||
// the extension API exposes. Persistence is delegated to the Store, container
|
||||
// work to the Backend.
|
||||
type Manager struct {
|
||||
store *Store
|
||||
backend Backend
|
||||
// generateName produces default job names for unnamed registrations.
|
||||
generateName NameGenerator
|
||||
// locks serializes state transitions per job ID: every fire, cancel,
|
||||
// removal and run-completion transition takes the job's lock, so two
|
||||
// triggers racing on the same job cannot both pass the concurrency
|
||||
// policy.
|
||||
locks *locker.Locker
|
||||
// now is the manager's clock, injectable in tests.
|
||||
now func() time.Time
|
||||
// validateCron validates a cron expression at registration time; a
|
||||
// field so tests can substitute it.
|
||||
validateCron func(expr string) error
|
||||
|
||||
mu sync.Mutex
|
||||
// queued holds at most one deferred fire per job (the queue concurrency
|
||||
// policy has depth one); fired when the current run completes.
|
||||
queued map[string]*jobsv0.TriggerEvidence
|
||||
// overrides forces a run's terminal state (cancelled or timed_out)
|
||||
// ahead of its container exit; first writer wins.
|
||||
overrides map[string]string
|
||||
// timers holds the timeout timer of each in-flight run.
|
||||
timers map[string]*time.Timer
|
||||
// notify carries one-shot subscriptions per job ID, closed on every
|
||||
// job or run transition; Wait re-checks its condition on each signal.
|
||||
notify map[string][]chan struct{}
|
||||
// background tracks every goroutine the manager spawns (run watchers,
|
||||
// queued fires, container stops, the scheduler loop) so Shutdown can
|
||||
// drain them. Goroutines spawned by a tracked goroutine register before
|
||||
// their parent finishes, so the counter cannot reach zero while work is
|
||||
// still being spawned.
|
||||
background sync.WaitGroup
|
||||
// sched fires schedule jobs; its loop runs between Start and Shutdown,
|
||||
// but arming works from construction so jobs registered before Start
|
||||
// are picked up when the loop begins.
|
||||
sched *scheduler
|
||||
started bool
|
||||
stopped bool
|
||||
// stop is closed by Shutdown so run watchers detach from containers
|
||||
// that are not exiting — under live-restore the daemon shuts down while
|
||||
// containers deliberately keep running, and waiting on them would stall
|
||||
// (or, with an unbounded shutdown timeout, deadlock) the daemon's exit.
|
||||
stop chan struct{}
|
||||
}
|
||||
|
||||
// NewManager builds a manager on top of a loaded store, generating default
|
||||
// job names with generateName. The constructor does not reconcile runs
|
||||
// orphaned by an unclean stop — callers run Restore before Start, as the
|
||||
// extension's activation does.
|
||||
func NewManager(store *Store, backend Backend, generateName NameGenerator) *Manager {
|
||||
m := &Manager{
|
||||
store: store,
|
||||
backend: backend,
|
||||
generateName: generateName,
|
||||
locks: locker.New(),
|
||||
now: time.Now,
|
||||
validateCron: func(expr string) error {
|
||||
_, err := parseCron(expr)
|
||||
return err
|
||||
},
|
||||
queued: make(map[string]*jobsv0.TriggerEvidence),
|
||||
overrides: make(map[string]string),
|
||||
timers: make(map[string]*time.Timer),
|
||||
notify: make(map[string][]chan struct{}),
|
||||
stop: make(chan struct{}),
|
||||
}
|
||||
m.sched = newScheduler(m)
|
||||
store.onRunsDropped = m.removeRunContainers
|
||||
|
||||
return m
|
||||
}
|
||||
|
||||
// Start arms every registered schedule job and starts the scheduler loop.
|
||||
// Occurrences missed while the daemon was down — the persisted next-fire
|
||||
// time is in the past — follow each job's missed-fires policy: one fires a
|
||||
// single catch-up run, skip drops them; either way the schedule then re-arms
|
||||
// from the next occurrence, never replaying the backlog.
|
||||
// A persisted occurrence that a reschedule moved past the upcoming one is
|
||||
// kept.
|
||||
func (m *Manager) Start(ctx context.Context) {
|
||||
m.mu.Lock()
|
||||
if m.started {
|
||||
m.mu.Unlock()
|
||||
return
|
||||
}
|
||||
m.started = true
|
||||
m.mu.Unlock()
|
||||
|
||||
now := m.now()
|
||||
for _, job := range m.store.Jobs() {
|
||||
if triggerKind(job.Spec.Trigger) != jobsv0.TriggerKindSchedule || job.Paused {
|
||||
continue
|
||||
}
|
||||
trigger := job.Spec.Trigger.Schedule
|
||||
schedule, loc, err := parseScheduleTrigger(trigger)
|
||||
if err != nil {
|
||||
// Specs are validated at registration; only a store record from
|
||||
// a buggy or newer daemon gets here. Leave the job disarmed
|
||||
// rather than fail the whole startup.
|
||||
log.G(ctx).WithError(err).WithFields(log.Fields{"job": job.ID}).Warn("leaving job with unparseable schedule disarmed")
|
||||
// The record must not keep advertising an occurrence that will
|
||||
// never fire.
|
||||
m.persistNextFire(ctx, job.ID)
|
||||
|
||||
continue
|
||||
}
|
||||
// A job still running has its run re-attached by reconciliation,
|
||||
// which precedes Start: its catch-up goes through the concurrency
|
||||
// policy like any other fire.
|
||||
missed := job.NextFireAtNano
|
||||
catchUp := missed > 0 && !time.Unix(0, missed).After(now) &&
|
||||
(trigger.MissedFires == "" || trigger.MissedFires == jobsv0.MissedFiresOne) &&
|
||||
!m.firedFor(job.ID, missed)
|
||||
|
||||
if next, ok := schedule.next(now, loc); ok {
|
||||
// A reschedule persisted the occurrence after the upcoming one;
|
||||
// keep it rather than fire the occurrence it skipped. Only that
|
||||
// exact occurrence is kept, so a clock stepped back cannot
|
||||
// postpone the job any further.
|
||||
if job.NextFireAtNano > next.UnixNano() {
|
||||
if after, ok := schedule.next(next, loc); ok && after.UnixNano() == job.NextFireAtNano {
|
||||
next = after
|
||||
}
|
||||
}
|
||||
|
||||
m.sched.arm(job.ID, schedule, loc, next)
|
||||
}
|
||||
|
||||
if catchUp {
|
||||
// The catch-up records the next occurrence once it fired, so a
|
||||
// crash before then leaves it to the next startup.
|
||||
m.background.Go(func() { m.fireScheduled(ctx, job.ID, time.Unix(0, missed)) })
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
m.persistNextFire(ctx, job.ID)
|
||||
}
|
||||
m.background.Go(func() { m.sched.run(ctx) })
|
||||
}
|
||||
|
||||
// firedFor reports whether the job's latest run was fired for the scheduled
|
||||
// occurrence: a fire is recorded before the next occurrence is, so a crash in
|
||||
// between leaves a missed occurrence that already has its run.
|
||||
func (m *Manager) firedFor(jobID string, scheduledAtNano int64) bool {
|
||||
run, err := m.store.LatestRun(jobID)
|
||||
return err == nil && run.Trigger != nil && run.Trigger.ScheduledAtNano == scheduledAtNano
|
||||
}
|
||||
|
||||
// parseScheduleTrigger resolves a validated schedule trigger into its parsed
|
||||
// expression and timezone (empty means UTC).
|
||||
func parseScheduleTrigger(trigger *jobsv0.ScheduleTrigger) (*cronSchedule, *time.Location, error) {
|
||||
schedule, err := parseCron(trigger.Cron)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
loc := time.UTC
|
||||
if trigger.Timezone != "" {
|
||||
if loc, err = time.LoadLocation(trigger.Timezone); err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
}
|
||||
return schedule, loc, nil
|
||||
}
|
||||
|
||||
// Shutdown waits for in-flight run watchers to finish recording their
|
||||
// outcomes, or for the context to expire. Callers must stop issuing API
|
||||
// calls before invoking Shutdown: a call racing the drain could register
|
||||
// new background work while the wait group is being awaited, which
|
||||
// sync.WaitGroup forbids. When the context expires first, the internal
|
||||
// waiter goroutine lingers until the background work eventually drains.
|
||||
func (m *Manager) Shutdown(ctx context.Context) error {
|
||||
m.mu.Lock()
|
||||
if !m.stopped {
|
||||
m.stopped = true
|
||||
close(m.stop)
|
||||
if m.started {
|
||||
close(m.sched.stop)
|
||||
}
|
||||
// Disarm run timeouts: a deadline landing in the shutdown window
|
||||
// would stop a container that live-restore deliberately keeps
|
||||
// running.
|
||||
for runID, timer := range m.timers {
|
||||
timer.Stop()
|
||||
delete(m.timers, runID)
|
||||
}
|
||||
}
|
||||
m.mu.Unlock()
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
m.background.Wait()
|
||||
close(done)
|
||||
}()
|
||||
select {
|
||||
case <-done:
|
||||
return nil
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
// Create registers a job without starting a run. It is idempotent on the
|
||||
// canonical spec hash: re-submitting an identical name+spec returns the
|
||||
// existing job with created=false, and a name collision with a different
|
||||
// spec is a conflict carrying both hashes.
|
||||
func (m *Manager) Create(ctx context.Context, name string, spec *jobsv0.JobSpec) (_ *jobsv0.Job, created bool, _ error) {
|
||||
return m.create(ctx, name, spec, nil)
|
||||
}
|
||||
|
||||
// create is Create, calling onCreated with a newly registered job's ID
|
||||
// before releasing the job's lock.
|
||||
func (m *Manager) create(ctx context.Context, name string, spec *jobsv0.JobSpec, onCreated func(jobID string)) (_ *jobsv0.Job, created bool, _ error) {
|
||||
if err := validateJobName(name); err != nil {
|
||||
return nil, false, errdefs.InvalidParameter(err)
|
||||
}
|
||||
decoded, err := m.validateSpec(spec)
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
hash, err := specHash(spec, decoded)
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
|
||||
// Registering a schedule job arms its trigger; the next occurrence is
|
||||
// computed up front so the returned job already advertises it, and an
|
||||
// expression with no future occurrence is rejected rather than silently
|
||||
// registering a job that would never fire.
|
||||
var (
|
||||
schedule *cronSchedule
|
||||
loc *time.Location
|
||||
next time.Time
|
||||
)
|
||||
if spec.Trigger != nil && spec.Trigger.Schedule != nil {
|
||||
if schedule, loc, err = parseScheduleTrigger(spec.Trigger.Schedule); err != nil {
|
||||
return nil, false, errdefs.InvalidParameter(err) // unreachable after validateSpec, kept for defense
|
||||
}
|
||||
var ok bool
|
||||
if next, ok = schedule.next(m.now(), loc); !ok {
|
||||
return nil, false, errdefs.InvalidParameter(fmt.Errorf("cron expression %q has no future occurrence", spec.Trigger.Schedule.Cron))
|
||||
}
|
||||
}
|
||||
|
||||
job := &jobsv0.Job{
|
||||
ID: stringid.GenerateRandomID(),
|
||||
Name: name,
|
||||
Spec: spec,
|
||||
SpecHash: hash,
|
||||
State: jobsv0.JobStateIdle,
|
||||
CreatedAtNano: m.now().UnixNano(),
|
||||
UpdatedAtNano: m.now().UnixNano(),
|
||||
}
|
||||
if schedule != nil {
|
||||
job.NextFireAtNano = next.UnixNano()
|
||||
}
|
||||
// Hold the new job's lock until its trigger is armed: a removal by name
|
||||
// landing between the store write and arming would otherwise leave the
|
||||
// trigger armed for a deleted job.
|
||||
m.locks.Lock(job.ID)
|
||||
defer m.locks.Unlock(job.ID)
|
||||
|
||||
if err := m.store.CreateJob(ctx, job); err != nil {
|
||||
// A name conflict is idempotent when the stored spec matches;
|
||||
// re-read under the store's authority rather than trusting a
|
||||
// pre-flight lookup, so concurrent creates converge.
|
||||
if cerrdefs.IsConflict(err) {
|
||||
existing, lookupErr := m.store.JobByName(name)
|
||||
if lookupErr != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
|
||||
existingHash := currentSpecHash(existing)
|
||||
if existingHash == hash {
|
||||
return m.composeJob(existing), false, nil
|
||||
}
|
||||
|
||||
return nil, false, cerrdefs.ErrAlreadyExists.WithMessage(fmt.Sprintf("job name %s is already in use by a different spec (stored %s, submitted %s)", name, existingHash, hash))
|
||||
}
|
||||
return nil, false, err
|
||||
}
|
||||
if schedule != nil {
|
||||
m.sched.arm(job.ID, schedule, loc, next)
|
||||
}
|
||||
|
||||
if onCreated != nil {
|
||||
onCreated(job.ID)
|
||||
}
|
||||
|
||||
return m.composeJob(job), true, nil
|
||||
}
|
||||
|
||||
// CreateAndRun atomically resolves or registers a manual job and fires a
|
||||
// run, implementing the name/spec resolution matrix of the API contract.
|
||||
func (m *Manager) CreateAndRun(ctx context.Context, name string, spec *jobsv0.JobSpec) (_ *jobsv0.Job, _ *jobsv0.Run, created bool, _ error) {
|
||||
if name == "" && spec == nil {
|
||||
return nil, nil, false, errdefs.InvalidParameter(errors.New("at least one of name and spec is required"))
|
||||
}
|
||||
if spec != nil && triggerKind(spec.Trigger) != jobsv0.TriggerKindManual {
|
||||
return nil, nil, false, errdefs.InvalidParameter(errors.New("create-and-run serves manual jobs only; register schedule jobs with create"))
|
||||
}
|
||||
|
||||
var (
|
||||
job *jobsv0.Job
|
||||
run *jobsv0.Run
|
||||
runErr error
|
||||
)
|
||||
// A newly registered job's first run fires before the job's lock is
|
||||
// released, so nothing can run or remove the job in between.
|
||||
fire := func(jobID string) { run, runErr = m.fireManualLocked(ctx, jobID) }
|
||||
|
||||
switch {
|
||||
case name == "":
|
||||
// No name: always a fresh job under a generated name.
|
||||
var err error
|
||||
if job, err = m.createNamed(ctx, spec, fire); err != nil {
|
||||
return nil, nil, false, err
|
||||
}
|
||||
created = true
|
||||
case spec == nil:
|
||||
// The request field is a name, strictly: the create branches treat
|
||||
// it as one, so resolving IDs here would be an observable asymmetry.
|
||||
existing, err := m.store.JobByName(name)
|
||||
if err != nil {
|
||||
return nil, nil, false, err
|
||||
}
|
||||
if triggerKind(existing.Spec.Trigger) != jobsv0.TriggerKindManual {
|
||||
return nil, nil, false, errdefs.InvalidParameter(fmt.Errorf("job %s is not a manual job", name))
|
||||
}
|
||||
job = existing
|
||||
default:
|
||||
var err error
|
||||
if job, created, err = m.create(ctx, name, spec, fire); err != nil {
|
||||
return nil, nil, false, err
|
||||
}
|
||||
}
|
||||
|
||||
if run == nil && runErr == nil {
|
||||
run, runErr = m.fireManual(ctx, job.ID)
|
||||
}
|
||||
|
||||
if runErr != nil {
|
||||
return nil, nil, false, runErr
|
||||
}
|
||||
|
||||
job, err := m.Inspect(ctx, job.ID)
|
||||
if err != nil {
|
||||
return nil, nil, false, err
|
||||
}
|
||||
return job, run, created, nil
|
||||
}
|
||||
|
||||
// createNamed registers spec under a generated name, retrying on collisions
|
||||
// like the daemon does for container names. A random name landing on an
|
||||
// existing job with an identical spec also counts as a collision: adopting
|
||||
// that job would silently hand out foreign history.
|
||||
func (m *Manager) createNamed(ctx context.Context, spec *jobsv0.JobSpec, onCreated func(jobID string)) (*jobsv0.Job, error) {
|
||||
var lastErr error
|
||||
for i := range nameGenerationRetries {
|
||||
name, err := m.generateName(ctx, int64(i))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("generating a job name: %w", err)
|
||||
}
|
||||
|
||||
job, created, err := m.create(ctx, name, spec, onCreated)
|
||||
switch {
|
||||
case err == nil && created:
|
||||
return job, nil
|
||||
case err == nil:
|
||||
lastErr = fmt.Errorf("generated name %s is already in use", name)
|
||||
case cerrdefs.IsAlreadyExists(err) || cerrdefs.IsConflict(err):
|
||||
lastErr = err
|
||||
default:
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return nil, fmt.Errorf("generating a unique job name: %w", lastErr)
|
||||
}
|
||||
|
||||
// resolveJob resolves a job reference, trying IDs first and names second,
|
||||
// like the container API does.
|
||||
func (m *Manager) resolveJob(ref string) (*jobsv0.Job, error) {
|
||||
if ref == "" {
|
||||
return nil, errdefs.InvalidParameter(errors.New("job reference must not be empty"))
|
||||
}
|
||||
job, err := m.store.Job(ref)
|
||||
if err == nil {
|
||||
return job, nil
|
||||
}
|
||||
return m.store.JobByName(ref)
|
||||
}
|
||||
|
||||
// composeJob attaches the derived LatestRun to a stored job. A running job
|
||||
// advertises no next fire. Its record keeps the scheduler's cursor, which
|
||||
// Start needs to catch up on fires missed while the daemon was down.
|
||||
func (m *Manager) composeJob(job *jobsv0.Job) *jobsv0.Job {
|
||||
if latest, err := m.store.LatestRun(job.ID); err == nil {
|
||||
job.LatestRun = latest
|
||||
}
|
||||
|
||||
if job.State == jobsv0.JobStateRunning {
|
||||
job.NextFireAtNano = 0
|
||||
}
|
||||
|
||||
return job
|
||||
}
|
||||
|
||||
// maxContainerSpecSize bounds the raw container definition, the dominant
|
||||
// term of the job record's size against the storage point's value limit
|
||||
// (storagekvv0.MaxValueSize, 1 MiB): the spec is base64-encoded inside the
|
||||
// record's JSON (4/3 growth). Rejecting it here names the offending field;
|
||||
// the store caps the whole encoded record, covering the remaining
|
||||
// user-controlled fields.
|
||||
const maxContainerSpecSize = 512 * 1024
|
||||
|
||||
// validateSpec checks everything the contract rejects at registration time
|
||||
// and returns the decoded container definition.
|
||||
func (m *Manager) validateSpec(spec *jobsv0.JobSpec) (*container.CreateRequest, error) {
|
||||
if spec == nil {
|
||||
return nil, errdefs.InvalidParameter(errors.New("a job spec is required"))
|
||||
}
|
||||
if len(spec.ContainerSpec) > maxContainerSpecSize {
|
||||
return nil, errdefs.InvalidParameter(fmt.Errorf("container spec: definition exceeds the maximum size of %d bytes", maxContainerSpecSize))
|
||||
}
|
||||
if err := validateTrigger(spec.Trigger, m.validateCron); err != nil {
|
||||
return nil, errdefs.InvalidParameter(err)
|
||||
}
|
||||
if spec.TimeoutSeconds < 0 {
|
||||
return nil, errdefs.InvalidParameter(errors.New("timeout must not be negative"))
|
||||
}
|
||||
|
||||
if spec.TimeoutSeconds > int64(math.MaxInt64/time.Second) {
|
||||
return nil, errdefs.InvalidParameter(fmt.Errorf("timeout must not exceed %d seconds", int64(math.MaxInt64/time.Second)))
|
||||
}
|
||||
|
||||
for label := range spec.Labels {
|
||||
if strings.HasPrefix(label, "com.docker.job.") {
|
||||
return nil, errdefs.InvalidParameter(fmt.Errorf("label %s is reserved", label))
|
||||
}
|
||||
}
|
||||
|
||||
decoded, err := decodeContainerSpec(spec.ContainerSpec, true)
|
||||
if err != nil {
|
||||
return nil, errdefs.InvalidParameter(fmt.Errorf("container spec: %w", err))
|
||||
}
|
||||
if decoded.Image == "" {
|
||||
return nil, errdefs.InvalidParameter(errors.New("container spec: an image is required"))
|
||||
}
|
||||
for label := range decoded.Labels {
|
||||
if strings.HasPrefix(label, "com.docker.job.") {
|
||||
return nil, errdefs.InvalidParameter(fmt.Errorf("container spec: label %s is reserved", label))
|
||||
}
|
||||
}
|
||||
// Cancels, timeouts and reconciliation wait for the container to stop;
|
||||
// a stop that never escalates to a kill would hold the job forever.
|
||||
if decoded.StopTimeout != nil && *decoded.StopTimeout < 0 {
|
||||
return nil, errdefs.InvalidParameter(errors.New("container spec: a negative StopTimeout is not supported for jobs; jobs must be stoppable"))
|
||||
}
|
||||
|
||||
if hc := decoded.HostConfig; hc != nil {
|
||||
// AutoRemove races the exit-code capture the run record depends on;
|
||||
// RemoveOnSuccess/RemoveOnFailure are the job-level equivalents.
|
||||
if hc.AutoRemove {
|
||||
return nil, errdefs.InvalidParameter(errors.New("container spec: AutoRemove is not supported for jobs; use RemoveOnSuccess or RemoveOnFailure"))
|
||||
}
|
||||
// A job is expected to terminate; keep-alive restart policies would
|
||||
// make the run never reach an outcome. The same goes for on-failure
|
||||
// without a retry cap, which Docker treats as retry-forever.
|
||||
switch hc.RestartPolicy.Name {
|
||||
case "", container.RestartPolicyDisabled:
|
||||
case container.RestartPolicyOnFailure:
|
||||
if hc.RestartPolicy.MaximumRetryCount <= 0 {
|
||||
return nil, errdefs.InvalidParameter(errors.New("container spec: on-failure requires a maximum retry count for jobs; unbounded retries would keep the run from ever completing"))
|
||||
}
|
||||
default:
|
||||
return nil, errdefs.InvalidParameter(fmt.Errorf("container spec: restart policy %s is not supported for jobs; use no or on-failure", hc.RestartPolicy.Name))
|
||||
}
|
||||
}
|
||||
return decoded, nil
|
||||
}
|
||||
|
||||
// validateTrigger enforces the trigger sum-type rules of the API contract.
|
||||
func validateTrigger(trigger *jobsv0.Trigger, validateCron func(string) error) error {
|
||||
if trigger == nil {
|
||||
return nil // the manual shorthand
|
||||
}
|
||||
switch {
|
||||
case trigger.Manual && trigger.Schedule != nil:
|
||||
return errors.New("trigger declares more than one kind")
|
||||
case trigger.Schedule != nil:
|
||||
sched := trigger.Schedule
|
||||
if err := validateCron(sched.Cron); err != nil {
|
||||
return fmt.Errorf("schedule: %w", err)
|
||||
}
|
||||
if sched.Timezone != "" {
|
||||
// Local resolves to the daemon host's timezone, which a spec must
|
||||
// not depend on.
|
||||
if sched.Timezone == "Local" {
|
||||
return errors.New("schedule: timezone Local is not supported; use an IANA timezone name")
|
||||
}
|
||||
|
||||
if _, err := time.LoadLocation(sched.Timezone); err != nil {
|
||||
return fmt.Errorf("schedule: unknown timezone %s", sched.Timezone)
|
||||
}
|
||||
}
|
||||
switch sched.Concurrency {
|
||||
case "", jobsv0.ConcurrencyForbid, jobsv0.ConcurrencyQueue:
|
||||
default:
|
||||
return fmt.Errorf("schedule: unsupported concurrency policy %s", sched.Concurrency)
|
||||
}
|
||||
switch sched.MissedFires {
|
||||
case "", jobsv0.MissedFiresOne, jobsv0.MissedFiresSkip:
|
||||
default:
|
||||
return fmt.Errorf("schedule: unsupported missed-fires policy %s", sched.MissedFires)
|
||||
}
|
||||
return nil
|
||||
case trigger.Manual:
|
||||
return nil
|
||||
default:
|
||||
// Deliberate forward-compatibility failure: a trigger kind from a
|
||||
// newer contract version reaches an older daemon as an empty
|
||||
// message, because protobuf drops unknown fields. Registering it
|
||||
// as anything would silently disarm it.
|
||||
return errors.New("trigger declares no known kind; the daemon may be too old for this trigger type")
|
||||
}
|
||||
}
|
||||
|
||||
// validateJobName applies the container-name character rules to job names.
|
||||
func validateJobName(name string) error {
|
||||
if name == "" {
|
||||
return errors.New("a job name is required")
|
||||
}
|
||||
if !names.RestrictedNamePattern.MatchString(name) {
|
||||
return fmt.Errorf("invalid job name %q, only %s are allowed", name, names.RestrictedNameChars)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// decodeContainerSpec decodes the JSON container definition, in the exact
|
||||
// format of the container-create API request body.
|
||||
// Registration passes strict to reject unknown fields, so a typo fails
|
||||
// instead of silently altering the container.
|
||||
// Stored specs decode leniently: a field accepted by the daemon that
|
||||
// registered the job may have been removed from the API types since.
|
||||
func decodeContainerSpec(raw []byte, strict bool) (*container.CreateRequest, error) {
|
||||
if len(raw) == 0 {
|
||||
return nil, errors.New("a container definition is required")
|
||||
}
|
||||
decoder := json.NewDecoder(bytes.NewReader(raw))
|
||||
if strict {
|
||||
decoder.DisallowUnknownFields()
|
||||
}
|
||||
|
||||
req := &container.CreateRequest{}
|
||||
if err := decoder.Decode(req); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if decoder.More() {
|
||||
return nil, errors.New("unexpected trailing data after the container definition")
|
||||
}
|
||||
if req.Config == nil {
|
||||
req.Config = &container.Config{}
|
||||
}
|
||||
return req, nil
|
||||
}
|
||||
|
||||
// canonicalSpec is the hashing form of a job spec: daemon defaults applied,
|
||||
// container definition re-encoded from its decoded form so JSON formatting
|
||||
// differences (whitespace, key order — encoding/json sorts map keys) do not
|
||||
// change a job's identity. Structural presence still matters: an explicit
|
||||
// empty HostConfig decodes to a non-nil pointer and hashes differently from
|
||||
// an absent one, a conservative trade-off over normalizing every substruct.
|
||||
type canonicalSpec struct {
|
||||
ContainerSpec *container.CreateRequest
|
||||
TriggerKind string
|
||||
Schedule *jobsv0.ScheduleTrigger `json:",omitempty"`
|
||||
Labels map[string]string `json:",omitempty"`
|
||||
TimeoutSeconds int64
|
||||
RemoveOnSuccess bool
|
||||
RemoveOnFailure bool
|
||||
RunHistoryLimit uint32
|
||||
}
|
||||
|
||||
// specHash computes the canonical identity hash of a validated spec.
|
||||
func specHash(spec *jobsv0.JobSpec, decoded *container.CreateRequest) (string, error) {
|
||||
canonical := canonicalSpec{
|
||||
ContainerSpec: decoded,
|
||||
TriggerKind: triggerKind(spec.Trigger),
|
||||
Labels: spec.Labels,
|
||||
TimeoutSeconds: spec.TimeoutSeconds,
|
||||
RemoveOnSuccess: spec.RemoveOnSuccess,
|
||||
RemoveOnFailure: spec.RemoveOnFailure,
|
||||
RunHistoryLimit: spec.RunHistoryLimit,
|
||||
}
|
||||
if canonical.RunHistoryLimit == 0 {
|
||||
canonical.RunHistoryLimit = DefaultRunHistoryLimit
|
||||
}
|
||||
if spec.Trigger != nil && spec.Trigger.Schedule != nil {
|
||||
sched := *spec.Trigger.Schedule
|
||||
if sched.Timezone == "" {
|
||||
sched.Timezone = "UTC"
|
||||
}
|
||||
if sched.Concurrency == "" {
|
||||
sched.Concurrency = jobsv0.ConcurrencyForbid
|
||||
}
|
||||
if sched.MissedFires == "" {
|
||||
sched.MissedFires = jobsv0.MissedFiresOne
|
||||
}
|
||||
canonical.Schedule = &sched
|
||||
}
|
||||
data, err := json.Marshal(canonical)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("encoding canonical spec: %w", err)
|
||||
}
|
||||
return fmt.Sprintf("sha256:%x", sha256.Sum256(data)), nil
|
||||
}
|
||||
|
||||
// currentSpecHash computes a stored job's spec hash with this daemon's
|
||||
// canonicalization.
|
||||
// The persisted hash comes from the daemon that registered the job, and the
|
||||
// canonical encoding changes whenever the container API types gain fields.
|
||||
func currentSpecHash(job *jobsv0.Job) string {
|
||||
decoded, err := decodeContainerSpec(job.Spec.ContainerSpec, false)
|
||||
if err != nil {
|
||||
return job.SpecHash
|
||||
}
|
||||
|
||||
hash, err := specHash(job.Spec, decoded)
|
||||
if err != nil {
|
||||
return job.SpecHash
|
||||
}
|
||||
|
||||
return hash
|
||||
}
|
||||
|
||||
// triggerKind names a validated trigger's kind; a nil trigger is the manual
|
||||
// shorthand.
|
||||
func triggerKind(trigger *jobsv0.Trigger) string {
|
||||
if trigger != nil && trigger.Schedule != nil {
|
||||
return jobsv0.TriggerKindSchedule
|
||||
}
|
||||
return jobsv0.TriggerKindManual
|
||||
}
|
||||
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,432 @@
|
||||
package jobs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
cerrdefs "github.com/containerd/errdefs"
|
||||
"github.com/containerd/log"
|
||||
"github.com/moby/moby/v2/errdefs"
|
||||
jobsv0 "github.com/moby/moby/v2/extpoints/jobs/api/v0"
|
||||
)
|
||||
|
||||
// Inspect returns a job with its latest run composed in.
|
||||
func (m *Manager) Inspect(ctx context.Context, jobRef string) (*jobsv0.Job, error) {
|
||||
job, err := m.resolveJob(jobRef)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return m.composeJob(job), nil
|
||||
}
|
||||
|
||||
// List returns the jobs matching every filter of the request, latest runs
|
||||
// composed in.
|
||||
func (m *Manager) List(ctx context.Context, req *jobsv0.ListRequest) ([]*jobsv0.Job, error) {
|
||||
if req == nil {
|
||||
req = &jobsv0.ListRequest{}
|
||||
}
|
||||
switch req.Paused {
|
||||
case "", "true", "false":
|
||||
default:
|
||||
return nil, errdefs.InvalidParameter(fmt.Errorf("invalid paused filter %q: must be true or false", req.Paused))
|
||||
}
|
||||
for _, state := range req.States {
|
||||
if state != jobsv0.JobStateIdle && state != jobsv0.JobStateRunning {
|
||||
return nil, errdefs.InvalidParameter(fmt.Errorf("invalid state filter %q", state))
|
||||
}
|
||||
}
|
||||
for _, kind := range req.TriggerKinds {
|
||||
if kind != jobsv0.TriggerKindManual && kind != jobsv0.TriggerKindSchedule {
|
||||
return nil, errdefs.InvalidParameter(fmt.Errorf("invalid trigger filter %q", kind))
|
||||
}
|
||||
}
|
||||
for _, state := range req.LatestRunStates {
|
||||
if !isTerminalRunState(state) && state != jobsv0.RunStatePending && state != jobsv0.RunStateRunning {
|
||||
return nil, errdefs.InvalidParameter(fmt.Errorf("invalid latest-run-state filter %q", state))
|
||||
}
|
||||
}
|
||||
|
||||
var jobs []*jobsv0.Job
|
||||
for _, job := range m.store.Jobs() {
|
||||
if len(req.Names) > 0 && !slices.Contains(req.Names, job.Name) {
|
||||
continue
|
||||
}
|
||||
if len(req.States) > 0 && !slices.Contains(req.States, job.State) {
|
||||
continue
|
||||
}
|
||||
if len(req.TriggerKinds) > 0 && !slices.Contains(req.TriggerKinds, triggerKind(job.Spec.Trigger)) {
|
||||
continue
|
||||
}
|
||||
if (req.Paused == "true" && !job.Paused) || (req.Paused == "false" && job.Paused) {
|
||||
continue
|
||||
}
|
||||
if !matchesLabels(job.Spec.Labels, req.Labels) {
|
||||
continue
|
||||
}
|
||||
job = m.composeJob(job)
|
||||
if len(req.LatestRunStates) > 0 && (job.LatestRun == nil || !slices.Contains(req.LatestRunStates, job.LatestRun.State)) {
|
||||
continue
|
||||
}
|
||||
jobs = append(jobs, job)
|
||||
}
|
||||
slices.SortFunc(jobs, func(a, b *jobsv0.Job) int { return strings.Compare(a.Name, b.Name) })
|
||||
return jobs, nil
|
||||
}
|
||||
|
||||
// Pause suppresses a job's schedule trigger. An in-flight run is not
|
||||
// affected; explicit runs still work. Idempotent, a no-op for manual jobs.
|
||||
func (m *Manager) Pause(ctx context.Context, jobRef string) error {
|
||||
return m.setPaused(ctx, jobRef, true)
|
||||
}
|
||||
|
||||
// Resume re-arms a paused job's schedule trigger from the next tick.
|
||||
// Idempotent, a no-op for manual jobs.
|
||||
func (m *Manager) Resume(ctx context.Context, jobRef string) error {
|
||||
return m.setPaused(ctx, jobRef, false)
|
||||
}
|
||||
|
||||
func (m *Manager) setPaused(ctx context.Context, jobRef string, paused bool) error {
|
||||
job, err := m.resolveJob(jobRef)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if triggerKind(job.Spec.Trigger) == jobsv0.TriggerKindManual {
|
||||
return nil
|
||||
}
|
||||
m.locks.Lock(job.ID)
|
||||
defer m.locks.Unlock(job.ID)
|
||||
job, err = m.store.Job(job.ID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if job.Paused == paused {
|
||||
return nil
|
||||
}
|
||||
// The scheduler and the queued fire change only once the new state is
|
||||
// persisted, so a failed write leaves the job as it was.
|
||||
job.Paused = paused
|
||||
|
||||
var (
|
||||
schedule *cronSchedule
|
||||
loc *time.Location
|
||||
next time.Time
|
||||
rearm bool
|
||||
)
|
||||
|
||||
if paused {
|
||||
job.NextFireAtNano = 0
|
||||
} else {
|
||||
// Re-arm from the next occurrence; fires missed while paused are
|
||||
// not backfilled.
|
||||
if schedule, loc, err = parseScheduleTrigger(job.Spec.Trigger.Schedule); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if next, rearm = schedule.next(m.now(), loc); rearm {
|
||||
job.NextFireAtNano = next.UnixNano()
|
||||
}
|
||||
}
|
||||
job.UpdatedAtNano = m.now().UnixNano()
|
||||
if err := m.store.UpdateJob(ctx, job); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if paused {
|
||||
m.sched.disarm(job.ID)
|
||||
// A deferred fire must not survive a pause/resume cycle.
|
||||
m.mu.Lock()
|
||||
delete(m.queued, job.ID)
|
||||
m.mu.Unlock()
|
||||
} else if rearm {
|
||||
m.sched.arm(job.ID, schedule, loc, next)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// removeRunsPageSize is the page size Remove drains terminal runs with; a
|
||||
// smaller page than the store default keeps each critical section short.
|
||||
const removeRunsPageSize = 200
|
||||
|
||||
// Remove deletes a job, cancelling its in-flight run and applying the
|
||||
// requested run-history retention mode.
|
||||
func (m *Manager) Remove(ctx context.Context, jobRef, runsRemoval string) error {
|
||||
switch runsRemoval {
|
||||
case "", jobsv0.RunsKeep, jobsv0.RunsRemove, jobsv0.RunsRemoveFinished:
|
||||
default:
|
||||
return errdefs.InvalidParameter(fmt.Errorf("invalid runs-removal mode %q", runsRemoval))
|
||||
}
|
||||
|
||||
jobID, err := m.resolveHistoryJobID(jobRef)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
m.locks.Lock(jobID)
|
||||
defer m.locks.Unlock(jobID)
|
||||
|
||||
job, err := m.store.Job(jobID)
|
||||
if err != nil {
|
||||
if cerrdefs.IsNotFound(err) && runsRemoval == jobsv0.RunsRemove {
|
||||
return m.store.PurgeJob(ctx, jobID)
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
// Cancel the in-flight run. Its record stays non-terminal until the
|
||||
// watcher observes the container exit, which makes the remove-finished
|
||||
// retention rule hold naturally: the run being cancelled by this very
|
||||
// removal still counts as in-flight below.
|
||||
if run, err := m.store.LatestRun(job.ID); err == nil && !isTerminalRunState(run.State) {
|
||||
// Like Cancel, stop even when the terminal decision was claimed
|
||||
// already, as an earlier stop may have failed.
|
||||
m.setOverride(run.ID, jobsv0.RunStateCancelled)
|
||||
containerID := run.ContainerID
|
||||
stopCtx := context.WithoutCancel(ctx)
|
||||
|
||||
m.background.Go(func() {
|
||||
if err := m.backend.ContainerStop(stopCtx, containerID); err != nil {
|
||||
log.G(stopCtx).WithError(err).WithFields(log.Fields{"job": job.ID, "run": run.ID}).Warn("could not stop run container of removed job")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
var removeErr error
|
||||
switch runsRemoval {
|
||||
case jobsv0.RunsRemove:
|
||||
removeErr = m.store.PurgeJob(ctx, job.ID)
|
||||
case jobsv0.RunsRemoveFinished:
|
||||
for {
|
||||
page, cursor, _, err := m.store.Runs(job.ID, removeRunsPageSize, "")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
deleted := false
|
||||
for _, run := range page {
|
||||
if isTerminalRunState(run.State) {
|
||||
if err := m.store.DeleteRun(ctx, job.ID, run.ID); err != nil {
|
||||
return err
|
||||
}
|
||||
deleted = true
|
||||
}
|
||||
}
|
||||
// At most one run is ever non-terminal (the in-flight one), so
|
||||
// a page with nothing left to delete means only that run
|
||||
// remains; the guard keeps the loop finite even if that
|
||||
// invariant is ever broken.
|
||||
if cursor == "" || !deleted {
|
||||
break
|
||||
}
|
||||
}
|
||||
removeErr = m.store.DeleteJob(ctx, job.ID)
|
||||
default:
|
||||
removeErr = m.store.DeleteJob(ctx, job.ID)
|
||||
}
|
||||
// Removing the history can fail after the job record is gone; the job
|
||||
// must not stay armed or keep its waiters blocked then.
|
||||
if _, err := m.store.Job(job.ID); cerrdefs.IsNotFound(err) {
|
||||
m.sched.disarm(job.ID)
|
||||
m.mu.Lock()
|
||||
delete(m.queued, job.ID)
|
||||
m.mu.Unlock()
|
||||
// Wake waiters so a Wait on the removed job re-resolves to
|
||||
// not-found instead of blocking until its context expires.
|
||||
m.broadcast(job.ID)
|
||||
}
|
||||
return removeErr
|
||||
}
|
||||
|
||||
// Prune removes idle manual jobs matching the label filters, keeping their
|
||||
// run history. Schedule jobs are never pruned: idle means armed, not
|
||||
// abandoned.
|
||||
func (m *Manager) Prune(ctx context.Context, labels []string) ([]string, error) {
|
||||
var removed []string
|
||||
for _, job := range m.store.Jobs() {
|
||||
if triggerKind(job.Spec.Trigger) != jobsv0.TriggerKindManual {
|
||||
continue
|
||||
}
|
||||
if !matchesLabels(job.Spec.Labels, labels) {
|
||||
continue
|
||||
}
|
||||
m.locks.Lock(job.ID)
|
||||
// Re-read under the lock: the job may have started running or been
|
||||
// removed since the snapshot.
|
||||
current, err := m.store.Job(job.ID)
|
||||
if err == nil && current.State == jobsv0.JobStateIdle {
|
||||
if err := m.store.DeleteJob(ctx, job.ID); err == nil {
|
||||
removed = append(removed, job.ID)
|
||||
m.broadcast(job.ID)
|
||||
}
|
||||
}
|
||||
m.locks.Unlock(job.ID)
|
||||
}
|
||||
slices.Sort(removed)
|
||||
return removed, nil
|
||||
}
|
||||
|
||||
// RunsPage returns one page of a job's runs, newest first.
|
||||
func (m *Manager) RunsPage(ctx context.Context, jobRef string, limit int, before string) (page []*jobsv0.Run, nextCursor string, staleCursor bool, err error) {
|
||||
jobID, err := m.resolveHistoryJobID(jobRef)
|
||||
if err != nil {
|
||||
return nil, "", false, err
|
||||
}
|
||||
|
||||
return m.store.Runs(jobID, limit, before)
|
||||
}
|
||||
|
||||
// InspectRun returns one run; runRef accepts an ID, "latest", or empty for
|
||||
// latest.
|
||||
func (m *Manager) InspectRun(ctx context.Context, jobRef, runRef string) (*jobsv0.Run, error) {
|
||||
jobID, err := m.resolveHistoryJobID(jobRef)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return m.resolveRun(jobID, runRef)
|
||||
}
|
||||
|
||||
// resolveHistoryJobID includes retained history in ID resolution, before
|
||||
// trying live job names. Removed job names may belong to a new job now.
|
||||
func (m *Manager) resolveHistoryJobID(ref string) (string, error) {
|
||||
if _, err := m.store.LatestRun(ref); err == nil {
|
||||
return ref, nil
|
||||
}
|
||||
|
||||
job, err := m.resolveJob(ref)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
return job.ID, nil
|
||||
}
|
||||
|
||||
func (m *Manager) resolveRun(jobID, runRef string) (*jobsv0.Run, error) {
|
||||
if runRef == "" || runRef == "latest" {
|
||||
return m.store.LatestRun(jobID)
|
||||
}
|
||||
return m.store.Run(jobID, runRef)
|
||||
}
|
||||
|
||||
// Wait blocks until the referenced run satisfies the condition and returns
|
||||
// it. Waiting for "latest" on a job with no runs blocks until one is fired,
|
||||
// and follows the first run it observes, not later ones. The job is
|
||||
// resolved once, so a job re-created under the same name is not followed
|
||||
// either. A run that jumps past the condition satisfies it: waiting for
|
||||
// running on a run that went straight to a terminal state returns that
|
||||
// terminal run.
|
||||
func (m *Manager) Wait(ctx context.Context, jobRef, runRef, condition string) (*jobsv0.Run, error) {
|
||||
switch condition {
|
||||
case "", jobsv0.WaitConditionTerminal, jobsv0.WaitConditionRunning:
|
||||
default:
|
||||
return nil, errdefs.InvalidParameter(fmt.Errorf("invalid wait condition %q", condition))
|
||||
}
|
||||
|
||||
job, err := m.resolveJob(jobRef)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// check reports the run once it meets the condition, pinning runRef to
|
||||
// the first run observed.
|
||||
check := func() (_ *jobsv0.Run, met bool, _ error) {
|
||||
run, err := m.resolveRun(job.ID, runRef)
|
||||
if err != nil {
|
||||
if runRef == "" || runRef == "latest" {
|
||||
// No run fired yet.
|
||||
return nil, false, nil
|
||||
}
|
||||
// An explicit run ID either exists or never will.
|
||||
return nil, false, err
|
||||
}
|
||||
|
||||
runRef = run.ID
|
||||
|
||||
return run, waitConditionMet(run, condition), nil
|
||||
}
|
||||
for {
|
||||
if run, met, err := check(); err != nil || met {
|
||||
return run, err
|
||||
}
|
||||
// Subscribe, then re-check both the job and the run: a transition —
|
||||
// or the job's removal, which broadcasts exactly once — between the
|
||||
// checks above and the subscription would otherwise be missed
|
||||
// forever. Returns that were not woken withdraw their subscription
|
||||
// so entries do not accumulate on jobs that stopped transitioning.
|
||||
ch := m.subscribe(job.ID)
|
||||
if _, err := m.store.Job(job.ID); err != nil {
|
||||
m.unsubscribe(job.ID, ch)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if run, met, err := check(); err != nil || met {
|
||||
m.unsubscribe(job.ID, ch)
|
||||
return run, err
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
m.unsubscribe(job.ID, ch)
|
||||
return nil, ctx.Err()
|
||||
case <-ch:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// waitConditionMet reports whether a run satisfies a wait condition. The
|
||||
// zero condition means terminal.
|
||||
func waitConditionMet(run *jobsv0.Run, condition string) bool {
|
||||
if condition == jobsv0.WaitConditionRunning {
|
||||
return run.State != jobsv0.RunStatePending
|
||||
}
|
||||
return isTerminalRunState(run.State)
|
||||
}
|
||||
|
||||
// matchesLabels reports whether labels satisfy every filter entry, each
|
||||
// either "key" (presence) or "key=value" (equality), matching the container
|
||||
// API's label-filter semantics.
|
||||
func matchesLabels(labels map[string]string, filterEntries []string) bool {
|
||||
for _, entry := range filterEntries {
|
||||
key, value, hasValue := strings.Cut(entry, "=")
|
||||
stored, present := labels[key]
|
||||
if !present || (hasValue && stored != value) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// Run executes an existing job, creating its next run. The returned run may
|
||||
// already be terminal when the container could not be created or started;
|
||||
// the failure is recorded on the run record, per the API contract.
|
||||
//
|
||||
// With reschedule, the manual fire stands in for the job's upcoming
|
||||
// scheduled occurrence: the schedule re-arms on the occurrence after it,
|
||||
// instead of keeping the original cadence. An occurrence a tick already
|
||||
// consumed but has not fired yet counts as the upcoming one.
|
||||
func (m *Manager) Run(ctx context.Context, jobRef string, reschedule bool) (*jobsv0.Run, error) {
|
||||
job, err := m.resolveJob(jobRef)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if reschedule && triggerKind(job.Spec.Trigger) != jobsv0.TriggerKindSchedule {
|
||||
return nil, errdefs.InvalidParameter(errors.New("reschedule requires a schedule trigger"))
|
||||
}
|
||||
|
||||
// Holding the job's lock across the fire and the replacement keeps a
|
||||
// tick's fire for the replaced occurrence waiting until it is dropped.
|
||||
m.locks.Lock(job.ID)
|
||||
defer m.locks.Unlock(job.ID)
|
||||
|
||||
run, err := m.fireManualLocked(ctx, job.ID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if reschedule && m.sched.replaceNext(job.ID) {
|
||||
m.persistNextFireLocked(ctx, job.ID)
|
||||
}
|
||||
return run, nil
|
||||
}
|
||||
@@ -0,0 +1,604 @@
|
||||
package jobs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
"time"
|
||||
|
||||
cerrdefs "github.com/containerd/errdefs"
|
||||
"github.com/containerd/log"
|
||||
"github.com/moby/moby/v2/daemon/internal/stringid"
|
||||
jobsv0 "github.com/moby/moby/v2/extpoints/jobs/api/v0"
|
||||
)
|
||||
|
||||
// errFireQueued reports that a fire was deferred behind the current run by
|
||||
// the queue concurrency policy; the deferred fire happens when that run
|
||||
// completes. Callers (the scheduler) treat it as a skip, not a failure.
|
||||
var errFireQueued = errors.New("fire queued behind the current run")
|
||||
|
||||
// errFireDropped reports that a fire was dropped because a fire is already
|
||||
// queued (the queue has depth one).
|
||||
var errFireDropped = errors.New("fire dropped: a fire is already queued")
|
||||
|
||||
var errFirePaused = errors.New("scheduled fire dropped: the job is paused")
|
||||
|
||||
// terminalOutcome is what a run ends with.
|
||||
type terminalOutcome struct {
|
||||
state string
|
||||
exitCode *int64
|
||||
errMsg string
|
||||
}
|
||||
|
||||
// fireManual fires a run for an explicit Run or CreateAndRun call.
|
||||
func (m *Manager) fireManual(ctx context.Context, jobID string) (*jobsv0.Run, error) {
|
||||
m.locks.Lock(jobID)
|
||||
defer m.locks.Unlock(jobID)
|
||||
|
||||
return m.fireManualLocked(ctx, jobID)
|
||||
}
|
||||
|
||||
// fireManualLocked is fireManual for callers already holding the job's lock.
|
||||
func (m *Manager) fireManualLocked(ctx context.Context, jobID string) (*jobsv0.Run, error) {
|
||||
return m.tryFireLocked(ctx, jobID, &jobsv0.TriggerEvidence{
|
||||
Kind: jobsv0.TriggerKindManual,
|
||||
FiredAtNano: m.now().UnixNano(),
|
||||
})
|
||||
}
|
||||
|
||||
// tryFireLocked is the single chokepoint every fire goes through, whatever
|
||||
// the trigger: called under the job's lock, it applies the concurrency policy
|
||||
// against the job's state re-read under that lock, and only then creates the
|
||||
// run. Two triggers racing on the same job therefore cannot both pass.
|
||||
func (m *Manager) tryFireLocked(ctx context.Context, jobID string, evidence *jobsv0.TriggerEvidence) (*jobsv0.Run, error) {
|
||||
// A fire outlives its trigger. The daemon completes a container start
|
||||
// even when the API caller goes away, so a cancelled request must not
|
||||
// record a failure and release the job while its container runs.
|
||||
ctx = context.WithoutCancel(ctx)
|
||||
|
||||
job, err := m.store.Job(jobID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if job.Paused && evidence.Kind == jobsv0.TriggerKindSchedule {
|
||||
return nil, errFirePaused
|
||||
}
|
||||
|
||||
if job.State == jobsv0.JobStateRunning {
|
||||
if evidence.Kind == jobsv0.TriggerKindSchedule && scheduleConcurrency(job) == jobsv0.ConcurrencyQueue {
|
||||
m.mu.Lock()
|
||||
_, full := m.queued[jobID]
|
||||
if !full {
|
||||
m.queued[jobID] = evidence
|
||||
}
|
||||
m.mu.Unlock()
|
||||
if full {
|
||||
return nil, errFireDropped
|
||||
}
|
||||
return nil, errFireQueued
|
||||
}
|
||||
return nil, cerrdefs.ErrFailedPrecondition.WithMessage(fmt.Sprintf("job %s is already running", job.Name))
|
||||
}
|
||||
|
||||
now := m.now().UnixNano()
|
||||
run := &jobsv0.Run{
|
||||
ID: stringid.GenerateRandomID(),
|
||||
JobID: job.ID,
|
||||
State: jobsv0.RunStatePending,
|
||||
CreatedAtNano: now,
|
||||
Trigger: evidence,
|
||||
}
|
||||
// The run record is persisted BEFORE the container is created. Every
|
||||
// crash-safety property depends on this ordering: a daemon crash or a
|
||||
// container-create failure leaves a discoverable run record instead of
|
||||
// an orphaned container. Do not reorder.
|
||||
if err := m.store.CreateRun(ctx, run); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
job.State = jobsv0.JobStateRunning
|
||||
job.UpdatedAtNano = now
|
||||
|
||||
if err := m.store.UpdateJob(ctx, job); err != nil {
|
||||
m.completeRunLocked(ctx, job, run, terminalOutcome{state: jobsv0.RunStateFailed, errMsg: "recording job state: " + err.Error()})
|
||||
return m.store.Run(job.ID, run.ID)
|
||||
}
|
||||
m.broadcast(job.ID)
|
||||
|
||||
if err := m.startContainer(ctx, job, run); err != nil {
|
||||
// The failure is recorded on the run, not returned: the run exists
|
||||
// and its record is the outcome, per the API contract.
|
||||
m.failUnstartedLocked(ctx, job, run, err.Error())
|
||||
}
|
||||
return m.store.Run(job.ID, run.ID)
|
||||
}
|
||||
|
||||
// startContainer creates and starts the run's container, arms the timeout,
|
||||
// and hands off to the exit watcher. Called under the job's lock.
|
||||
func (m *Manager) startContainer(ctx context.Context, job *jobsv0.Job, run *jobsv0.Run) error {
|
||||
req, err := decodeContainerSpec(job.Spec.ContainerSpec, false)
|
||||
if err != nil {
|
||||
// Validated at registration; only a corrupted store record gets here.
|
||||
return fmt.Errorf("decoding container spec: %w", err)
|
||||
}
|
||||
if req.Config.Labels == nil {
|
||||
req.Config.Labels = make(map[string]string, 2)
|
||||
}
|
||||
req.Config.Labels[LabelJobID] = job.ID
|
||||
req.Config.Labels[LabelRunID] = run.ID
|
||||
|
||||
created, err := m.backend.ContainerCreate(ctx, ContainerCreateRequest{
|
||||
Name: runContainerName(job, run),
|
||||
Config: req.Config,
|
||||
HostConfig: req.HostConfig,
|
||||
NetworkingConfig: req.NetworkingConfig,
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("creating run container: %w", err)
|
||||
}
|
||||
run.ContainerID = created.ID
|
||||
if err := m.store.UpdateRun(ctx, run); err != nil {
|
||||
return fmt.Errorf("recording run container: %w", err)
|
||||
}
|
||||
|
||||
if err := m.backend.ContainerStart(ctx, created.ID); err != nil {
|
||||
return fmt.Errorf("starting run container: %w", err)
|
||||
}
|
||||
run.State = jobsv0.RunStateRunning
|
||||
run.StartedAtNano = m.now().UnixNano()
|
||||
|
||||
if err := m.store.UpdateRun(ctx, run); err != nil {
|
||||
// The container is already running. Keep observing it until exit,
|
||||
// even if stopping fails, rather than free the job for another run.
|
||||
run.Error = "recording run start: " + err.Error()
|
||||
|
||||
m.background.Go(func() {
|
||||
if err := m.backend.ContainerStop(ctx, run.ContainerID); err != nil {
|
||||
log.G(ctx).WithError(err).WithFields(log.Fields{"job": job.ID, "run": run.ID}).Warn("could not stop run container after recording its start failed")
|
||||
}
|
||||
})
|
||||
} else {
|
||||
m.broadcast(job.ID)
|
||||
}
|
||||
|
||||
m.watch(ctx, job, run)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// runContainerName names a run's container. The run-ID suffix keeps the name
|
||||
// unique across job generations: a removed job leaves its kept containers
|
||||
// behind, and a re-created job under the same name restarts its iterations
|
||||
// at one.
|
||||
func runContainerName(job *jobsv0.Job, run *jobsv0.Run) string {
|
||||
return fmt.Sprintf("job-%s-%d-%s", job.Name, run.Iteration, stringid.TruncateID(run.ID))
|
||||
}
|
||||
|
||||
// failUnstartedLocked fails a run whose start is unknown: a failed start
|
||||
// request may still have started the container. A container left behind is
|
||||
// stopped before the job is released, so it cannot overlap the job's next
|
||||
// run; if stopping fails, the job stays occupied until the container is
|
||||
// observed exiting. Called under the job's lock.
|
||||
func (m *Manager) failUnstartedLocked(ctx context.Context, job *jobsv0.Job, run *jobsv0.Run, errMsg string) {
|
||||
run.Error = errMsg
|
||||
if run.ContainerID != "" {
|
||||
if err := m.backend.ContainerStop(ctx, run.ContainerID); err != nil && !cerrdefs.IsNotFound(err) {
|
||||
log.G(ctx).WithError(err).WithFields(log.Fields{"job": job.ID, "run": run.ID}).Warn("could not stop container of unstarted run; watching it until it exits")
|
||||
m.watch(ctx, job, run)
|
||||
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
m.completeRunLocked(ctx, job, run, terminalOutcome{state: jobsv0.RunStateFailed, errMsg: errMsg})
|
||||
}
|
||||
|
||||
// watch records an existing exit or arms the timeout and observes a future
|
||||
// exit in the background. Called under the job's lock.
|
||||
func (m *Manager) watch(ctx context.Context, job *jobsv0.Job, run *jobsv0.Run) {
|
||||
// Observation ends with the manager: under live-restore the container
|
||||
// keeps running past shutdown.
|
||||
waitCtx, cancelWait := context.WithCancel(ctx)
|
||||
|
||||
waitC, err := m.backend.ContainerWait(waitCtx, run.ContainerID)
|
||||
if err != nil {
|
||||
cancelWait()
|
||||
// Per the Backend contract, the container no longer exists.
|
||||
m.completeRunLocked(ctx, job, run, terminalOutcome{state: jobsv0.RunStateFailed, errMsg: "waiting on run container: " + err.Error()})
|
||||
return
|
||||
}
|
||||
// Reconcile an exit before enforcing a deadline: the container may have
|
||||
// finished within its timeout while the daemon was down.
|
||||
select {
|
||||
case status := <-waitC:
|
||||
cancelWait()
|
||||
m.completeRunStatusLocked(ctx, job, run, status)
|
||||
return
|
||||
default:
|
||||
}
|
||||
|
||||
// A run whose start was never recorded has no deadline to enforce.
|
||||
if job.Spec.TimeoutSeconds > 0 && run.StartedAtNano != 0 {
|
||||
deadline := time.Unix(0, run.StartedAtNano).Add(time.Duration(job.Spec.TimeoutSeconds) * time.Second)
|
||||
m.armTimeout(run.ID, run.ContainerID, deadline.Sub(m.now()))
|
||||
}
|
||||
|
||||
m.background.Go(func() {
|
||||
defer cancelWait()
|
||||
|
||||
select {
|
||||
case status := <-waitC:
|
||||
m.locks.Lock(job.ID)
|
||||
defer m.locks.Unlock(job.ID)
|
||||
|
||||
m.completeRunStatusLocked(ctx, job, run, status)
|
||||
case <-m.stop:
|
||||
// Live-restore keeps the container running. Leave the run in
|
||||
// flight on disk for reconciliation on the next startup.
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func (m *Manager) completeRunStatusLocked(ctx context.Context, job *jobsv0.Job, run *jobsv0.Run, status StateStatus) {
|
||||
if status == nil {
|
||||
// The Backend contract forbids closing without a status.
|
||||
m.completeRunLocked(ctx, job, run, terminalOutcome{state: jobsv0.RunStateFailed, errMsg: "waiting on run container: wait channel closed without a status"})
|
||||
return
|
||||
}
|
||||
|
||||
code := int64(status.ExitCode())
|
||||
|
||||
outcome := terminalOutcome{exitCode: &code, errMsg: run.Error}
|
||||
if waitErr := status.Err(); waitErr != nil {
|
||||
outcome.errMsg = waitErr.Error()
|
||||
}
|
||||
switch override := m.peekOverride(run.ID); {
|
||||
case override != "":
|
||||
outcome.state = override
|
||||
case outcome.errMsg != "":
|
||||
outcome.state = jobsv0.RunStateFailed
|
||||
case code == 0:
|
||||
outcome.state = jobsv0.RunStateSucceeded
|
||||
default:
|
||||
outcome.state = jobsv0.RunStateFailed
|
||||
}
|
||||
m.completeRunLocked(ctx, job, run, outcome)
|
||||
}
|
||||
|
||||
// completeRunLocked writes the run's terminal record, removes its container
|
||||
// when the spec asks for it, returns the job to idle, and fires any queued
|
||||
// trigger. Called under the job's lock; job is the snapshot taken at fire
|
||||
// time (its spec is immutable, its runtime state is re-read).
|
||||
func (m *Manager) completeRunLocked(ctx context.Context, job *jobsv0.Job, run *jobsv0.Run, outcome terminalOutcome) {
|
||||
// A terminal outcome must be recorded even when the triggering context
|
||||
// was cancelled: dropping the write would leave the job and run stuck
|
||||
// in running until restart reconciliation.
|
||||
ctx = context.WithoutCancel(ctx)
|
||||
m.disarmTimeout(run.ID)
|
||||
m.clearOverride(run.ID)
|
||||
|
||||
run.State = outcome.state
|
||||
run.ExitCode = nil
|
||||
if outcome.exitCode != nil {
|
||||
run.ExitCode = &jobsv0.ExitCode{Value: *outcome.exitCode}
|
||||
}
|
||||
run.Error = outcome.errMsg
|
||||
run.FinishedAtNano = m.now().UnixNano()
|
||||
m.finishRunLocked(ctx, job, run)
|
||||
}
|
||||
|
||||
// recordRetryInterval is how long a failed write of a run's terminal record,
|
||||
// or of its job's return to idle, waits before it is retried.
|
||||
const recordRetryInterval = 5 * time.Second
|
||||
|
||||
// finishRunLocked persists a terminal run record and releases the job.
|
||||
// Called under the job's lock.
|
||||
func (m *Manager) finishRunLocked(ctx context.Context, job *jobsv0.Job, run *jobsv0.Run) {
|
||||
switch err := m.store.UpdateRun(ctx, run); {
|
||||
case err == nil:
|
||||
// The outcome is recorded before the container is removed, so a
|
||||
// crash in between cannot lose it.
|
||||
// Timed-out and cancelled runs always keep their container for
|
||||
// postmortem inspection.
|
||||
remove := (run.State == jobsv0.RunStateSucceeded && job.Spec.RemoveOnSuccess) ||
|
||||
(run.State == jobsv0.RunStateFailed && job.Spec.RemoveOnFailure)
|
||||
if !remove || run.ContainerID == "" {
|
||||
break
|
||||
}
|
||||
|
||||
if err := m.backend.ContainerRm(ctx, run.ContainerID); err != nil {
|
||||
log.G(ctx).WithError(err).WithFields(log.Fields{"job": job.ID, "run": run.ID}).Warn("keeping run container that could not be auto-removed")
|
||||
break
|
||||
}
|
||||
|
||||
if err := m.store.MarkContainerGone(ctx, job.ID, run.ID); err != nil {
|
||||
log.G(ctx).WithError(err).WithFields(log.Fields{"job": job.ID, "run": run.ID}).Warn("could not record run container removal")
|
||||
}
|
||||
case cerrdefs.IsNotFound(err):
|
||||
// The job was purged while the run was in flight, which left its
|
||||
// still-running container behind; it goes with the record.
|
||||
m.removeRunContainers(ctx, job.ID, []jobsv0.Run{*run})
|
||||
default:
|
||||
// Releasing the job would leave the run in flight with nothing left
|
||||
// to finish it, and a later run would hide it from restart
|
||||
// reconciliation. Keep the job occupied and retry; a shutdown
|
||||
// leaves the run in flight for that reconciliation.
|
||||
log.G(ctx).WithError(err).WithFields(log.Fields{"job": job.ID, "run": run.ID}).Warn("could not record run outcome; retrying")
|
||||
m.retryLater(job.ID, func() { m.finishRunLocked(ctx, job, run) })
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
m.releaseJobLocked(ctx, job.ID, run.ID)
|
||||
}
|
||||
|
||||
// releaseJobLocked returns a job whose run is terminal to idle, and fires any
|
||||
// queued trigger. A failed write is retried on its own, as the terminal run
|
||||
// record cannot be written again; until it lands the job stays running, so
|
||||
// no other run starts. Called under the job's lock.
|
||||
func (m *Manager) releaseJobLocked(ctx context.Context, jobID, runID string) {
|
||||
// Only the job's latest run releases it. A run whose completion was
|
||||
// retried can be overtaken when marking its job running failed: the job
|
||||
// stayed idle on disk, and a newer run took it meanwhile.
|
||||
if latest, err := m.store.LatestRun(jobID); err == nil && latest.ID != runID {
|
||||
m.broadcast(jobID)
|
||||
return
|
||||
}
|
||||
|
||||
// The job may have been removed while the run was in flight; its runs
|
||||
// are then a tombstone and there is no state to restore.
|
||||
if current, err := m.store.Job(jobID); err == nil {
|
||||
current.State = jobsv0.JobStateIdle
|
||||
current.UpdatedAtNano = m.now().UnixNano()
|
||||
|
||||
if err := m.store.UpdateJob(ctx, current); err != nil {
|
||||
log.G(ctx).WithError(err).WithFields(log.Fields{"job": jobID}).Warn("could not return job to idle; retrying")
|
||||
m.retryLater(jobID, func() { m.releaseJobLocked(ctx, jobID, runID) })
|
||||
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
m.broadcast(jobID)
|
||||
|
||||
m.mu.Lock()
|
||||
_, queued := m.queued[jobID]
|
||||
m.mu.Unlock()
|
||||
|
||||
if queued {
|
||||
// Fire outside the held job lock.
|
||||
m.background.Go(func() { m.fireQueued(ctx, jobID) })
|
||||
}
|
||||
}
|
||||
|
||||
// retryLater runs retry under the job's lock after recordRetryInterval,
|
||||
// unless the manager shuts down first.
|
||||
func (m *Manager) retryLater(jobID string, retry func()) {
|
||||
m.background.Go(func() {
|
||||
select {
|
||||
case <-m.stop:
|
||||
return
|
||||
case <-time.After(recordRetryInterval):
|
||||
}
|
||||
|
||||
m.locks.Lock(jobID)
|
||||
defer m.locks.Unlock(jobID)
|
||||
|
||||
retry()
|
||||
})
|
||||
}
|
||||
|
||||
// fireQueued fires a job's deferred fire. The fire is claimed under the job's
|
||||
// lock, so a pause or removal that takes the lock first discards it for good.
|
||||
func (m *Manager) fireQueued(ctx context.Context, jobID string) {
|
||||
m.locks.Lock(jobID)
|
||||
defer m.locks.Unlock(jobID)
|
||||
|
||||
m.mu.Lock()
|
||||
evidence := m.queued[jobID]
|
||||
delete(m.queued, jobID)
|
||||
m.mu.Unlock()
|
||||
|
||||
if evidence == nil {
|
||||
return
|
||||
}
|
||||
// A failure to fire is the queued trigger's outcome, reported like a
|
||||
// scheduler skip.
|
||||
if _, err := m.tryFireLocked(ctx, jobID, evidence); err != nil {
|
||||
log.G(ctx).WithError(err).WithFields(log.Fields{"job": jobID}).Warn("queued fire was not started")
|
||||
}
|
||||
// The deferred occurrence stayed recorded until now.
|
||||
m.persistNextFireLocked(ctx, jobID)
|
||||
}
|
||||
|
||||
// removeRunContainers removes the containers of dropped run records in the
|
||||
// background: a run container lives as long as its record, so neither history
|
||||
// eviction nor removing a job's runs leaves containers behind. In-flight runs
|
||||
// are skipped, as their containers cannot be removed while running; their
|
||||
// completion removes them instead.
|
||||
func (m *Manager) removeRunContainers(ctx context.Context, jobID string, runs []jobsv0.Run) {
|
||||
var ids []string
|
||||
|
||||
for _, run := range runs {
|
||||
if isTerminalRunState(run.State) && run.ContainerID != "" && !run.ContainerGone {
|
||||
ids = append(ids, run.ContainerID)
|
||||
}
|
||||
}
|
||||
|
||||
if len(ids) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
ctx = context.WithoutCancel(ctx)
|
||||
|
||||
m.background.Go(func() {
|
||||
for _, id := range ids {
|
||||
if err := m.backend.ContainerRm(ctx, id); err != nil && !cerrdefs.IsNotFound(err) {
|
||||
log.G(ctx).WithError(err).WithFields(log.Fields{"job": jobID, "container": id}).Warn("could not remove container of dropped run")
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// Cancel stops the job's in-flight run. It returns the in-flight run's ID,
|
||||
// empty when no run was in flight; triggers keep firing. When a timeout
|
||||
// claims the run's terminal decision first, the returned run is recorded
|
||||
// timed_out: the run's recorded terminal state is authoritative, not this
|
||||
// reply.
|
||||
func (m *Manager) Cancel(ctx context.Context, jobRef string) (string, error) {
|
||||
job, err := m.resolveJob(jobRef)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
m.locks.Lock(job.ID)
|
||||
defer m.locks.Unlock(job.ID)
|
||||
|
||||
run, err := m.store.LatestRun(job.ID)
|
||||
if err != nil {
|
||||
if cerrdefs.IsNotFound(err) {
|
||||
return "", nil
|
||||
}
|
||||
return "", err
|
||||
}
|
||||
if isTerminalRunState(run.State) {
|
||||
return "", nil
|
||||
}
|
||||
// A timeout or an earlier cancel may have claimed the terminal decision
|
||||
// already; the container is stopped regardless, as an earlier stop may
|
||||
// have failed.
|
||||
m.setOverride(run.ID, jobsv0.RunStateCancelled)
|
||||
// A pending run without a container cannot be observed here: container
|
||||
// creation happens synchronously under the job lock this method holds.
|
||||
containerID := run.ContainerID
|
||||
stopCtx := context.WithoutCancel(ctx)
|
||||
m.background.Go(func() {
|
||||
// Stopping can take the stop grace period; do not hold the job lock
|
||||
// for it. The watcher records the terminal state.
|
||||
if err := m.backend.ContainerStop(stopCtx, containerID); err != nil {
|
||||
log.G(stopCtx).WithError(err).WithFields(log.Fields{"job": job.ID, "run": run.ID}).Warn("could not stop cancelled run container")
|
||||
}
|
||||
})
|
||||
return run.ID, nil
|
||||
}
|
||||
|
||||
// armTimeout schedules the run's deadline: past it, the container is
|
||||
// stopped and the run ends timed_out (unless already cancelled).
|
||||
func (m *Manager) armTimeout(runID, containerID string, timeout time.Duration) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
// Shutdown disarmed the timers already; a timer armed after it would
|
||||
// stop a container that live-restore keeps running.
|
||||
if m.stopped {
|
||||
return
|
||||
}
|
||||
|
||||
m.timers[runID] = time.AfterFunc(timeout, func() {
|
||||
// Claim the timer entry and the override in one critical section:
|
||||
// a completion racing this firing has already disarmed the entry,
|
||||
// and claiming the override past that point would leak it forever
|
||||
// (nothing clears overrides of completed runs) and stop a fresh
|
||||
// container.
|
||||
m.mu.Lock()
|
||||
if _, armed := m.timers[runID]; !armed {
|
||||
m.mu.Unlock()
|
||||
return
|
||||
}
|
||||
delete(m.timers, runID)
|
||||
// A cancel may have claimed the terminal decision already; the
|
||||
// container is stopped regardless, as the cancel's stop may have
|
||||
// failed.
|
||||
if _, claimed := m.overrides[runID]; !claimed {
|
||||
m.overrides[runID] = jobsv0.RunStateTimedOut
|
||||
}
|
||||
// Register the stop while still holding the mutex: Shutdown
|
||||
// disarms timers under it before draining background work, so the
|
||||
// drain cannot miss the stop or close the backend under it.
|
||||
m.background.Go(func() {
|
||||
if err := m.backend.ContainerStop(context.Background(), containerID); err != nil {
|
||||
log.G(context.Background()).WithError(err).WithFields(log.Fields{"run": runID}).Warn("could not stop timed-out run container")
|
||||
}
|
||||
})
|
||||
m.mu.Unlock()
|
||||
})
|
||||
}
|
||||
|
||||
// disarmTimeout stops and forgets the run's timeout timer.
|
||||
func (m *Manager) disarmTimeout(runID string) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
if timer, armed := m.timers[runID]; armed {
|
||||
timer.Stop()
|
||||
delete(m.timers, runID)
|
||||
}
|
||||
}
|
||||
|
||||
// setOverride claims the run's terminal decision; the first claimant wins.
|
||||
func (m *Manager) setOverride(runID, state string) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
if _, claimed := m.overrides[runID]; !claimed {
|
||||
m.overrides[runID] = state
|
||||
}
|
||||
}
|
||||
|
||||
// peekOverride reads the run's forced terminal state without consuming it;
|
||||
// clearOverride forgets it once the terminal record is written.
|
||||
func (m *Manager) peekOverride(runID string) string {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
return m.overrides[runID]
|
||||
}
|
||||
|
||||
func (m *Manager) clearOverride(runID string) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
delete(m.overrides, runID)
|
||||
}
|
||||
|
||||
// subscribe returns a channel closed on the job's next state transition.
|
||||
// Callers that return without being woken must unsubscribe, or the entry
|
||||
// lingers until the job's next transition, which may never come.
|
||||
func (m *Manager) subscribe(jobID string) chan struct{} {
|
||||
ch := make(chan struct{})
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.notify[jobID] = append(m.notify[jobID], ch)
|
||||
return ch
|
||||
}
|
||||
|
||||
// unsubscribe withdraws a subscription that was not woken. A channel already
|
||||
// consumed by broadcast is simply no longer found.
|
||||
func (m *Manager) unsubscribe(jobID string, ch chan struct{}) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
subs := m.notify[jobID]
|
||||
for i, sub := range subs {
|
||||
if sub == ch {
|
||||
m.notify[jobID] = slices.Delete(subs, i, i+1)
|
||||
break
|
||||
}
|
||||
}
|
||||
if len(m.notify[jobID]) == 0 {
|
||||
delete(m.notify, jobID)
|
||||
}
|
||||
}
|
||||
|
||||
// broadcast wakes every waiter subscribed to the job.
|
||||
func (m *Manager) broadcast(jobID string) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
for _, ch := range m.notify[jobID] {
|
||||
close(ch)
|
||||
}
|
||||
delete(m.notify, jobID)
|
||||
}
|
||||
|
||||
// scheduleConcurrency returns the job's effective concurrency policy.
|
||||
func scheduleConcurrency(job *jobsv0.Job) string {
|
||||
if job.Spec != nil && job.Spec.Trigger != nil && job.Spec.Trigger.Schedule != nil && job.Spec.Trigger.Schedule.Concurrency != "" {
|
||||
return job.Spec.Trigger.Schedule.Concurrency
|
||||
}
|
||||
return jobsv0.ConcurrencyForbid
|
||||
}
|
||||
@@ -0,0 +1,377 @@
|
||||
package jobs
|
||||
|
||||
import (
|
||||
"container/heap"
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
cerrdefs "github.com/containerd/errdefs"
|
||||
"github.com/containerd/log"
|
||||
jobsv0 "github.com/moby/moby/v2/extpoints/jobs/api/v0"
|
||||
)
|
||||
|
||||
// schedulerIdleWait bounds the loop's sleep when no job is armed, so a
|
||||
// missed wake signal can only ever delay a fire, not lose it.
|
||||
const schedulerIdleWait = time.Minute
|
||||
|
||||
// schedEntry is one armed schedule job.
|
||||
type schedEntry struct {
|
||||
jobID string
|
||||
schedule *cronSchedule
|
||||
loc *time.Location
|
||||
next time.Time
|
||||
index int // position in the heap, -1 once removed
|
||||
}
|
||||
|
||||
// schedHeap is a min-heap of armed entries ordered by next fire time.
|
||||
type schedHeap []*schedEntry
|
||||
|
||||
func (h schedHeap) Len() int { return len(h) }
|
||||
func (h schedHeap) Less(i, j int) bool { return h[i].next.Before(h[j].next) }
|
||||
func (h schedHeap) Swap(i, j int) {
|
||||
h[i], h[j] = h[j], h[i]
|
||||
h[i].index = i
|
||||
h[j].index = j
|
||||
}
|
||||
|
||||
func (h *schedHeap) Push(x any) {
|
||||
entry := x.(*schedEntry)
|
||||
entry.index = len(*h)
|
||||
*h = append(*h, entry)
|
||||
}
|
||||
|
||||
func (h *schedHeap) Pop() any {
|
||||
old := *h
|
||||
entry := old[len(old)-1]
|
||||
old[len(old)-1] = nil
|
||||
entry.index = -1
|
||||
*h = old[:len(old)-1]
|
||||
return entry
|
||||
}
|
||||
|
||||
// scheduler fires schedule jobs on the daemon's clock: a single goroutine
|
||||
// sleeping until the earliest armed entry is due, dispatching fires through
|
||||
// the manager's tryFireLocked chokepoint like any other trigger.
|
||||
//
|
||||
// Lock order: the scheduler's mutex is taken while holding a job lock (Arm
|
||||
// and Disarm run under the manager's transitions) but never the other way
|
||||
// around — tick releases the mutex before dispatching, so fires and
|
||||
// persistence take job locks lock-free of the scheduler.
|
||||
type scheduler struct {
|
||||
m *Manager
|
||||
|
||||
mu sync.Mutex
|
||||
heap schedHeap
|
||||
byJob map[string]*schedEntry
|
||||
// gens holds the generation each job was last armed with. A tick's fire
|
||||
// is dropped when its job was disarmed, re-armed or rescheduled since,
|
||||
// so a delayed fire cannot run an occurrence of a superseded schedule or
|
||||
// one a manual run replaced. An entry that tick exhausts keeps its
|
||||
// generation: its last fire still stands.
|
||||
gens map[string]uint64
|
||||
gen uint64
|
||||
// pending counts each job's occurrences consumed by a tick whose fires
|
||||
// have not been claimed yet.
|
||||
pending map[string]int
|
||||
|
||||
wake chan struct{} // buffered kick after arm/disarm, so the loop re-plans
|
||||
stop chan struct{}
|
||||
}
|
||||
|
||||
func newScheduler(m *Manager) *scheduler {
|
||||
return &scheduler{
|
||||
m: m,
|
||||
byJob: make(map[string]*schedEntry),
|
||||
gens: make(map[string]uint64),
|
||||
pending: make(map[string]int),
|
||||
wake: make(chan struct{}, 1),
|
||||
stop: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
// arm registers (or re-registers) a job's next occurrence.
|
||||
func (s *scheduler) arm(jobID string, schedule *cronSchedule, loc *time.Location, next time.Time) {
|
||||
s.mu.Lock()
|
||||
|
||||
s.gen++
|
||||
s.gens[jobID] = s.gen
|
||||
delete(s.pending, jobID)
|
||||
|
||||
if entry, armed := s.byJob[jobID]; armed {
|
||||
entry.schedule, entry.loc, entry.next = schedule, loc, next
|
||||
heap.Fix(&s.heap, entry.index)
|
||||
} else {
|
||||
entry := &schedEntry{jobID: jobID, schedule: schedule, loc: loc, next: next}
|
||||
heap.Push(&s.heap, entry)
|
||||
s.byJob[jobID] = entry
|
||||
}
|
||||
s.mu.Unlock()
|
||||
s.kick()
|
||||
}
|
||||
|
||||
// disarm forgets a job. It is a no-op for unarmed jobs.
|
||||
func (s *scheduler) disarm(jobID string) {
|
||||
s.mu.Lock()
|
||||
delete(s.gens, jobID)
|
||||
delete(s.pending, jobID)
|
||||
|
||||
if entry, armed := s.byJob[jobID]; armed {
|
||||
heap.Remove(&s.heap, entry.index)
|
||||
delete(s.byJob, jobID)
|
||||
}
|
||||
s.mu.Unlock()
|
||||
s.kick()
|
||||
}
|
||||
|
||||
// replaceNext implements the reschedule semantics of the run API: a manual
|
||||
// fire stands in for the job's upcoming occurrence. When a tick already
|
||||
// consumed that occurrence and its fire is still waiting for the job's lock,
|
||||
// which the caller holds, the fire is dropped; otherwise the upcoming
|
||||
// occurrence is skipped. Either way exactly one occurrence is replaced. It
|
||||
// reports whether the job was armed.
|
||||
func (s *scheduler) replaceNext(jobID string) bool {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
if _, armed := s.gens[jobID]; !armed {
|
||||
return false
|
||||
}
|
||||
|
||||
if s.pending[jobID] > 0 {
|
||||
s.gen++
|
||||
s.gens[jobID] = s.gen
|
||||
delete(s.pending, jobID)
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
entry, armed := s.byJob[jobID]
|
||||
if !armed {
|
||||
return true
|
||||
}
|
||||
|
||||
if next, ok := entry.schedule.next(entry.next, entry.loc); ok {
|
||||
entry.next = next
|
||||
heap.Fix(&s.heap, entry.index)
|
||||
} else {
|
||||
heap.Remove(&s.heap, entry.index)
|
||||
delete(s.byJob, jobID)
|
||||
}
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
// nextFor reports a job's armed occurrence.
|
||||
func (s *scheduler) nextFor(jobID string) (time.Time, bool) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if entry, armed := s.byJob[jobID]; armed {
|
||||
return entry.next, true
|
||||
}
|
||||
return time.Time{}, false
|
||||
}
|
||||
|
||||
// claimDue claims a fire a tick dispatched, reporting false when its job was
|
||||
// disarmed, re-armed or rescheduled since.
|
||||
func (s *scheduler) claimDue(jobID string, gen uint64) bool {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
if s.gens[jobID] != gen {
|
||||
return false
|
||||
}
|
||||
|
||||
s.pending[jobID]--
|
||||
if s.pending[jobID] <= 0 {
|
||||
delete(s.pending, jobID)
|
||||
}
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
// kick nudges the loop to re-plan its sleep after the heap changed.
|
||||
func (s *scheduler) kick() {
|
||||
select {
|
||||
case s.wake <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
// dueFire is an occurrence a tick consumed, with the generation its job was
|
||||
// armed with.
|
||||
type dueFire struct {
|
||||
jobID string
|
||||
gen uint64
|
||||
scheduledAt time.Time
|
||||
}
|
||||
|
||||
// tick fires every due entry and returns how long the loop may sleep before
|
||||
// the next one. Dispatch happens after the scheduler lock is released: fires
|
||||
// take job locks, which are held around arm/disarm calls into the scheduler.
|
||||
func (s *scheduler) tick(ctx context.Context) time.Duration {
|
||||
fires, wait := s.takeDue(s.m.now())
|
||||
for _, fire := range fires {
|
||||
// Firing takes the job's lock, which a run may hold across container
|
||||
// operations; doing it here would stall every other schedule.
|
||||
s.m.background.Go(func() { s.m.fireDue(ctx, fire.jobID, fire.gen, fire.scheduledAt) })
|
||||
}
|
||||
|
||||
return wait
|
||||
}
|
||||
|
||||
// takeDue consumes the occurrences due at now and returns how long the loop
|
||||
// may sleep before the next one.
|
||||
func (s *scheduler) takeDue(now time.Time) ([]dueFire, time.Duration) {
|
||||
var fires []dueFire
|
||||
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
// An entry overdue by several occurrences (a long sleep or system
|
||||
// suspend) fires once and re-arms from now: the intermediate backlog is
|
||||
// deliberately dropped, mirroring how the missed-fires policy caps the
|
||||
// catch-up at one run across a daemon restart.
|
||||
for len(s.heap) > 0 && !s.heap[0].next.After(now) {
|
||||
entry := s.heap[0]
|
||||
fires = append(fires, dueFire{jobID: entry.jobID, gen: s.gens[entry.jobID], scheduledAt: entry.next})
|
||||
s.pending[entry.jobID]++
|
||||
|
||||
if next, ok := entry.schedule.next(now, entry.loc); ok {
|
||||
entry.next = next
|
||||
heap.Fix(&s.heap, entry.index)
|
||||
} else {
|
||||
// The expression has no future occurrence (e.g. a specific
|
||||
// date now past): the job stays registered but disarmed.
|
||||
heap.Remove(&s.heap, entry.index)
|
||||
delete(s.byJob, entry.jobID)
|
||||
}
|
||||
}
|
||||
wait := schedulerIdleWait
|
||||
if len(s.heap) > 0 {
|
||||
wait = min(s.heap[0].next.Sub(now), schedulerIdleWait)
|
||||
}
|
||||
|
||||
return fires, wait
|
||||
}
|
||||
|
||||
// run is the scheduler loop; it exits when stop is closed.
|
||||
func (s *scheduler) run(ctx context.Context) {
|
||||
timer := time.NewTimer(0)
|
||||
defer timer.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-s.stop:
|
||||
return
|
||||
case <-s.wake:
|
||||
case <-timer.C:
|
||||
}
|
||||
if !timer.Stop() {
|
||||
// Drain a concurrently fired timer so Reset arms cleanly.
|
||||
select {
|
||||
case <-timer.C:
|
||||
default:
|
||||
}
|
||||
}
|
||||
timer.Reset(s.tick(ctx))
|
||||
}
|
||||
}
|
||||
|
||||
// fireDue fires the occurrence a tick consumed and records the job's next
|
||||
// one, unless the job was disarmed, re-armed or rescheduled since the tick: a
|
||||
// pause, resume, removal or reschedule taking the job's lock first
|
||||
// supersedes the fire.
|
||||
func (m *Manager) fireDue(ctx context.Context, jobID string, gen uint64, scheduledAt time.Time) {
|
||||
m.locks.Lock(jobID)
|
||||
defer m.locks.Unlock(jobID)
|
||||
|
||||
if !m.sched.claimDue(jobID, gen) {
|
||||
return
|
||||
}
|
||||
|
||||
m.fireScheduledLocked(ctx, jobID, scheduledAt)
|
||||
}
|
||||
|
||||
// fireScheduled routes a cron fire through the tryFireLocked chokepoint and
|
||||
// turns its refusals into skip logs: for a scheduler, a fire refused by the
|
||||
// concurrency policy is normal operation, not an error. The job's next
|
||||
// occurrence is recorded only after the fire, so a crash in between leaves
|
||||
// the occurrence to the restart's catch-up instead of losing it. A fire
|
||||
// deferred by the queue policy leaves the occurrence recorded until the
|
||||
// deferred fire runs, for the same reason.
|
||||
func (m *Manager) fireScheduled(ctx context.Context, jobID string, scheduledAt time.Time) {
|
||||
m.locks.Lock(jobID)
|
||||
defer m.locks.Unlock(jobID)
|
||||
|
||||
m.fireScheduledLocked(ctx, jobID, scheduledAt)
|
||||
}
|
||||
|
||||
// fireScheduledLocked is fireScheduled for callers already holding the job's
|
||||
// lock.
|
||||
func (m *Manager) fireScheduledLocked(ctx context.Context, jobID string, scheduledAt time.Time) {
|
||||
evidence := &jobsv0.TriggerEvidence{
|
||||
Kind: jobsv0.TriggerKindSchedule,
|
||||
ScheduledAtNano: scheduledAt.UnixNano(),
|
||||
FiredAtNano: m.now().UnixNano(),
|
||||
}
|
||||
|
||||
_, err := m.tryFireLocked(ctx, jobID, evidence)
|
||||
switch {
|
||||
case err == nil:
|
||||
case errors.Is(err, errFirePaused):
|
||||
log.G(ctx).WithFields(log.Fields{"job": jobID}).Debug("scheduled fire skipped: the job is paused")
|
||||
case errors.Is(err, errFireQueued):
|
||||
log.G(ctx).WithFields(log.Fields{"job": jobID}).Debug("scheduled fire queued behind the current run")
|
||||
|
||||
return
|
||||
case errors.Is(err, errFireDropped):
|
||||
log.G(ctx).WithFields(log.Fields{"job": jobID}).Warn("scheduled fire dropped: a fire is already queued")
|
||||
|
||||
return
|
||||
case cerrdefs.IsFailedPrecondition(err):
|
||||
log.G(ctx).WithFields(log.Fields{"job": jobID}).Warn("scheduled fire skipped: the previous run is still in flight")
|
||||
case cerrdefs.IsNotFound(err):
|
||||
// The job was removed while the fire was in flight; the removal
|
||||
// already disarmed it.
|
||||
default:
|
||||
log.G(ctx).WithError(err).WithFields(log.Fields{"job": jobID}).Warn("scheduled fire failed")
|
||||
}
|
||||
|
||||
m.persistNextFireLocked(ctx, jobID)
|
||||
}
|
||||
|
||||
// persistNextFire records a job's armed occurrence (zero for none) on its
|
||||
// stored record, where inspect and list read it from, and where Start finds
|
||||
// the fires missed while the daemon was down.
|
||||
func (m *Manager) persistNextFire(ctx context.Context, jobID string) {
|
||||
m.locks.Lock(jobID)
|
||||
defer m.locks.Unlock(jobID)
|
||||
|
||||
m.persistNextFireLocked(ctx, jobID)
|
||||
}
|
||||
|
||||
// persistNextFireLocked is persistNextFire for callers already holding the
|
||||
// job's lock. It reads the occurrence from the scheduler under that lock
|
||||
// instead of taking it from the caller, so a delayed write cannot replace a
|
||||
// newer one.
|
||||
func (m *Manager) persistNextFireLocked(ctx context.Context, jobID string) {
|
||||
var atNano int64
|
||||
if next, armed := m.sched.nextFor(jobID); armed {
|
||||
atNano = next.UnixNano()
|
||||
}
|
||||
|
||||
job, err := m.store.Job(jobID)
|
||||
if err != nil {
|
||||
return // removed meanwhile; nothing to record on
|
||||
}
|
||||
if job.NextFireAtNano == atNano {
|
||||
return
|
||||
}
|
||||
|
||||
job.NextFireAtNano = atNano
|
||||
job.UpdatedAtNano = m.now().UnixNano()
|
||||
if err := m.store.UpdateJob(ctx, job); err != nil {
|
||||
log.G(ctx).WithError(err).WithFields(log.Fields{"job": jobID}).Warn("could not record next fire time")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,373 @@
|
||||
package jobs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
cerrdefs "github.com/containerd/errdefs"
|
||||
jobsv0 "github.com/moby/moby/v2/extpoints/jobs/api/v0"
|
||||
"gotest.tools/v3/assert"
|
||||
is "gotest.tools/v3/assert/cmp"
|
||||
)
|
||||
|
||||
// fakeClock is an injectable manager clock; the scheduler is exercised by
|
||||
// advancing it and calling tick directly, so no test ever sleeps on cron
|
||||
// granularity.
|
||||
type fakeClock struct {
|
||||
mu sync.Mutex
|
||||
t time.Time
|
||||
}
|
||||
|
||||
func (c *fakeClock) Now() time.Time {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
return c.t
|
||||
}
|
||||
|
||||
func (c *fakeClock) Advance(d time.Duration) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.t = c.t.Add(d)
|
||||
}
|
||||
|
||||
// newTestManagerAt builds a manager whose clock starts at the given instant.
|
||||
func newTestManagerAt(t *testing.T, start time.Time) (*Manager, *fakeBackend, *fakeClock) {
|
||||
t.Helper()
|
||||
m, fake := newTestManager(t)
|
||||
clk := &fakeClock{t: start}
|
||||
m.now = clk.Now
|
||||
return m, fake, clk
|
||||
}
|
||||
|
||||
// jan15 is an arbitrary fixed reference: 2026-01-15 00:00 UTC, a Thursday.
|
||||
var jan15 = time.Date(2026, time.January, 15, 0, 0, 0, 0, time.UTC)
|
||||
|
||||
func TestSchedulerFiresAndRearms(t *testing.T) {
|
||||
m, fake, clk := newTestManagerAt(t, jan15)
|
||||
job, _, err := m.Create(t.Context(), "nightly", scheduleSpec(""))
|
||||
assert.NilError(t, err)
|
||||
|
||||
// Registration armed the trigger and advertised the next occurrence.
|
||||
assert.Check(t, is.Equal(job.NextFireAtNano, jan15.Add(3*time.Hour).UnixNano()))
|
||||
|
||||
// Before the occurrence, a tick fires nothing.
|
||||
m.sched.tick(t.Context())
|
||||
_, err = m.store.LatestRun(job.ID)
|
||||
assert.Check(t, cerrdefs.IsNotFound(err))
|
||||
|
||||
// Past the occurrence, a tick fires exactly one run with schedule
|
||||
// evidence carrying the planned time.
|
||||
clk.Advance(3*time.Hour + 10*time.Second)
|
||||
m.sched.tick(t.Context())
|
||||
run, err := m.Wait(t.Context(), job.ID, "latest", jobsv0.WaitConditionRunning)
|
||||
assert.NilError(t, err)
|
||||
assert.Check(t, is.Equal(run.Trigger.Kind, jobsv0.TriggerKindSchedule))
|
||||
assert.Check(t, is.Equal(run.Trigger.ScheduledAtNano, jan15.Add(3*time.Hour).UnixNano()))
|
||||
|
||||
// While the run is in flight the record advertises no next fire.
|
||||
inFlight, err := m.Inspect(t.Context(), job.ID)
|
||||
assert.NilError(t, err)
|
||||
assert.Check(t, is.Equal(inFlight.NextFireAtNano, int64(0)))
|
||||
|
||||
fake.exit(run.ContainerID, 0, nil)
|
||||
done := waitTerminal(t, m, job.ID, run.ID)
|
||||
assert.Check(t, is.Equal(done.State, jobsv0.RunStateSucceeded))
|
||||
|
||||
// After completion the job advertises the re-armed next occurrence.
|
||||
rearmed := jan15.Add(27 * time.Hour).UnixNano()
|
||||
waitJobSettled(t, m, job.ID, func(j *jobsv0.Job) bool { return j.NextFireAtNano == rearmed })
|
||||
}
|
||||
|
||||
func TestSchedulerForbidSkipsWhileRunning(t *testing.T) {
|
||||
m, fake, clk := newTestManagerAt(t, jan15)
|
||||
job, _, err := m.Create(t.Context(), "nightly", scheduleSpec(""))
|
||||
assert.NilError(t, err)
|
||||
|
||||
clk.Advance(3*time.Hour + time.Second)
|
||||
m.sched.tick(t.Context())
|
||||
first, err := m.Wait(t.Context(), job.ID, "latest", jobsv0.WaitConditionRunning)
|
||||
assert.NilError(t, err)
|
||||
|
||||
// The next occurrence comes due while the first run is still in
|
||||
// flight: forbid (the default) drops the fire. The fire is driven
|
||||
// synchronously here — tick's asynchronous dispatch of due entries is
|
||||
// covered above — so the count below cannot race the dispatch.
|
||||
clk.Advance(24 * time.Hour)
|
||||
m.fireScheduled(t.Context(), job.ID, jan15.Add(27*time.Hour))
|
||||
page, _, _, err := m.store.Runs(job.ID, 10, "")
|
||||
assert.NilError(t, err)
|
||||
assert.Check(t, is.Equal(len(page), 1))
|
||||
|
||||
fake.exit(first.ContainerID, 0, nil)
|
||||
waitTerminal(t, m, job.ID, first.ID)
|
||||
}
|
||||
|
||||
// shutdownCtx bounds the drain waits used mid-test.
|
||||
func shutdownCtx(t *testing.T) context.Context {
|
||||
t.Helper()
|
||||
ctx, cancel := context.WithTimeout(context.WithoutCancel(t.Context()), 10*time.Second)
|
||||
t.Cleanup(cancel)
|
||||
return ctx
|
||||
}
|
||||
|
||||
func TestSchedulerPauseResume(t *testing.T) {
|
||||
m, _, clk := newTestManagerAt(t, jan15)
|
||||
job, _, err := m.Create(t.Context(), "nightly", scheduleSpec(""))
|
||||
assert.NilError(t, err)
|
||||
|
||||
assert.NilError(t, m.Pause(t.Context(), job.ID))
|
||||
paused, err := m.Inspect(t.Context(), job.ID)
|
||||
assert.NilError(t, err)
|
||||
assert.Check(t, is.Equal(paused.NextFireAtNano, int64(0)))
|
||||
|
||||
// The occurrence passes while paused: nothing fires, and resuming
|
||||
// re-arms from the next tick without backfilling.
|
||||
clk.Advance(4 * time.Hour)
|
||||
m.sched.tick(t.Context())
|
||||
_, err = m.store.LatestRun(job.ID)
|
||||
assert.Check(t, cerrdefs.IsNotFound(err))
|
||||
|
||||
assert.NilError(t, m.Resume(t.Context(), job.ID))
|
||||
resumed, err := m.Inspect(t.Context(), job.ID)
|
||||
assert.NilError(t, err)
|
||||
assert.Check(t, is.Equal(resumed.NextFireAtNano, jan15.Add(27*time.Hour).UnixNano()))
|
||||
}
|
||||
|
||||
func TestSchedulerMissedFiresOnStart(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
doc string
|
||||
policy string
|
||||
expectCatchUp bool
|
||||
}{
|
||||
{doc: "one fires a single catch-up", policy: "", expectCatchUp: true},
|
||||
{doc: "skip drops missed fires", policy: jobsv0.MissedFiresSkip, expectCatchUp: false},
|
||||
} {
|
||||
t.Run(tc.doc, func(t *testing.T) {
|
||||
m, fake, clk := newTestManagerAt(t, jan15)
|
||||
spec := scheduleSpec("")
|
||||
spec.Trigger.Schedule.MissedFires = tc.policy
|
||||
job, _, err := m.Create(t.Context(), "nightly", spec)
|
||||
assert.NilError(t, err)
|
||||
|
||||
// The daemon was down across the 03:00 occurrence: the clock
|
||||
// advances but no tick observed it. Start applies the policy
|
||||
// from the persisted next-fire time.
|
||||
clk.Advance(5 * time.Hour)
|
||||
m.Start(t.Context())
|
||||
|
||||
if tc.expectCatchUp {
|
||||
run, err := m.Wait(t.Context(), job.ID, "latest", jobsv0.WaitConditionRunning)
|
||||
assert.NilError(t, err)
|
||||
assert.Check(t, is.Equal(run.Trigger.ScheduledAtNano, jan15.Add(3*time.Hour).UnixNano()))
|
||||
fake.exit(run.ContainerID, 0, nil)
|
||||
waitTerminal(t, m, job.ID, run.ID)
|
||||
} else {
|
||||
assert.NilError(t, m.Shutdown(shutdownCtx(t)))
|
||||
_, err := m.store.LatestRun(job.ID)
|
||||
assert.Check(t, cerrdefs.IsNotFound(err))
|
||||
}
|
||||
|
||||
// Either way the schedule re-armed from the next occurrence,
|
||||
// not from the backlog. The catch-up fire runs concurrently
|
||||
// with Start's own arm+persist, so the re-arm write can land
|
||||
// an instant after the run turns terminal.
|
||||
rearmed := jan15.Add(27 * time.Hour).UnixNano()
|
||||
waitJobSettled(t, m, job.ID, func(j *jobsv0.Job) bool { return j.NextFireAtNano == rearmed })
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestSchedulerReschedule(t *testing.T) {
|
||||
m, fake, clk := newTestManagerAt(t, jan15)
|
||||
job, _, err := m.Create(t.Context(), "nightly", scheduleSpec(""))
|
||||
assert.NilError(t, err)
|
||||
|
||||
// A manual fire with reschedule stands in for the upcoming occurrence:
|
||||
// the schedule skips 03:00 today and re-arms on tomorrow's.
|
||||
run, err := m.Run(t.Context(), job.ID, true)
|
||||
assert.NilError(t, err)
|
||||
fake.exit(run.ContainerID, 0, nil)
|
||||
waitTerminal(t, m, job.ID, run.ID)
|
||||
|
||||
rearmed := jan15.Add(27 * time.Hour).UnixNano()
|
||||
waitJobSettled(t, m, job.ID, func(j *jobsv0.Job) bool { return j.NextFireAtNano == rearmed })
|
||||
|
||||
// Today's occurrence passes without a fire.
|
||||
clk.Advance(4 * time.Hour)
|
||||
m.sched.tick(t.Context())
|
||||
assert.NilError(t, m.Shutdown(shutdownCtx(t)))
|
||||
page, _, _, err := m.store.Runs(job.ID, 10, "")
|
||||
assert.NilError(t, err)
|
||||
assert.Check(t, is.Equal(len(page), 1))
|
||||
}
|
||||
|
||||
func TestSchedulerRemoveDisarms(t *testing.T) {
|
||||
m, _, clk := newTestManagerAt(t, jan15)
|
||||
job, _, err := m.Create(t.Context(), "nightly", scheduleSpec(""))
|
||||
assert.NilError(t, err)
|
||||
|
||||
assert.NilError(t, m.Remove(t.Context(), job.ID, ""))
|
||||
clk.Advance(4 * time.Hour)
|
||||
m.sched.tick(t.Context())
|
||||
assert.NilError(t, m.Shutdown(shutdownCtx(t)))
|
||||
_, err = m.store.LatestRun(job.ID)
|
||||
assert.Check(t, cerrdefs.IsNotFound(err), "a removed job must not fire")
|
||||
}
|
||||
|
||||
func TestSchedulerStartWithUnparseableStoredCron(t *testing.T) {
|
||||
// A schedule the validation would never accept can still sit in the
|
||||
// store (written by a newer or buggy daemon); Start must leave the job
|
||||
// disarmed instead of failing the whole startup.
|
||||
store, _ := newTestStore(t)
|
||||
job := makeJob("job1", "broken")
|
||||
job.Spec.Trigger.Schedule.Cron = "not a cron"
|
||||
assert.NilError(t, store.CreateJob(t.Context(), job))
|
||||
|
||||
m := NewManager(store, newFakeBackend(), testNameGenerator)
|
||||
clk := &fakeClock{t: jan15}
|
||||
m.now = clk.Now
|
||||
m.Start(t.Context())
|
||||
t.Cleanup(func() { assert.Check(t, m.Shutdown(shutdownCtx(t))) })
|
||||
|
||||
clk.Advance(24 * time.Hour)
|
||||
m.sched.tick(t.Context())
|
||||
_, err := m.store.LatestRun("job1")
|
||||
assert.Check(t, cerrdefs.IsNotFound(err), "a job with an unparseable schedule must never fire")
|
||||
}
|
||||
|
||||
func TestSchedulerHonorsTimezone(t *testing.T) {
|
||||
m, _, _ := newTestManagerAt(t, jan15)
|
||||
spec := scheduleSpec("")
|
||||
spec.Trigger.Schedule.Timezone = "Europe/Paris"
|
||||
job, _, err := m.Create(t.Context(), "paris-nightly", spec)
|
||||
assert.NilError(t, err)
|
||||
|
||||
// 03:00 Paris is 02:00 UTC in January.
|
||||
paris, err := time.LoadLocation("Europe/Paris")
|
||||
assert.NilError(t, err)
|
||||
assert.Check(t, is.Equal(job.NextFireAtNano, time.Date(2026, time.January, 15, 3, 0, 0, 0, paris).UnixNano()))
|
||||
assert.Check(t, is.Equal(job.NextFireAtNano, jan15.Add(2*time.Hour).UnixNano()))
|
||||
}
|
||||
|
||||
func TestSchedulerCreateRejectsNeverFiringCron(t *testing.T) {
|
||||
m, _ := newTestManager(t)
|
||||
spec := scheduleSpec("")
|
||||
spec.Trigger.Schedule.Cron = "0 0 30 2 *"
|
||||
_, _, err := m.Create(t.Context(), "never", spec)
|
||||
assert.Check(t, cerrdefs.IsInvalidArgument(err), "want invalid-argument, got %v", err)
|
||||
}
|
||||
|
||||
// TestSchedulerStartKeepsRescheduledOccurrence checks that a restart keeps the
|
||||
// occurrence a reschedule persisted instead of firing the one it skipped, and
|
||||
// keeps no later one.
|
||||
func TestSchedulerStartKeepsRescheduledOccurrence(t *testing.T) {
|
||||
schedule, err := parseCron("0 3 * * *")
|
||||
assert.NilError(t, err)
|
||||
|
||||
upcoming, _ := schedule.next(time.Now(), time.UTC)
|
||||
rescheduled, _ := schedule.next(upcoming, time.UTC)
|
||||
|
||||
for _, tc := range []struct {
|
||||
doc string
|
||||
stored time.Time
|
||||
expected time.Time
|
||||
}{
|
||||
{doc: "the occurrence after the upcoming one is kept", stored: rescheduled, expected: rescheduled},
|
||||
{doc: "a later occurrence is recomputed", stored: rescheduled.Add(24 * time.Hour), expected: upcoming},
|
||||
} {
|
||||
t.Run(tc.doc, func(t *testing.T) {
|
||||
store, _ := newTestStore(t)
|
||||
job := makeJob("job1", "nightly")
|
||||
job.NextFireAtNano = tc.stored.UnixNano()
|
||||
assert.NilError(t, store.CreateJob(t.Context(), job))
|
||||
|
||||
m := NewManager(store, newFakeBackend(), testNameGenerator)
|
||||
m.Start(t.Context())
|
||||
t.Cleanup(func() { assert.Check(t, m.Shutdown(shutdownCtx(t))) })
|
||||
|
||||
next, armed := m.sched.nextFor(job.ID)
|
||||
assert.Assert(t, armed)
|
||||
assert.Check(t, is.Equal(next.UnixNano(), tc.expected.UnixNano()))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestSchedulerDropsSupersededFire checks that a tick's fire delayed past a
|
||||
// pause neither overwrites the cleared next-fire time nor, after a resume,
|
||||
// runs the occurrence of the superseded schedule.
|
||||
func TestSchedulerDropsSupersededFire(t *testing.T) {
|
||||
m, _ := newTestManager(t)
|
||||
job, _, err := m.Create(t.Context(), "nightly", scheduleSpec(""))
|
||||
assert.NilError(t, err)
|
||||
|
||||
m.sched.mu.Lock()
|
||||
gen := m.sched.gens[job.ID]
|
||||
m.sched.mu.Unlock()
|
||||
|
||||
scheduledAt := time.Unix(0, job.NextFireAtNano)
|
||||
|
||||
assert.NilError(t, m.Pause(t.Context(), job.ID))
|
||||
m.fireDue(t.Context(), job.ID, gen, scheduledAt)
|
||||
paused, err := m.store.Job(job.ID)
|
||||
assert.NilError(t, err)
|
||||
assert.Check(t, is.Equal(paused.NextFireAtNano, int64(0)))
|
||||
|
||||
assert.NilError(t, m.Resume(t.Context(), job.ID))
|
||||
m.fireDue(t.Context(), job.ID, gen, scheduledAt)
|
||||
_, err = m.store.LatestRun(job.ID)
|
||||
assert.Check(t, cerrdefs.IsNotFound(err), "a superseded fire must not run")
|
||||
}
|
||||
|
||||
// TestSchedulerRescheduleReplacesConsumedOccurrence checks that a reschedule
|
||||
// landing after a tick consumed the upcoming occurrence, but before its fire
|
||||
// ran, replaces that occurrence instead of letting it fire and skipping the
|
||||
// following one too.
|
||||
func TestSchedulerRescheduleReplacesConsumedOccurrence(t *testing.T) {
|
||||
m, fake := newTestManager(t)
|
||||
job, _, err := m.Create(t.Context(), "nightly", scheduleSpec(""))
|
||||
assert.NilError(t, err)
|
||||
|
||||
fires, _ := m.sched.takeDue(time.Unix(0, job.NextFireAtNano))
|
||||
assert.Assert(t, is.Len(fires, 1))
|
||||
|
||||
following, _ := m.sched.nextFor(job.ID)
|
||||
|
||||
run, err := m.Run(t.Context(), job.ID, true)
|
||||
assert.NilError(t, err)
|
||||
fake.exit(run.ContainerID, 0, nil)
|
||||
waitTerminal(t, m, job.ID, run.ID)
|
||||
waitJobSettled(t, m, job.ID, func(j *jobsv0.Job) bool { return j.State == jobsv0.JobStateIdle })
|
||||
|
||||
m.fireDue(t.Context(), fires[0].jobID, fires[0].gen, fires[0].scheduledAt)
|
||||
page, _, _, err := m.store.Runs(job.ID, 10, "")
|
||||
assert.NilError(t, err)
|
||||
assert.Check(t, is.Len(page, 1), "the replaced occurrence must not fire")
|
||||
|
||||
next, _ := m.sched.nextFor(job.ID)
|
||||
assert.Check(t, is.Equal(next.UnixNano(), following.UnixNano()))
|
||||
}
|
||||
|
||||
// TestSchedulerStartSkipsCatchUpAlreadyFired checks that a restart does not
|
||||
// catch up an occurrence whose run was recorded before the daemon died, ahead
|
||||
// of the next fire time.
|
||||
func TestSchedulerStartSkipsCatchUpAlreadyFired(t *testing.T) {
|
||||
store, _ := newTestStore(t)
|
||||
missed := time.Now().Add(-time.Hour).UnixNano()
|
||||
job := makeJob("job1", "nightly")
|
||||
job.NextFireAtNano = missed
|
||||
assert.NilError(t, store.CreateJob(t.Context(), job))
|
||||
|
||||
run := makeRun(job.ID, "run1", jobsv0.RunStateSucceeded)
|
||||
run.Trigger = &jobsv0.TriggerEvidence{Kind: jobsv0.TriggerKindSchedule, ScheduledAtNano: missed, FiredAtNano: missed}
|
||||
assert.NilError(t, store.CreateRun(t.Context(), run))
|
||||
|
||||
m := NewManager(store, newFakeBackend(), testNameGenerator)
|
||||
m.Start(t.Context())
|
||||
assert.NilError(t, m.Shutdown(shutdownCtx(t)))
|
||||
|
||||
page, _, _, err := m.store.Runs(job.ID, 10, "")
|
||||
assert.NilError(t, err)
|
||||
assert.Check(t, is.Len(page, 1))
|
||||
}
|
||||
Reference in new issue
Block a user