jobs

Documentation

Overview

Package jobs runs keyforge's lifecycle work as durable jobs on OmniWorkflow.

Rotation, re-wrapping, retirement and destruction are already idempotent and resumable operations. This package supplies what turns them into a system that runs unattended: a scheduler that notices what is due, and the omniworkflow.StepExecutor that does each unit of work, so that any OmniWorkflow provider (in-memory, durable, Temporal, Step Functions) can run it and restart it.

Exactly one execution

Every job has a deterministic run ID built from what it does and to what. OmniWorkflow's Start is idempotent by run ID, so several replicas that each find the same work due all start the same run, and one execution results. There is no leader election and no lease in keyforge.

Time

A plan has no timers, so time is the scheduler's: it evaluates what is due on each tick and starts a short job for it. A retirement delay, a guard that is still pending, and a version waiting to be destroyed are all conditions the next tick finds still true or no longer true.

Index

• Constants
• Variables
• func Plan(s Spec) omniworkflow.Plan
• type Embedded
  • func NewEmbedded(o EmbeddedOptions) (*Embedded, error)
  • func (e *Embedded) Run(ctx context.Context) error
  • func (e *Embedded) Work(ctx context.Context) error
• type EmbeddedOptions
• type Executor
  • func NewExecutor(r *rotate.Rotator, s registry.Store) *Executor
  • func (e *Executor) ExecuteStep(ctx context.Context, _ string, step omniworkflow.PlanStep) (any, error)
• type FailedJob
• type Kind
• type Overdue
• type Scheduler
  • func NewScheduler(o SchedulerOptions) (*Scheduler, error)
  • func (s *Scheduler) Last() TickResult
  • func (s *Scheduler) Run(ctx context.Context) error
  • func (s *Scheduler) Tick(ctx context.Context) (TickResult, error)
• type SchedulerOptions
• type Spec
• type Started
• type TickResult

Constants

const (
	// DefaultInterval is how often the scheduler looks for work.
	DefaultInterval = time.Minute
	// DefaultRecheck is how long to wait before starting a job again for a
	// condition that is still true after the last one succeeded, such as a guard
	// that is still pending. It keeps a long wait for an approval from filling
	// the run store with identical runs.
	DefaultRecheck = 10 * time.Minute
	// DefaultBackoff and DefaultMaxBackoff bound the wait before retrying a job
	// that failed. The wait doubles with each consecutive failure.
	DefaultBackoff    = time.Minute
	DefaultMaxBackoff = time.Hour
)

Defaults for SchedulerOptions.

Variables

var (
	// ErrTooManyAttempts is returned when a job has been attempted so often that
	// the scheduler stops numbering new attempts.
	ErrTooManyAttempts = errors.New("jobs: too many attempts of one job")
)

Errors reported by the scheduler.

Functions

func Plan

func Plan(s Spec) omniworkflow.Plan

Plan returns the plan that does the work a spec describes.

Types

type Embedded

type Embedded struct {
	// Scheduler is the scheduler, for ticking by hand and for reading its
	// results.
	Scheduler *Scheduler
	// Provider is the provider running the jobs.
	Provider omniworkflow.WorkflowProvider
	// contains filtered or unexported fields
}

Embedded runs keyforge's lifecycle jobs inside the host process: the scheduler, an OmniWorkflow provider and its worker. It is the default way to run keyforge unattended on one host.

The durable provider is single-process. Deployments with several replicas give the scheduler and the worker a shared provider such as Temporal or Step Functions instead, with the same Scheduler and Executor.

func NewEmbedded

func NewEmbedded(o EmbeddedOptions) (*Embedded, error)

NewEmbedded wires a provider, a worker running the executor, and a scheduler. SchedulerOptions.Provider is ignored.

func (*Embedded) Run

func (e *Embedded) Run(ctx context.Context) error

Run starts the worker, which first resumes any run a previous process left unfinished, and the scheduler's loop, and blocks until ctx is cancelled. The first error ends both.

func (*Embedded) Work

func (e *Embedded) Work(ctx context.Context) error

Work runs only the worker: it resumes any run a previous process left unfinished and executes steps until ctx is cancelled. The scheduler is not ticked, so the host drives it by calling Scheduler.Tick from its own timer or cron. Use Run to have the embedded scheduler tick itself.

type EmbeddedOptions

type EmbeddedOptions struct {
	SchedulerOptions
	// StateDir is where the durable provider keeps run state, so jobs resume
	// after a restart. Empty uses the in-memory provider, which keeps nothing:
	// right for tests, and for a deployment that can lose a run because the next
	// tick starts another.
	StateDir string
}

EmbeddedOptions configures an Embedded runner: a scheduler, a provider and a worker in one process.

type Executor

type Executor struct {
	// contains filtered or unexported fields
}

Executor does the work of each step by calling the rotation operations. It is the omniworkflow.StepExecutor a provider's worker is given.

Steps are delivered at least once, and every operation here is safe to repeat: a rotation re-checks that it is due, a re-wrap moves only what has not moved, and the guarded steps ask the guard again.

func NewExecutor

func NewExecutor(r *rotate.Rotator, s registry.Store) *Executor

NewExecutor returns an executor over a rotator and its registry.

func (*Executor) ExecuteStep

func (e *Executor) ExecuteStep(ctx context.Context, _ string, step omniworkflow.PlanStep) (any, error)

ExecuteStep implements omniworkflow.StepExecutor. A step that did its work returns a small, JSON-serializable summary. A step that could not returns an error, which fails the run; the scheduler reports it and tries again.

type FailedJob

type FailedJob struct {
	RunID string
	Spec  Spec
	// Consecutive is how many attempts in a row have failed.
	Consecutive int
}

FailedJob describes a run that failed.

type Kind

type Kind string

Kind names a unit of lifecycle work.

const (
	// RotateKEK activates the next KEK version for a purpose and then moves
	// every data key off the older ones.
	RotateKEK Kind = "rotate-kek"
	// Rewrap moves data keys off every non-active version of a KEK.
	Rewrap Kind = "rewrap"
	// RequestRetirement starts the retirement delay for a version outside the
	// retained floor.
	RequestRetirement Kind = "request-retirement"
	// CompleteRetirement retires a version whose delay has passed, if the guard
	// allows.
	CompleteRetirement Kind = "complete-retirement"
	// Destroy destroys a retired version's material, if the guard allows.
	Destroy Kind = "destroy"
	// CompleteShreds records every subject erasure whose KEK versions are all
	// destroyed.
	CompleteShreds Kind = "complete-shreds"
)

Kinds.

type Overdue

type Overdue struct {
	Purpose keyforge.Purpose
	Version uint32
	DueAt   time.Time
	Late    time.Duration
}

Overdue describes a KEK version past its rotation date that has not been rotated, so it is reported and not silently kept.

type Scheduler

type Scheduler struct {
	// contains filtered or unexported fields
}

Scheduler notices what lifecycle work is due and starts a job for it.

It is stateless in the sense that matters: everything it decides comes from the registry, and a job's identity is derived from what it does, so a scheduler that restarts, or several that run at once, start the same runs and not extra ones. The few things it remembers (when to look again) only space out retries.

func NewScheduler

func NewScheduler(o SchedulerOptions) (*Scheduler, error)

NewScheduler returns a scheduler.

func (*Scheduler) Last

func (s *Scheduler) Last() TickResult

Last returns the result of the most recent tick, or the zero value before the first. It is what a status report shows of the scheduler: the work that is overdue, in flight, waiting or failed.

func (*Scheduler) Run

func (s *Scheduler) Run(ctx context.Context) error

Run ticks until ctx is cancelled. A tick that fails is reported through the audit sink and does not stop the loop: an outage of the registry is not a reason for the scheduler to give up.

func (*Scheduler) Tick

func (s *Scheduler) Tick(ctx context.Context) (TickResult, error)

Tick looks at the registry once, starts a job for each piece of work that is due and has no run in progress, and reports. It does not wait for the jobs.

type SchedulerOptions

type SchedulerOptions struct {
	// Provider runs the jobs. Required. Any OmniWorkflow provider will do.
	Provider omniworkflow.WorkflowProvider
	// Rotator supplies due dates and eligibility. Required.
	Rotator *rotate.Rotator
	// Store is the registry the scheduler reads. Required.
	Store registry.Store
	// Purposes are the purposes whose keys are managed. Required.
	Purposes []keyforge.Purpose
	// Clock defaults to the system clock.
	Clock keyforge.Clock
	// Audit receives job failures. The default discards them.
	Audit audit.Sink
	// Metrics receives job counts and how overdue each rotation is. The default
	// discards them.
	Metrics metrics.Metrics
	// Interval is the time between ticks of [Scheduler.Run].
	Interval time.Duration
	// Recheck, Backoff and MaxBackoff are described on the defaults above.
	Recheck    time.Duration
	Backoff    time.Duration
	MaxBackoff time.Duration
}

SchedulerOptions configures a Scheduler.

type Spec

type Spec struct {
	Kind Kind `json:"kind"`
	// Purpose names the purpose for a rotation.
	Purpose string `json:"purpose,omitempty"`
	// KEKID is the KEK's identifier as lowercase hex.
	KEKID string `json:"kekId,omitempty"`
	// Version is the KEK version the step concerns.
	Version uint32 `json:"version,omitempty"`
	// Force makes a rotation run although the version is not yet due, for an
	// operator who asked for it. A scheduled rotation never sets it.
	Force bool `json:"force,omitempty"`
}

Spec is the input of one step. It crosses the provider boundary as JSON.

type Started

type Started struct {
	RunID   string
	Spec    Spec
	Attempt int
}

Started describes a job a tick started.

type TickResult

type TickResult struct {
	// Wanted is how many pieces of work were due.
	Wanted int
	// Started lists the jobs this tick started.
	Started []Started
	// InFlight is how many due pieces of work already had a run in progress.
	InFlight int
	// Waiting is how many were held back by a retry or recheck interval.
	Waiting int
	// Overdue lists rotations that are past due.
	Overdue []Overdue
	// Failed lists runs that failed and have not succeeded since.
	Failed []FailedJob
}

TickResult reports what one tick found and did.