README
¶
mailer control plane
The Mailer job control plane (job-orchestration plan, phase P3): submit a workflow, and the controller launches one runner container for it, tracks its lifecycle, and keeps it converging on your desired state — the JobManager equivalent for the single-container-per-job model.
This is a separate Go module (control/) so the core engine stays
dependency-light: go get github.com/ASHUTOSH-SWAIN-GIT/mailer never pulls the
Docker/SQLite clients that live here.
Layout
| Package | Role |
|---|---|
store |
SQLite persistence (jobs, runs, transitions). Source of truth; never stores secrets. |
lifecycle |
Phases, legal transitions, restart policy. |
backend |
ContainerBackend interface + Docker + Kubernetes impls + in-memory fake. |
(root) control |
Controller: submit/cancel/restart + the reconciler loop. |
api |
REST server over the controller. |
cmd/mailer |
The mailer CLI (mailer dashboard). |
Run
Build the runner image once (from the repo root — see
cmd/mailer-runner/README.md), then launch the dashboard:
docker build -f Dockerfile.runner -t mailer-runner:dev . # repo root
cd control && go run ./cmd/mailer dashboard # starts controller + opens the UI
mailer dashboard boots the controller and opens the web UI in your browser.
Add -no-open to run it headless (e.g. on a server), or -addr :9000 to
change the port. Everything — submit, watch, cancel, restart, savepoint —
happens in that one UI.
Backends: Docker or Kubernetes
The controller drives one job per container through a ContainerBackend. The
same jobs, API, and UI work on either backend — only where the containers run
changes.
Docker (default) — one container per job on the local daemon:
mailer dashboard # -backend docker is the default
Kubernetes — one batch/v1 Job per job on a cluster (a Job, not a
Deployment, so a completed job isn't auto-restarted — mailer's reconciler owns
restarts). Each job gets a per-job PVC (state + checkpoints), a ConfigMap (the
workflow), an optional Secret (env), and a ClusterIP Service, with /healthz
liveness/readiness probes and fsGroup so the non-root runner can write the
volume.
# The image must be pullable by the cluster — push it, or for kind:
kind load docker-image mailer-runner:dev
mailer dashboard -backend kubernetes -namespace default -image mailer-runner:dev
Notes for the Kubernetes backend:
- Image:
mailer-runner:devis local; a real cluster needs it in a registry (-image <registry>/mailer-runner:tag), or loaded into kind. - Live state proxy: the dashboard reaches a job's
/stateand/metricsvia the ClusterIP Service DNS, so those work when the controller runs in-cluster. Run the controller on the host (against a remote cluster) and job lifecycle (submit/status/logs/cancel/restart) still works via the API, but the live-metrics proxy needs in-cluster networking (or a port-forward). - Savepoints live under the per-job PVC (
/data/savepoints), so same-job restart-from-savepoint works. Cross-host / cross-job savepoints need an object store (an S3Blobstoreadapter) — a planned follow-up.
API
| Method + path | Purpose |
|---|---|
POST /jobs |
Submit a workflow. Body: raw YAML, or JSON {"workflow": "...", "env": {...}} to pass secrets. Validated (dry-run compile) before launch. |
GET /jobs |
List jobs. |
GET /jobs/{id} |
Job detail: job + latest run + transition log. |
POST /jobs/{id}/cancel |
Graceful stop; desired state → stopped. |
POST /jobs/{id}/restart |
Stop any live run and launch a fresh one. Body {"savepoint":"<label>"} resumes from a savepoint. |
POST /jobs/{id}/savepoint |
Stop-with-savepoint. Label via ?label= or body {"label":"..."}. |
GET /jobs/{id}/logs?tail=N |
Container logs (tail=0 for all). |
GET /jobs/{id}/state |
Proxy to the job's live agent /state. |
GET /jobs/{id}/metrics |
Proxy to the job's live agent /metrics. |
Example
curl -X POST --data-binary @examples/workflows/order-totals.yaml \
-H 'content-type: application/yaml' localhost:9000/jobs
curl localhost:9000/jobs
curl localhost:9000/jobs/<id>/state
curl -X POST localhost:9000/jobs/<id>/cancel
Savepoints (stop-with-savepoint)
A savepoint is a named, durable snapshot you can restart from — for upgrades or redeploys:
curl -X POST 'localhost:9000/jobs/<id>/savepoint?label=before-upgrade'
# ...deploy new code...
curl -X POST localhost:9000/jobs/<id>/restart \
-H 'content-type: application/json' -d '{"savepoint":"before-upgrade"}'
The job drains, writes a final checkpoint, and the runner promotes it to a
blob under savepoints/<label> in a shared volume visible to every job (the
same namespace an S3 bucket gives across hosts — an S3 blobstore drops in for
P6 without touching the savepoint code). The workflow must have checkpointing
enabled (env.checkpointing).
Exactly-once across restarts
Two safeguards keep exactly-once intact when a job restarts:
- Single-live-run fencing — the controller refuses to launch a second container while one is live, so two transactional producers with the same id never coexist.
- Stable transactional id — the job's spec (with its
transactionalID) is stored and reused verbatim on every restart. Pin it to the job by referencing the injectedMAILER_JOB_ID(transactionalID: ${MAILER_JOB_ID}).
Design notes
- Store is the source of truth. On restart the controller's reconciler re-reads active runs and re-attaches to their containers — a controller crash never loses track of a running job.
- Secrets are never persisted. The workflow doc is stored with its
${VAR}placeholders intact; resolved values are passed to the container and held in process memory only. A job that needs secrets cannot be relaunched after a controller restart without re-supplying them (real secret management is a later concern); already-running containers are unaffected. - Reconciler enforces desired state, applies the restart policy to crashed containers (bounded attempts + backoff), and marks clean exits Finished.
- Submit-time validation compiles the workflow in a throwaway data dir, so a Postgres sink is checked by opening its pool — an unreachable database fails the submit.
Documentation
¶
Overview ¶
Package control is the Mailer job control plane: it accepts workflow submissions, launches one container per job through a ContainerBackend, persists everything to a Store, and reconciles the running containers toward each job's desired state. It is the JobManager equivalent for the single-container-per-job model (see the job-orchestration plan, P3).
Index ¶
- type Controller
- func (c *Controller) Cancel(ctx context.Context, jobID string) error
- func (c *Controller) ControlAddress(ctx context.Context, jobID string) (string, error)
- func (c *Controller) GetJob(id string) (*store.Job, error)
- func (c *Controller) LatestRun(jobID string) (*store.Run, error)
- func (c *Controller) ListJobs() ([]*store.Job, error)
- func (c *Controller) Logs(ctx context.Context, jobID string, tail int) (string, error)
- func (c *Controller) Reconcile(ctx context.Context) error
- func (c *Controller) Restart(ctx context.Context, jobID string) (*store.Job, error)
- func (c *Controller) RestartFromSavepoint(ctx context.Context, jobID, label string) (*store.Job, error)
- func (c *Controller) RunReconciler(ctx context.Context, interval time.Duration)
- func (c *Controller) Savepoint(ctx context.Context, jobID, label string) error
- func (c *Controller) Submit(ctx context.Context, doc []byte, env map[string]string) (*store.Job, error)
- func (c *Controller) Transitions(jobID string) ([]*store.Transition, error)
- func (c *Controller) Validate(doc []byte, env map[string]string) (string, compiler.DeliveryGuarantee, compiler.PipelineGraph, error)
- type Options
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Controller ¶
type Controller struct {
// contains filtered or unexported fields
}
Controller ties the store, backend, and lifecycle rules together.
func (*Controller) Cancel ¶
func (c *Controller) Cancel(ctx context.Context, jobID string) error
Cancel sets the job's desired state to stopped and stops its live container gracefully. The container is kept (not removed) so logs stay available; a later Restart or GC removes it.
func (*Controller) ControlAddress ¶
ControlAddress returns the host:port of a job's live control surface, or "" if the job has no reachable running container.
func (*Controller) GetJob ¶
func (c *Controller) GetJob(id string) (*store.Job, error)
GetJob returns one job.
func (*Controller) LatestRun ¶
func (c *Controller) LatestRun(jobID string) (*store.Run, error)
LatestRun returns the most recent run for a job (nil if never launched).
func (*Controller) ListJobs ¶
func (c *Controller) ListJobs() ([]*store.Job, error)
ListJobs returns all jobs, newest first.
func (*Controller) Reconcile ¶
func (c *Controller) Reconcile(ctx context.Context) error
Reconcile drives every active run toward its job's desired state once. It is the single place that observes real container status and updates the store — so after a controller restart, calling Reconcile re-attaches to the containers the store still knows about (the store is the source of truth). Safe to call repeatedly.
func (*Controller) Restart ¶
Restart stops any live run and launches a fresh one, resetting the desired state to running and the attempt counter.
func (*Controller) RestartFromSavepoint ¶
func (c *Controller) RestartFromSavepoint(ctx context.Context, jobID, label string) (*store.Job, error)
RestartFromSavepoint restarts the job, seeding the new run's state from the named savepoint instead of the last automatic checkpoint.
func (*Controller) RunReconciler ¶
func (c *Controller) RunReconciler(ctx context.Context, interval time.Duration)
RunReconciler runs Reconcile on a ticker until ctx is cancelled. Call it in a goroutine from the controller process.
func (*Controller) Savepoint ¶
func (c *Controller) Savepoint(ctx context.Context, jobID, label string) error
Savepoint triggers a stop-with-savepoint on the job's live container: the job drains, writes its final checkpoint, and the runner promotes it to a named savepoint. The job's desired state becomes stopped.
func (*Controller) Submit ¶
func (c *Controller) Submit(ctx context.Context, doc []byte, env map[string]string) (*store.Job, error)
Submit validates a workflow document, records the job (desired: running), and launches its first container. The env carries any secrets; it is used for validation and passed to the container but never persisted. A validation failure rejects the job before any container starts.
func (*Controller) Transitions ¶
func (c *Controller) Transitions(jobID string) ([]*store.Transition, error)
Transitions returns a job's lifecycle audit log.
func (*Controller) Validate ¶
func (c *Controller) Validate(doc []byte, env map[string]string) (string, compiler.DeliveryGuarantee, compiler.PipelineGraph, error)
Validate compiles a workflow without launching it — the dry-run preview behind the submit form. Returns the name, delivery guarantee, and graph a submit would produce, or an error if the workflow is invalid.
type Options ¶
type Options struct {
Store store.Store
Backend backend.ContainerBackend
Image string // runner image tag
ControlPort int // container control port (default 8080)
Restart lifecycle.RestartPolicy // default: lifecycle.DefaultRestartPolicy()
StopTimeout time.Duration // graceful stop wait (default 30s)
NewID func() string // override for deterministic tests
Logf func(string, ...any) // optional logger
}
Options configures a Controller.
Source Files
Directories
¶
| Path | Synopsis |
|---|---|
|
Package api exposes the Controller over HTTP: submit, list, inspect, cancel, restart, and log/metric access for jobs.
|
Package api exposes the Controller over HTTP: submit, list, inspect, cancel, restart, and log/metric access for jobs. |
|
Package backend abstracts where a job's container actually runs.
|
Package backend abstracts where a job's container actually runs. |
|
cmd
|
|
|
mailer
command
Command mailer is the control-plane CLI.
|
Command mailer is the control-plane CLI. |
|
Package lifecycle defines the controller's view of a job's phases and the rules that move a job between them.
|
Package lifecycle defines the controller's view of a job's phases and the rules that move a job between them. |
|
Package store persists the control plane's durable state: the jobs a user submitted and the container runs launched for them.
|
Package store persists the control plane's durable state: the jobs a user submitted and the container runs launched for them. |
|
Package ui embeds and serves the control-plane dashboard — a single self-contained HTML page (inline CSS/JS, no external assets) so the controller binary needs no build step or static file deployment.
|
Package ui embeds and serves the control-plane dashboard — a single self-contained HTML page (inline CSS/JS, no external assets) so the controller binary needs no build step or static file deployment. |