Documentation
¶
Overview ¶
Package admin is the operator surface of jobrund: the Broker that fans job transitions out to subscribers, the /admin/events SSE feed, /admin/stats, the hand-written /metrics text exposer, the depth cache the two read from, and the embedded /admin page.
The package imports internal/job and web only; the worker, scheduler and reaper reach it through the small interfaces they define themselves (worker.EventPublisher is satisfied by *Broker, DurationObserver by *Histogram).
Index ¶
- Constants
- Variables
- func Events(b *Broker, snapshot func(context.Context) ([]job.DepthRow, error)) http.Handler
- func Handler(d Deps) http.Handler
- func Metrics(cache *DepthCache, counters func() Counters, hist *Histogram, ...) http.Handler
- func NewServer(addr string, d Deps) *http.Server
- func Stats(d StatsDeps) http.Handler
- type Broker
- type Counters
- type Deps
- type DepthCache
- type DurationObserver
- type Histogram
- type JobKey
- type Series
- type SeriesKey
- type StatsDeps
Constants ¶
const ( OutcomeSucceeded = "succeeded" OutcomeDead = "dead" OutcomeCancelled = "cancelled" OutcomeRetried = "retried" )
Outcome names counted by the Broker.
const ContentType = "text/plain; version=0.0.4; charset=utf-8"
ContentType is the Prometheus text exposition media type.
const DefaultDepthInterval = 5 * time.Second
DefaultDepthInterval is how often the depth cache refreshes.
const ( // KeepaliveInterval is how long the feed may be silent before a // `: keepalive` comment is sent so proxies and EventSource keep the // connection alive. KeepaliveInterval = 15 * time.Second )
Variables ¶
var DurationBuckets = durationBuckets[:]
DurationBuckets are the explicit upper bounds (seconds) of jobrun_job_duration_seconds; +Inf is implicit.
Functions ¶
func Events ¶
Events streams job transitions from b as Server-Sent Events: one `event: snapshot` (the depth table from snapshot) on connect, then `event: job` per transition, with a keepalive comment while idle. It sends no CORS headers itself; Handler wraps it in the shared origin allowlist.
The handler holds the connection open for as long as the client stays, so it MUST be mounted outside any http.TimeoutHandler (whose ResponseWriter cannot flush and whose deadline would cut the stream); it clears the server's per-request write deadline for its own connection. Handler and NewServer mount it correctly.
func Handler ¶
Handler mounts the admin surface on a fresh mux:
GET /admin the embedded synthwave page GET /admin/events SSE feed (snapshot, then one event per transition) GET /admin/stats JSON summary GET /metrics Prometheus text GET /healthz liveness, "ok"
Nothing here is wrapped in http.TimeoutHandler: /admin/events is a long-lived stream. If you add middleware around this handler, keep timeouts off the events route (or give the http.Server no WriteTimeout, as NewServer does).
func Metrics ¶
func Metrics(cache *DepthCache, counters func() Counters, hist *Histogram, build map[string]string) http.Handler
Metrics serves the queue metrics in Prometheus text format 0.0.4, written by hand on a pooled buffer (no client library). cache may be nil (no depth series), counters may be nil (no counter series), hist may be nil (no histogram). build becomes the labels of jobrun_build_info; "go_version" defaults to runtime.Version() and "version" to "dev".
Series: jobrun_queue_depth{queue,status}, jobrun_queue_oldest_age_seconds{queue} (oldest queued run_at), jobrun_jobs_total{queue,status}, jobrun_dead_total, jobrun_lease_expired_total, jobrun_scheduler_ticks_total, jobrun_job_duration_seconds{queue,type} (histogram), jobrun_depth_age_seconds, jobrun_build_info.
func NewServer ¶
NewServer builds the admin http.Server on addr (docs/PLAN.md: 127.0.0.1:9110) around Handler(d). WriteTimeout is deliberately zero so the SSE stream is not cut; ReadHeaderTimeout and IdleTimeout bound the rest. Stop it with Server.Shutdown in teardown stage 1; SSE connections end with their request contexts.
Types ¶
type Broker ¶
type Broker struct {
// contains filtered or unexported fields
}
Broker fans job.Event values out to subscribers. Publish never blocks: a subscriber whose buffer is full misses the event and the drop is counted. It also tallies outcomes per queue for /metrics.
func (*Broker) Counters ¶
Counters returns a copy of the outcome tally (Jobs, Dead). LeaseExpired and SchedulerTicks are zero here; the integrator fills them from obs.Snapshot.
func (*Broker) Dropped ¶
Dropped returns the total events dropped across all subscribers since start.
func (*Broker) Publish ¶
Publish delivers e to every subscriber that has room and counts the outcome. It never blocks and is safe from any goroutine.
func (*Broker) Subscribe ¶
Subscribe registers a subscriber with a buffer of buf events (min 1) and returns its channel and a cancel func. cancel is idempotent; after it returns the channel is closed and Publish no longer sees the subscriber.
func (*Broker) Subscribers ¶
Subscribers returns the number of live subscriptions.
type Counters ¶
type Counters struct {
Jobs map[JobKey]uint64
Dead uint64
LeaseExpired uint64
SchedulerTicks uint64
// LeaseFenced counts handlers whose lease a heartbeat found gone (in-process
// pool plus leases WorkerService reported lost to remote workers).
LeaseFenced uint64
}
Counters is what /metrics renders as counters. Jobs and Dead are fed by the Broker from the events it sees; LeaseExpired and SchedulerTicks come from the runtime (obs.Snapshot) and are filled in by the func the integrator wires.
type Deps ¶
type Deps struct {
Broker *Broker
// Depth feeds /metrics, /admin/stats and the SSE snapshot. Run it yourself
// (DepthCache.Run) in the process errgroup.
Depth *DepthCache
// Counters is read per scrape; wire it as
// func() Counters { c := broker.Counters(); s := snap(); c.LeaseExpired = s.LeaseLost; c.SchedulerTicks = s.SchedulerTicks; return c }.
Counters func() Counters
// Durations is the histogram the worker feeds through DurationObserver.
Durations *Histogram
// Build labels jobrun_build_info (version, go_version, ...).
Build map[string]string
Version string
StartedAt time.Time
// Pool returns worker.Pool.Stats() for /admin/stats.
Pool func() any
// AllowedOrigins (ADMIN_CORS_ORIGINS) are the cross-origin pages allowed
// to read /admin/stats and /admin/events. Empty (the default) sends no
// Access-Control-Allow-Origin at all: only the same-origin /admin page
// can read them.
AllowedOrigins []string
Log *slog.Logger
}
Deps is everything the admin surface reads. Only Broker is required; a nil field simply leaves its data out.
type DepthCache ¶
type DepthCache struct {
// contains filtered or unexported fields
}
DepthCache refreshes the (queue, status) depth table on a ticker and serves the last good value to /metrics, /admin/stats, the SSE snapshot and the Stats RPC, so a scrape never runs the full scan itself. It satisfies server.DepthSource.
func NewDepthCache ¶
func NewDepthCache(fetch func(context.Context) ([]job.DepthRow, error), every time.Duration, log *slog.Logger) *DepthCache
NewDepthCache builds a cache over fetch (normally job.Store.Depth). every <= 0 means DefaultDepthInterval. Nothing runs until Run is called; Depth returns an empty table with a zero time before the first refresh.
func (*DepthCache) Depth ¶
func (c *DepthCache) Depth() ([]job.DepthRow, time.Time)
Depth returns the last good table and when it was taken (zero before the first successful refresh). The slice is shared: callers must not modify it.
type DurationObserver ¶
DurationObserver is what the worker calls when a handler finishes, so jobrun_job_duration_seconds can be exposed without the worker importing this package. *Histogram implements it; cmd/jobrund passes it to the pool.
type Histogram ¶
type Histogram struct {
// contains filtered or unexported fields
}
Histogram is a fixed-bucket cumulative histogram per (queue, type). It is written by handler goroutines and read by /metrics; both paths are cheap.
func (*Histogram) ObserveDuration ¶
ObserveDuration records one job run.
type JobKey ¶
type JobKey struct {
Queue string
// Status is the outcome: succeeded, dead, cancelled or retried.
Status string
}
JobKey labels one jobrun_jobs_total series.
type Series ¶
type Series struct {
Key SeriesKey
Cumulative []uint64 // len(DurationBuckets); the +Inf bucket equals Count
Sum float64
Count uint64
}
Series is a read-only snapshot of one (queue, type) histogram with cumulative bucket counts in DurationBuckets order.
type StatsDeps ¶
type StatsDeps struct {
Depth *DepthCache
Broker *Broker
Version string
StartedAt time.Time
// Pool returns the worker pool's Stats() (any JSON-marshalable value);
// nil omits the field. It is `any` so admin does not import internal/worker.
Pool func() any
// Counters is the same func /metrics reads; nil omits the field.
Counters func() Counters
}
StatsDeps is what /admin/stats reports.
Source Files