diff --git a/DESIGN.md b/DESIGN.md index efc2265..97c498a 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -242,7 +242,10 @@ Brand (20px three-line SVG mark in accent, wordmark 16px/650). Nav: Dashboard, I `h1` + `.sub` scope line on one baseline, wrapping. Worker, queue and job pages add breadcrumbs (`/` separator, 13px, `subtle`). ### Stat strip -One `panel` (8px, hairline) holding auto-fit tiles (min 150px) split by hairlines (each tile draws left+top edge pulled 1px under the frame). Tile: label heading, display number, sparkline at foot. `.hot` (failed > 0): number and onset line in `bad`. Onset line: "since 15:04" under the failed number. Status tile lists status counts as status dots. Worker page counters use the same strip without sparklines. Above them, a "Queues" line: each queue the worker's `queue_ids` hold, as a queue name link; hidden when stale or without `queue_ids`. +One `panel` (8px, hairline) holding auto-fit tiles (min 150px) split by hairlines (each tile draws left+top edge pulled 1px under the frame). Tile: label heading, display number, sparkline at foot. `.hot` (failed > 0): number and onset line in `bad`. Onset line: "since 15:04" under the failed number. Status tile lists status counts as status dots. Worker page counters use the same strip without sparklines: Completed, Successful, Failed, Success rate (no Submitted: varmq API 1.1.0 dropped it; a queue's is its `totals.submitted`). + +### Worker sources +Under the worker's counters, `h2.section` "Sources" + a fan (`.fan`, `bg` well, hairline, 8px; no JS): left, one node per source in priority order (the list the strategy picks from) (`.srcn`: `panel`, `line-strong` 1px, 8px, as a DAG task box): `qname` (glyph, name, kind badge) + priority chip "p N" (`.prio`, 12px `subtle` pill) only under the `priority` strategy and when set; a link to the queue page. Topic: broadcast glyph + "topic"; storage with no Queue ID: italic `subtle` "custom storage"; neither links. Edges (1.5px `subtle` 70%, as DAG edges) join one bus, then a 7px arrow into the worker node (`.wnode`: `panel`, 1.5px `accent`, status dot + name 600, strategy in 12px `subtle`: "Priority: lowest number first", "Round robin: each source in turn", "Max length: longest queue first", "Min length: shortest non-empty first"; "Pulls directly" with one source). Under 640px the bus runs down the left, the worker below. Pods merge by kind + Queue ID (first pod's order, then what others add; first pod's strategy). Older varmq (no `sources`): nodes from `queue_ids`, no label, a `subtle` note. Hidden when stale. - **Gap:** tile sparklines are fixed 160x32 (row trends 120x24); they do not stretch with the tile. ### Queue name @@ -251,7 +254,7 @@ One `panel` (8px, hairline) holding auto-fit tiles (min 150px) split by hairline - **DAG node:** the name, cut at 12 chars, after the state. ### Tables -Inside `.table-wrap` panel. Head `raised`, 12px/500 `subtle`; cells 8px 12px, hairline rows, no zebra. Row hover `raised` 60%; selected row accent 8%. Names `fg`/500, accent + underline on hover. Under 640px worker and queue rows stack into a labelled grid (head moved off-screen, `data-label` captions, status on its own line). +Inside `.table-wrap` panel. Head `raised`, 12px/500 `subtle`; cells 8px 12px, hairline rows, no zebra. Row hover `raised` 60%; selected row accent 8%. Names `fg`/500, accent + underline on hover. Queue list "Workers" column: a chip per worker whose `queue_ids` hold the queue (link to its page, title "runs on "); none known: a 12px `subtle` "instances" then the instance chips. Under 640px worker and queue rows stack into a labelled grid (head moved off-screen, `data-label` captions, status on its own line). - **Gap:** no column sorting; needs a server-side sort param (no new JS). ### Status dot @@ -293,7 +296,7 @@ Default: `raised`, `line-strong` border, 13px/500, 5px 10px, hover border `subtl - **Waiting:** 16px `fg` line with a 10px teal dot; page meta-refreshes every 2s. ### Job page -- **Header:** `h1.qtitle`: queue name (20px/600, 18px glyph, kind badge); under it the full Queue ID in mono 12px `subtle` with a "Copy ID" button (`queue.js`, `data-copy`; hidden where `navigator.clipboard` is missing); "Namespace" chip (redisq) and "Reported by" chips. Breadcrumb ends in the queue name. +- **Header:** `h1.qtitle`: queue name (20px/600, 18px glyph, kind badge); under it the full Queue ID in mono 12px `subtle` with a "Copy ID" button (`queue.js`, `data-copy`; hidden where `navigator.clipboard` is missing); "Namespace" chip (redisq); "Read from instance-1, …" in 12px `subtle` (the workers table names the workers). Breadcrumb ends in the queue name. - **Workers on this queue:** `h2.section` + table (instance, worker link to its page, status dot, concurrency) of the workers whose `queue_ids` hold the queue; stacks under 640px like queue rows. None with an older varmq (no `queue_ids`). - **Queue controls** (`#qstate`, under the header): paused status dot ("paused since"), default buttons "Pause queue" / "Resume queue" (Resume alone when every instance says paused, Pause alone when none does, both when mixed or unknown); hidden, with a `subtle` line, when no instance reports `paused` (varmq < API 1.2.0). Confirm panel lists instance x sent, then results; `#qstate` refreshed out of band. - **State tabs:** one sticky strip, five equal columns split by hairlines (scrolls sideways under 640px); label (capitalized, state dot: processing warn, completed ok, failed bad, cancelled hollow `bad` ring, pending `subtle`) over 20px count, `—` in `subtle` when varmq does not count the state (cancelled); current `raised` + 2px teal underline. diff --git a/PRODUCT.md b/PRODUCT.md index 438ffba..facc6c2 100644 --- a/PRODUCT.md +++ b/PRODUCT.md @@ -34,7 +34,7 @@ Live dashboard and control panel for varmq worker fleets. Talks to varmq's REST ## Capabilities and Constraints -- Pages: Dashboard (fleet totals + per-worker rows with 10-min trend), Instance grid (every worker on every instance), Worker page (its queues, totals, success rate, completed/failed charts with 1h | 24h | 7d | 30d), Queues list (by name: store icon, FIFO/Priority, namespace; paused badge), Queue job page (full ID with copy, workers on the queue, jobs by state incl. cancelled, paged, payload, per-attempt errors; queue pause/resume), One-job page (varmq API 1.4.0: state, timeline with waited / ran / total timers, live while it waits or runs; per-attempt errors, payload, result; cancel; linked from queue lists and workflow tasks), Workflows list (varmq API 1.3.0: merged across instances, one row per workflow, state tabs, progress, duration), Workflow page (task DAG laid out server-side as SVG: pan/zoom, upstream/downstream trace, task panel, task table; live while it runs; cancel). Confirm dialogs and action results. +- Pages: Dashboard (fleet totals + per-worker rows with 10-min trend), Instance grid (every worker on every instance), Worker page (its sources fanning into it in priority order, with store glyphs, priorities and the pick strategy (varmq API 1.5.0; `queue_ids` before), totals, success rate, completed/failed charts with 1h | 24h | 7d | 30d), Queues list (by name: store icon, FIFO/Priority, namespace; paused badge; the workers consuming each), Queue job page (full ID with copy, workers on the queue, jobs by state incl. cancelled, paged, payload, per-attempt errors; queue pause/resume), One-job page (varmq API 1.4.0: state, timeline with waited / ran / total timers, live while it waits or runs; per-attempt errors, payload, result; cancel; linked from queue lists and workflow tasks), Workflows list (varmq API 1.3.0: merged across instances, one row per workflow, state tabs, progress, duration), Workflow page (task DAG laid out server-side as SVG: pan/zoom, upstream/downstream trace, task panel, task table; live while it runs; cancel). Confirm dialogs and action results. - Actions: pause, resume, restart, stop, concurrency; bulk pause/resume; queue pause/resume (varmq API 1.2.0+, storage-wide: every process reading the queue); workflow cancel, with or without undo jobs (API 1.3.0, stored with the workflow: every process working on it); job cancel (API 1.4.0, cooperative: sent to the first instance that has the job). No job is retried, removed or purged. - **Stack (binding):** one Go binary. Server-rendered `html/template` + vendored htmx + plain CSS/JS, embedded with `go:embed`. No Node toolchain, no build step, no front-end framework. - Safety behavior (product function, not styling): loopback only; bearer token never reaches the browser; actions refused and nothing sent while data is stale or from another configuration; stale data called out with its age and reason. diff --git a/README.md b/README.md index 4282d59..7656dae 100644 --- a/README.md +++ b/README.md @@ -44,7 +44,7 @@ There is no agent to deploy and no database to run: vartop is one binary on your - **Fleet dashboard.** Completed per second, failed per minute, pending jobs and workers by status, with a 10-minute trend per worker. - **Instance grid.** Every worker on every instance at a glance: status, busy/concurrency, pending. -- **Worker page.** Totals and success rate, and two charts (completed /s, failed /s), each with its own **1h | 24h** switch, plus **7d** and **30d** as far as the server's `history` reaches, axes and a hover readout. +- **Worker page.** The queues and topics it pulls from, fanning into it in the order it picks them, with priorities and its strategy (varmq API 1.5.0). Totals and success rate, and two charts (completed /s, failed /s), each with its own **1h | 24h** switch, plus **7d** and **30d** as far as the server's `history` reaches, axes and a hover readout. - **Workflows** (varmq API 1.3.0). Every workflow, newest first, once however many instances report it, with its state, tasks ended and duration; its task graph laid out left to right: pan and zoom, hover a task to trace what it waits for and what waits on it, click it for its queue, job, reason, result size and undo job. A running workflow's page follows it live. - **One job** (varmq API 1.4.0). Its state, a timeline with how long it waited and ran (ticking while it does), each failed attempt's error, its payload and result. Linked from a queue's job list and a workflow's tasks. - **History that survives restarts.** 24 hours of polls, and up to 30 days of each worker's rates per server, in a local SQLite store. @@ -128,7 +128,7 @@ The page above. One row per worker, merged across every instance (or one instanc - A worker page: submitted, completed, successful, failed and success rate; a 7-day completed/s chart with a daily rhythm; a 24-hour failed/s chart; the concurrency slider and the pause, resume, restart and stop buttons + A worker page: completed, successful, failed and success rate; a 7-day completed/s chart with a daily rhythm; a 24-hour failed/s chart; the concurrency slider and the pause, resume, restart and stop buttons Click a worker for its totals and success rate, its two charts, and its controls. Each chart has its own span: **1h** from the stored polls, **24h** from 5-minute buckets, and **7d**, **30d** (or the server's own [`history`](#schema), like `14d`) merged into longer steps. Hovering a chart shows how many instances answered and, on 1h, the poll's time and value; on longer spans, the exact window, its average and its peak: diff --git a/internal/api/client.go b/internal/api/client.go index dd8a51c..f3e8ae0 100644 --- a/internal/api/client.go +++ b/internal/api/client.go @@ -47,17 +47,35 @@ type WorkerSummary struct { Instance string `json:"-"` // varmq instance that served it (InstanceHeader), if reported } -// WorkerDetails matches the extended API response of GET /workers/{name} +// WorkerDetails matches the extended API response of GET /workers/{name}. +// Metrics has no submitted: varmq API 1.1.0 dropped it (a queue's is its +// totals.submitted under GET /queues). type WorkerDetails struct { WorkerSummary + // Strategy: how the worker picks the next of its Sources (varmq API + // 1.5.0): priority, round_robin, max_len, min_len, or one varmq adds + // later; "" from an older varmq. + Strategy string `json:"strategy,omitempty"` + // Sources: the queues and topic subscriptions it pulls from, in pick + // order (varmq API 1.5.0); nil from an older varmq. + Sources []WorkerSource `json:"sources,omitempty"` Metrics struct { - Submitted uint64 `json:"submitted"` Completed uint64 `json:"completed"` Successful uint64 `json:"successful"` Failed uint64 `json:"failed"` } `json:"metrics"` } +// WorkerSource is one source a worker pulls from (spec WorkerSource): +// QueueID "" for a topic subscription or storage with no Queue ID; Kind +// queue or topic (or one varmq adds later); Priority nil when none was set +// (the lowest). +type WorkerSource struct { + QueueID string `json:"queue_id"` + Kind string `json:"kind"` + Priority *int64 `json:"priority,omitempty"` +} + // APIError represents an error response from the varmq server. type APIError struct { StatusCode int diff --git a/internal/api/client_test.go b/internal/api/client_test.go index 5ba82b3..d4f5f68 100644 --- a/internal/api/client_test.go +++ b/internal/api/client_test.go @@ -3,6 +3,7 @@ package api import ( "context" "errors" + "math" "net" "net/http" "net/http/httptest" @@ -64,7 +65,7 @@ func recordingServer(t *testing.T, status int, header map[string]string, body st func TestGetWorker(t *testing.T) { const body = `{"name":"emails","status":"running","concurrency":4,"num_processing":2,"num_pending":7,"num_idle_workers":2, - "queue_ids":["redisq:fifo:{emails}"],"metrics":{"submitted":10,"completed":8,"successful":6,"failed":2}}` + "queue_ids":["redisq:fifo:{emails}"],"metrics":{"completed":8,"successful":6,"failed":2}}` tests := []struct { name string @@ -95,7 +96,7 @@ func TestGetWorker(t *testing.T) { if !slices.Equal(w.QueueIDs, []string{"redisq:fifo:{emails}"}) { t.Errorf("QueueIDs = %q", w.QueueIDs) } - if w.Metrics.Submitted != 10 || w.Metrics.Completed != 8 || w.Metrics.Successful != 6 || w.Metrics.Failed != 2 { + if w.Metrics.Completed != 8 || w.Metrics.Successful != 6 || w.Metrics.Failed != 2 { t.Errorf("metrics decoded wrong: %+v", w.Metrics) } if w.Origin != srv.URL { @@ -602,3 +603,23 @@ func TestFreshMethodsEscapeNothing(t *testing.T) { t.Fatalf("paths = %v, want %v", paths, want) } } + +// strategy and sources (varmq API 1.5.0) decode in priority order, a priority +// absent as nil; an older varmq sends neither: "" and nil. +func TestGetWorkerSources(t *testing.T) { + srv, _ := recordingServer(t, http.StatusOK, nil, `{"name":"w","status":"Running","concurrency":1,"num_processing":0,"num_pending":0,"num_idle_workers":1, + "strategy":"priority","sources":[{"queue_id":"redisq:fifo:{a}","kind":"queue","priority":-2},{"queue_id":"","kind":"topic"},{"queue_id":"","kind":"queue","priority":9223372036854775807}], + "metrics":{"completed":0,"successful":0,"failed":0}}`) + w, err := NewClient(srv.URL, "").GetWorkerCtx(context.Background(), "w") + if err != nil { + t.Fatal(err) + } + if w.Strategy != "priority" || len(w.Sources) != 3 || w.Sources[0].QueueID != "redisq:fifo:{a}" || *w.Sources[0].Priority != -2 || + w.Sources[1].Kind != "topic" || w.Sources[1].Priority != nil || *w.Sources[2].Priority != math.MaxInt64 { + t.Errorf("worker = %+v", w) + } + srv, _ = recordingServer(t, http.StatusOK, nil, `{"name":"w","status":"Running","concurrency":1,"num_processing":0,"num_pending":0,"num_idle_workers":1,"metrics":{"submitted":3,"completed":0,"successful":0,"failed":0}}`) + if w, err := NewClient(srv.URL, "").GetWorkerCtx(context.Background(), "w"); err != nil || w.Strategy != "" || w.Sources != nil { + t.Errorf("older varmq: %+v, %v", w, err) + } +} diff --git a/internal/fleet/fleettest/fakevarmq.go b/internal/fleet/fleettest/fakevarmq.go index 4d9c463..a43325f 100644 --- a/internal/fleet/fleettest/fakevarmq.go +++ b/internal/fleet/fleettest/fakevarmq.go @@ -17,9 +17,13 @@ import ( // Worker is one worker as a Varmq serves it. type Worker struct { - Status string - Concurrency, Busy, Pending, Idle int - Submitted, Completed, Successful, Failed uint64 + Status string + Concurrency, Busy, Pending, Idle int + Completed, Successful, Failed uint64 + // Strategy and Sources, served by GET /workers/{name} when Sources is + // non-nil (varmq API 1.5.0); nil: an older varmq. + Strategy string + Sources []api.WorkerSource // QueueIDs, served as queue_ids when non-nil; nil: an older varmq. QueueIDs []string // ActionStatus, when non-zero, makes every PATCH on this worker reply @@ -166,8 +170,10 @@ func (f *Varmq) get(w http.ResponseWriter, r *http.Request) { if fw != nil { failWith = fw.GetStatus body = fakeSummary(name, fw) + if fw.Sources != nil { + body["strategy"], body["sources"] = fw.Strategy, fw.Sources + } body["metrics"] = map[string]uint64{ - "submitted": fw.Submitted, "completed": fw.Completed, "successful": fw.Successful, "failed": fw.Failed, @@ -575,3 +581,24 @@ func NewInstances(t *testing.T, ids ...string) ([]*Varmq, []*httptest.Server) { } return fakes, srvs } + +// SourceWorkers are workers on queues redis, sqlite and memory (Queue IDs) +// as varmq API 1.5.0 serves them, for tests and screenshots: "orders" +// picks by priority from all three (1, 5, none set), "emails" pulls from +// redis alone, "fanout" takes turns between sqlite, a topic and storage +// with no Queue ID, and "legacy" is an older varmq's: queue_ids only. +func SourceWorkers(redis, sqlite, memory string) map[string]*Worker { + p := func(n int64) *int64 { return &n } + q := func(id string, prio *int64) api.WorkerSource { + return api.WorkerSource{QueueID: id, Kind: "queue", Priority: prio} + } + return map[string]*Worker{ + "orders": {Status: "Running", Concurrency: 8, Busy: 2, Idle: 6, QueueIDs: []string{memory, redis, sqlite}, + Strategy: "priority", Sources: []api.WorkerSource{q(redis, p(1)), q(sqlite, p(5)), q(memory, nil)}}, + "emails": {Status: "Running", Concurrency: 2, Idle: 2, QueueIDs: []string{redis}, + Strategy: "priority", Sources: []api.WorkerSource{q(redis, nil)}}, + "fanout": {Status: "Idle", Concurrency: 1, Idle: 1, QueueIDs: []string{sqlite}, + Strategy: "round_robin", Sources: []api.WorkerSource{q(sqlite, p(2)), {Kind: "topic"}, {Kind: "queue"}}}, + "legacy": {Status: "Running", Concurrency: 1, Idle: 1, QueueIDs: []string{redis, sqlite}}, + } +} diff --git a/internal/fleet/poller_test.go b/internal/fleet/poller_test.go index 27b973f..a3211c3 100644 --- a/internal/fleet/poller_test.go +++ b/internal/fleet/poller_test.go @@ -70,10 +70,10 @@ func countCalls(calls []string, call string) int { func TestPollOnceBuildsSnapshot(t *testing.T) { _, a := fleettest.NewVarmq(t, "host", map[string]*fleettest.Worker{ "orders": {Status: "Running", Concurrency: 8, Busy: 6, Pending: 80, Idle: 2, - Submitted: 100, Completed: 90, Successful: 85, Failed: 5}, + Completed: 90, Successful: 85, Failed: 5}, }) _, b := fleettest.NewVarmq(t, "host", map[string]*fleettest.Worker{ - "orders": {Status: "Paused", Concurrency: 4, Pending: 7, Idle: 4, Submitted: 3, Completed: 2, Successful: 2}, + "orders": {Status: "Paused", Concurrency: 4, Pending: 7, Idle: 4, Completed: 2, Successful: 2}, "billing": {Status: "Idle", Concurrency: 1, Idle: 1}, }) wantEPs := []string{a.URL, b.URL} @@ -121,7 +121,7 @@ func TestPollOnceBuildsSnapshot(t *testing.T) { Name: "orders", Status: "Running", Concurrency: 8, NumProcessing: 6, NumPending: 80, NumIdleWorkers: 2, Origin: a.URL, Instance: "host", }} - want.Metrics.Submitted, want.Metrics.Completed, want.Metrics.Successful, want.Metrics.Failed = 100, 90, 85, 5 + want.Metrics.Completed, want.Metrics.Successful, want.Metrics.Failed = 90, 85, 5 if !reflect.DeepEqual(sm.WorkerDetails, want) { t.Errorf("sample = %+v\nwant %+v", sm.WorkerDetails, want) } diff --git a/internal/fleet/view.go b/internal/fleet/view.go index ffb1538..905c2f8 100644 --- a/internal/fleet/view.go +++ b/internal/fleet/view.go @@ -8,6 +8,8 @@ import ( "strconv" "strings" "time" + + "github.com/goptics/vartop/internal/api" ) // This file turns snapshots into the view models the templates render. All @@ -111,13 +113,40 @@ type GridView struct { // DetailView is screen C. type DetailView struct { - Name, Status, Class, StatusTip string - Submitted, Completed, Successful, Failed uint64 - SuccessRate string // "99.5%" or "—" - ConcurrencyStart, ConcurrencyMax int // start = max current per-pod; max = max(64, 2*start) - Completed1h, Failed1h []Point // per second - Pods []PodColumn // pods reporting this worker - QueueIDs []string // its queue_ids on those pods, sorted, each once + Name, Status, Class, StatusTip string + Completed, Successful, Failed uint64 + SuccessRate string // "99.5%" or "—" + ConcurrencyStart, ConcurrencyMax int // start = max current per-pod; max = max(64, 2*start) + Completed1h, Failed1h []Point // per second + Pods []PodColumn // pods reporting this worker + QueueIDs []string // its queue_ids on those pods, sorted, each once + // Strategy and Sources: how it picks and what it pulls from (varmq + // API 1.5.0), merged over its pods (see mergeSources). Sources nil: + // no pod sent them (older varmq); QueueIDs is all there is. + Strategy string + Sources []api.WorkerSource +} + +// mergeSources merges pods' sources: the first pod's in its priority order, +// then each source another pod has and it lacks (same kind and Queue ID), +// in that pod's order. Strategy is the first pod's that sends one. Pods +// that differ are not told apart: one worker name, one view. +func mergeSources(d *DetailView, sm *Sample) { + if sm.Sources == nil { + return + } + if d.Strategy == "" { + d.Strategy = sm.Strategy + } + if d.Sources == nil { + d.Sources = slices.Clone(sm.Sources) + return + } + for _, s := range sm.Sources { + if !slices.ContainsFunc(d.Sources, func(o api.WorkerSource) bool { return o.Kind == s.Kind && o.QueueID == s.QueueID }) { + d.Sources = append(d.Sources, s) + } + } } // StatusClass maps a varmq worker status to its colour class. Anything varmq @@ -303,12 +332,19 @@ func BuildDetail(snaps []*Snapshot, pod, worker string, _ time.Duration, n Noun) if p.Err != "" || sm == nil { continue } - d.Submitted += sm.Metrics.Submitted d.Completed += sm.Metrics.Completed d.Successful += sm.Metrics.Successful d.Failed += sm.Metrics.Failed d.Pods = append(d.Pods, PodColumn{Key: p.Key, Label: p.Label, Err: p.Err, Tip: p.Tip()}) d.QueueIDs = append(d.QueueIDs, sm.QueueIDs...) + mergeSources(&d, sm) + } + // A pod of an older varmq says only queue_ids: its queues join as + // sources, so a mixed fleet draws them all. + for _, id := range d.QueueIDs { + if d.Sources != nil && !slices.ContainsFunc(d.Sources, func(s api.WorkerSource) bool { return s.QueueID == id }) { + d.Sources = append(d.Sources, api.WorkerSource{QueueID: id, Kind: "queue"}) + } } slices.Sort(d.QueueIDs) d.QueueIDs = slices.Compact(d.QueueIDs) diff --git a/internal/fleet/view_test.go b/internal/fleet/view_test.go index 03efdf8..d464db8 100644 --- a/internal/fleet/view_test.go +++ b/internal/fleet/view_test.go @@ -8,6 +8,8 @@ import ( "strings" "testing" "time" + + "github.com/goptics/vartop/internal/api" ) var viewT0 = time.Date(2026, 10, 25, 12, 0, 0, 0, time.UTC) @@ -715,15 +717,15 @@ func TestBuildGridAlignsCells(t *testing.T) { } func TestBuildDetail(t *testing.T) { - counters := func(sm *Sample, submitted, completed, ok, failed uint64) *Sample { - sm.Metrics.Submitted, sm.Metrics.Completed, sm.Metrics.Successful, sm.Metrics.Failed = submitted, completed, ok, failed + counters := func(sm *Sample, completed, ok, failed uint64) *Sample { + sm.Metrics.Completed, sm.Metrics.Successful, sm.Metrics.Failed = completed, ok, failed return sm } iv := 5 * time.Second s := snap(atTick(0), - okPod("p0", map[string]*Sample{"orders": counters(smp("Running", 8, 0, 8, 0), 10, 8, 6, 2)}), - okPod("p1", map[string]*Sample{"orders": counters(smp("Idle", 8, 0, 8, 0), 1, 1, 1, 0)}), + okPod("p0", map[string]*Sample{"orders": counters(smp("Running", 8, 0, 8, 0), 8, 6, 2)}), + okPod("p1", map[string]*Sample{"orders": counters(smp("Idle", 8, 0, 8, 0), 1, 1, 0)}), okPod("p2", map[string]*Sample{"billing": smp("Running", 2, 0, 2, 0)}), ) d, ok := BuildDetail([]*Snapshot{s}, "", "orders", iv, podNoun) @@ -736,8 +738,8 @@ func TestBuildDetail(t *testing.T) { if want := "p0: Running\np1: Idle"; d.StatusTip != want { t.Errorf("StatusTip = %q, want %q", d.StatusTip, want) } - if d.Submitted != 11 || d.Completed != 9 || d.Successful != 7 || d.Failed != 2 { - t.Errorf("counters = %d/%d/%d/%d, want 11/9/7/2 (summed over reporting pods)", d.Submitted, d.Completed, d.Successful, d.Failed) + if d.Completed != 9 || d.Successful != 7 || d.Failed != 2 { + t.Errorf("counters = %d/%d/%d, want 9/7/2 (summed over reporting pods)", d.Completed, d.Successful, d.Failed) } if d.SuccessRate != "77.8%" { t.Errorf("SuccessRate = %q, want 77.8%% (7/9)", d.SuccessRate) @@ -756,7 +758,7 @@ func TestBuildDetail(t *testing.T) { } // 6 ok / 2 failed on one pod. - one := snap(atTick(0), okPod("p0", map[string]*Sample{"orders": counters(smp("Running", 8, 0, 8, 0), 8, 8, 6, 2)})) + one := snap(atTick(0), okPod("p0", map[string]*Sample{"orders": counters(smp("Running", 8, 0, 8, 0), 8, 6, 2)})) if d, _ = BuildDetail([]*Snapshot{one}, "", "orders", iv, podNoun); d.SuccessRate != "75.0%" { t.Errorf("SuccessRate = %q, want 75.0%%", d.SuccessRate) } @@ -780,7 +782,7 @@ func TestBuildDetail(t *testing.T) { // Pod filter narrows counters, status and the pod list. d, ok = BuildDetail([]*Snapshot{s}, "p1", "orders", iv, podNoun) - if !ok || d.Status != "Idle" || d.Submitted != 1 || len(d.Pods) != 1 || d.Pods[0].Key != "p1" { + if !ok || d.Status != "Idle" || len(d.Pods) != 1 || d.Pods[0].Key != "p1" { t.Errorf("p1 detail = %+v ok=%v", d, ok) } @@ -1058,3 +1060,58 @@ func TestDetailSeriesBoundedToTheHour(t *testing.T) { } } } + +// A worker's sources merge over its pods: the first pod's in priority order, +// then what others add (same kind and Queue ID once); its strategy is the +// first pod's. A pod of an older varmq adds none; with no pod sending +// them, Sources is nil. +func TestBuildDetailMergesSources(t *testing.T) { + one := int64(1) + src := func(sm *Sample, strategy string, s ...api.WorkerSource) *Sample { + sm.Strategy, sm.Sources = strategy, s + return sm + } + a, b, topic := api.WorkerSource{QueueID: "a", Kind: "queue", Priority: &one}, api.WorkerSource{QueueID: "b", Kind: "queue"}, api.WorkerSource{Kind: "topic"} + s := snap(atTick(0), + okPod("p0", map[string]*Sample{"w": src(smp("Running", 1, 0, 1, 0), "priority", a, topic)}), + okPod("p1", map[string]*Sample{"w": src(smp("Running", 1, 0, 1, 0), "round_robin", b, a)}), + okPod("p2", map[string]*Sample{"w": smp("Running", 1, 0, 1, 0)}), + ) + d, ok := BuildDetail([]*Snapshot{s}, "", "w", 0, instanceNoun) + if !ok || d.Strategy != "priority" || !reflect.DeepEqual(d.Sources, []api.WorkerSource{a, topic, b}) { + t.Errorf("merged = %q %+v", d.Strategy, d.Sources) + } + if d, _ := BuildDetail([]*Snapshot{s}, "p1", "w", 0, instanceNoun); d.Strategy != "round_robin" || !reflect.DeepEqual(d.Sources, []api.WorkerSource{b, a}) { + t.Errorf("one pod = %q %+v", d.Strategy, d.Sources) + } + if d, _ := BuildDetail([]*Snapshot{s}, "p2", "w", 0, instanceNoun); d.Strategy != "" || d.Sources != nil { + t.Errorf("older varmq = %q %+v", d.Strategy, d.Sources) + } + + // mixed fleet: an older pod's queue_ids join as sources + legacy := smp("Running", 1, 0, 1, 0) + legacy.QueueIDs = []string{"c"} + newer := src(smp("Running", 1, 0, 1, 0), "priority", a) + newer.QueueIDs = []string{"a"} + mixed := snap(atTick(0), okPod("p0", map[string]*Sample{"w": legacy}), okPod("p1", map[string]*Sample{"w": newer})) + if d, _ := BuildDetail([]*Snapshot{mixed}, "", "w", 0, instanceNoun); !reflect.DeepEqual(d.Sources, []api.WorkerSource{a, {QueueID: "c", Kind: "queue"}}) { + t.Errorf("mixed fleet = %+v, want a then c", d.Sources) + } + + // kind is part of the key: a topic and an ID-less queue both stay, in order + q := api.WorkerSource{Kind: "queue"} + kinds := snap(atTick(0), okPod("p0", map[string]*Sample{"w": src(smp("Running", 1, 0, 1, 0), "round_robin", topic)}), + okPod("p1", map[string]*Sample{"w": src(smp("Running", 1, 0, 1, 0), "round_robin", q)})) + if d, _ := BuildDetail([]*Snapshot{kinds}, "", "w", 0, instanceNoun); !reflect.DeepEqual(d.Sources, []api.WorkerSource{topic, q}) { + t.Errorf("topic + ID-less queue = %+v", d.Sources) + } + + // the merge appends to a copy, never into the first pod's array + first := make([]api.WorkerSource, 1, 4) + first[0] = a + shared := snap(atTick(0), okPod("p0", map[string]*Sample{"w": src(smp("Running", 1, 0, 1, 0), "priority", first...)}), + okPod("p1", map[string]*Sample{"w": src(smp("Running", 1, 0, 1, 0), "priority", b)})) + if d, _ := BuildDetail([]*Snapshot{shared}, "", "w", 0, instanceNoun); len(d.Sources) != 2 || first[:2][1] != (api.WorkerSource{}) { + t.Errorf("merge wrote into the pod's sources: %+v, pod %+v", d.Sources, first[:2]) + } +} diff --git a/internal/store/bench_test.go b/internal/store/bench_test.go index 8a5ebb0..27f0a9b 100644 --- a/internal/store/bench_test.go +++ b/internal/store/bench_test.go @@ -71,7 +71,6 @@ func fleetTicks(n, units, workers int, interval time.Duration) []*fleet.Snapshot sm.Name, sm.Status, sm.Concurrency = name, "running", 16 sm.NumProcessing, sm.NumPending, sm.NumIdleWorkers = rng.IntN(17), rng.IntN(5000), rng.IntN(17) sm.Instance, sm.Origin = inst, pod.Via - sm.Metrics.Submitted = c.completed + uint64(sm.NumPending) + uint64(sm.NumProcessing) sm.Metrics.Completed, sm.Metrics.Failed = c.completed, c.failed sm.Metrics.Successful = c.completed - c.failed pod.Workers[name] = sm diff --git a/internal/store/codec.go b/internal/store/codec.go index 9afcdd0..361e76e 100644 --- a/internal/store/codec.go +++ b/internal/store/codec.go @@ -9,6 +9,7 @@ import ( "sync" "time" + "github.com/goptics/vartop/internal/api" "github.com/goptics/vartop/internal/fleet" ) @@ -51,12 +52,15 @@ type sampleRec struct { Idle int `json:"id,omitempty"` Instance *string `json:"i,omitempty"` // nil: the unit's Instance Origin *string `json:"o,omitempty"` // nil: the unit's Via - Submitted uint64 `json:"su,omitempty"` Completed uint64 `json:"co,omitempty"` Successful uint64 `json:"ok,omitempty"` Failed uint64 `json:"fa,omitempty"` Rates []float64 `json:"r,omitempty"` // completed/s, failed/s to 3 dp; absent unless RateOK QueueIDs []string `json:"q,omitempty"` // added without a new StoreVersion: older rows read nil + // Strategy and Sources, as QueueIDs: no new StoreVersion; older rows + // read none. Older rows' "su" (submitted) is read and dropped. + Strategy string `json:"st,omitempty"` + Sources []api.WorkerSource `json:"src,omitempty"` } // labelsRec is a fleet.LabelState. @@ -129,9 +133,8 @@ func encodeSnapshot(s *fleet.Snapshot) ([]byte, error) { Status: sm.Status, Concurrency: sm.Concurrency, Processing: sm.NumProcessing, Pending: sm.NumPending, Idle: sm.NumIdleWorkers, Instance: ifDiffers(sm.Instance, p.Instance), Origin: ifDiffers(sm.Origin, p.Via), - Submitted: sm.Metrics.Submitted, Completed: sm.Metrics.Completed, - Successful: sm.Metrics.Successful, Failed: sm.Metrics.Failed, - QueueIDs: sm.QueueIDs, + Completed: sm.Metrics.Completed, Successful: sm.Metrics.Successful, Failed: sm.Metrics.Failed, + QueueIDs: sm.QueueIDs, Strategy: sm.Strategy, Sources: sm.Sources, } if c, f := round3(sm.CompletedRate), round3(sm.FailedRate); sm.RateOK && finite(c) && finite(f) { w.Rates = []float64{c, f} @@ -192,9 +195,8 @@ func decodeSnapshot(atMs int64, data []byte) (*fleet.Snapshot, error) { sm.Name, sm.Status, sm.Concurrency = orDefault(w.Name, name), w.Status, w.Concurrency sm.NumProcessing, sm.NumPending, sm.NumIdleWorkers = w.Processing, w.Pending, w.Idle sm.Instance, sm.Origin = orDefault(w.Instance, u.Instance), orDefault(w.Origin, u.Via) - sm.Metrics.Submitted, sm.Metrics.Completed = w.Submitted, w.Completed - sm.Metrics.Successful, sm.Metrics.Failed = w.Successful, w.Failed - sm.QueueIDs = w.QueueIDs + sm.Metrics.Completed, sm.Metrics.Successful, sm.Metrics.Failed = w.Completed, w.Successful, w.Failed + sm.QueueIDs, sm.Strategy, sm.Sources = w.QueueIDs, w.Strategy, w.Sources if len(w.Rates) == 2 { sm.CompletedRate, sm.FailedRate, sm.RateOK = w.Rates[0], w.Rates[1], true } diff --git a/internal/store/server_test.go b/internal/store/server_test.go index 70da4d2..97f52ef 100644 --- a/internal/store/server_test.go +++ b/internal/store/server_test.go @@ -2,6 +2,7 @@ package store import ( "bytes" + "compress/flate" "database/sql" "errors" "fmt" @@ -16,6 +17,7 @@ import ( "testing" "time" + "github.com/goptics/vartop/internal/api" "github.com/goptics/vartop/internal/fleet" ) @@ -94,7 +96,7 @@ func sample(snapAt time.Time, offMs int64, name, instance, origin string, comple s.Name, s.Status, s.Concurrency = name, "running", 8 s.NumProcessing, s.NumPending, s.NumIdleWorkers = 3, 120, 5 s.Instance, s.Origin = instance, origin - s.Metrics.Submitted, s.Metrics.Completed, s.Metrics.Successful, s.Metrics.Failed = completed+120, completed, completed-failed, failed + s.Metrics.Completed, s.Metrics.Successful, s.Metrics.Failed = completed, completed-failed, failed return s } @@ -119,7 +121,7 @@ func fullSnapshot(round bool) *fleet.Snapshot { { Key: direct, Via: direct, Instance: "inst-a", Label: "instance-1", Workers: map[string]*fleet.Sample{ - "emails": withQueues(sample(snapAt, 120, "emails", "inst-a", direct, 1000, 4, r(12.34567, 12.346), r(0.0004, 0), true), "redisq:prio:{emails}", "memory:inst-a:fifo:1"), + "emails": withSources(withQueues(sample(snapAt, 120, "emails", "inst-a", direct, 1000, 4, r(12.34567, 12.346), r(0.0004, 0), true), "redisq:prio:{emails}", "memory:inst-a:fifo:1")), "reports": sample(snapAt, 95, "reports", "inst-a", direct, 50, 0, 0, 0, false), "moved": sample(snapAt, 140, "moved", "inst-z", "http://10.0.0.9:8080/varmq", 7, 1, 0, 0, true), "anon": sample(snapAt, 77, "anon", "", direct, 9, 0, r(1.0/3, 0.333), 0, true), @@ -151,6 +153,15 @@ func fullSnapshot(round bool) *fleet.Snapshot { } } +// withSources sets s's strategy and sources (varmq API 1.5.0): a queue +// with a priority, an in-memory one and a topic without. +func withSources(s *fleet.Sample) *fleet.Sample { + one := int64(1) + s.Strategy = "priority" + s.Sources = []api.WorkerSource{{QueueID: "redisq:prio:{emails}", Kind: "queue", Priority: &one}, {QueueID: "memory:inst-a:fifo:1", Kind: "queue"}, {Kind: "topic"}} + return s +} + // withQueues sets s's QueueIDs. func withQueues(s *fleet.Sample, ids ...string) *fleet.Sample { s.QueueIDs = ids @@ -817,3 +828,25 @@ func TestUndecodableRecords(t *testing.T) { t.Errorf("logged %d times, want once per file:\n%s", n, log) } } + +// A row from before strategy and sources (and with the dropped submitted) +// reads its queue_ids, no strategy and nil sources. +func TestDecodeOlderWorkerRow(t *testing.T) { + var buf bytes.Buffer + zw, err := flate.NewWriter(&buf, flate.BestSpeed) + if err != nil { + t.Fatal(err) + } + _, _ = zw.Write([]byte(`{"u":[{"k":"http://a","w":{"w":{"t":0,"s":"Running","su":12,"co":9,"q":["redisq:fifo:{a}"]}}}]}`)) + if err := zw.Close(); err != nil { + t.Fatal(err) + } + s, err := decodeSnapshot(base.UnixMilli(), buf.Bytes()) + if err != nil { + t.Fatal(err) + } + sm := s.Pods[0].Workers["w"] + if !slices.Equal(sm.QueueIDs, []string{"redisq:fifo:{a}"}) || sm.Strategy != "" || sm.Sources != nil || sm.Metrics.Completed != 9 { + t.Errorf("older row = %+v", sm.WorkerDetails) + } +} diff --git a/internal/web/actions.go b/internal/web/actions.go index 9b1794b..e745585 100644 --- a/internal/web/actions.go +++ b/internal/web/actions.go @@ -396,6 +396,7 @@ func livePage(r *http.Request, key string, n fleet.Noun, interval time.Duration, p := Page{Server: key, Pod: label, Noun: n, IntervalSec: int(interval / time.Second), Stale: stale, Spans: spans, Ranges: ranges} fill(&p, snaps) + p.owners = ownersOf(latestOf(snaps)) // names in-memory sources, as page does if podGone(&p, snaps) { return "", Page{}, false // the next poll shows why } diff --git a/internal/web/actions_test.go b/internal/web/actions_test.go index b9596e5..a85d855 100644 --- a/internal/web/actions_test.go +++ b/internal/web/actions_test.go @@ -1657,3 +1657,16 @@ func TestJobCancel(t *testing.T) { t.Errorf("%d PATCHes after the refusals", n-sent) } } + +// An action's #live refresh names an in-memory source by its worker, as +// the page does: livePage reads the owners too. +func TestActionRefreshNamesMemorySources(t *testing.T) { + _, fake := fleettest.NewVarmq(t, "a", fleettest.SourceWorkers("redisq:fifo:{r}", "sqliteq:h:fifo:s", "memory:a:fifo:1")) + srv, _ := newTestServer(t, fake) + req := postRequest(t, srv, "/workers/orders/actions/pause", multi()) + req.Header.Set("HX-Current-URL", srv.URL+"/workers/orders?server=multi") + resp, body := send(t, srv, req) + if _, live, ok := strings.Cut(body, `
`); resp.StatusCode != http.StatusOK || !ok || !strings.Contains(live, `orders #1`) { + t.Errorf("refresh = %d, the in-memory source not named by its worker:\n%s", resp.StatusCode, body) + } +} diff --git a/internal/web/queues.go b/internal/web/queues.go index 6a3b22d..fcc2287 100644 --- a/internal/web/queues.go +++ b/internal/web/queues.go @@ -4,6 +4,7 @@ import ( "bytes" "encoding/json" "errors" + "maps" "math" "net/http" "net/url" @@ -85,6 +86,36 @@ func queueWorkers(s *fleet.Snapshot, id, server string) []QueueWorker { return out } +// WorkerChip is a worker consuming a queue, as the queue list names it: +// its page, and Tip, the units it runs on. +type WorkerChip struct{ Name, URL, Tip string } + +// workerChips are, by Queue ID, the workers of s whose queue_ids hold it, +// by name, each once; nil from an older varmq (no queue_ids). +func workerChips(s *fleet.Snapshot, server string) map[string][]WorkerChip { + on := map[string]map[string][]string{} // Queue ID → worker → unit labels + for _, p := range fleet.PodsOf(s, "") { + if p.Err != "" { + continue + } + for name, sm := range p.Workers { + for _, id := range sm.QueueIDs { + if on[id] == nil { + on[id] = map[string][]string{} + } + on[id][name] = append(on[id][name], p.Label) + } + } + } + out := make(map[string][]WorkerChip, len(on)) + for id, workers := range on { + for _, name := range slices.Sorted(maps.Keys(workers)) { + out[id] = append(out[id], WorkerChip{Name: name, URL: workerURL(name, server, "", ""), Tip: "runs on " + strings.Join(workers[name], ", ")}) + } + } + return out +} + // QueueCtl is the paused badge and the pause and resume buttons of a queue's // page (#qstate), and, with OOB, the same as an out-of-band swap for an // action response. Since, SinceFull and SinceISO say when it was paused @@ -264,6 +295,7 @@ func (h *handler) queues(w http.ResponseWriter, r *http.Request) { // by name, as the rows show them; v.Group needs ID order, unused here slices.SortStableFunc(v.Groups, func(a, b fleet.QueueGroup) int { return p.cmpQName(a.ID, b.ID) }) p.Queues = &v + p.qworkers = workerChips(latestOf(snaps), p.Server) } } h.render(w, "queues", p.status(r), p) diff --git a/internal/web/queues_test.go b/internal/web/queues_test.go index ca019b2..8425a2c 100644 --- a/internal/web/queues_test.go +++ b/internal/web/queues_test.go @@ -3,6 +3,7 @@ package web import ( "encoding/json" "fmt" + "math" "net/http" "net/http/httptest" "net/url" @@ -77,11 +78,18 @@ func TestQueuesPage(t *testing.T) { if n := strings.Count(body, tenantName+""); n != 1 { t.Errorf("queue listed %d times, want once", n) } + labels := fleettest.StaticLabels(fakes[0].URL, fakes[1].URL, fakes[2].URL) + la, lab := labels[fakes[0].URL], slices.Sorted(slices.Values([]string{labels[fakes[0].URL], labels[fakes[1].URL]})) for _, want := range []string{ `Redis: `, `Namespace`, `—`, `In memory: emails #7 FIFO`, // its worker's name `shop`, + // the workers consuming it, by queue_ids; none: the instances + `Workers`, + `orders`, + `emails`, + `instances ` + la + ``, } { if !strings.Contains(body, want) { t.Errorf("GET /queues lacks %q", want) @@ -96,7 +104,8 @@ func TestQueuesPage(t *testing.T) { } last = i } - if _, body := fetch(t, srv, "/queues/redisq:shop:prio:%7Bemails%7D?server=multi", false); !strings.Contains(body, `Namespace shop · Reported by`) { + if _, body := fetch(t, srv, "/queues/redisq:shop:prio:%7Bemails%7D?server=multi", false); !strings.Contains(body, `

Namespace shop

+

Read from instance-`) { t.Errorf("namespaced queue page lacks its namespace:\n%s", body) } text := visibleText(body) @@ -183,11 +192,11 @@ func TestQueueJobsPage(t *testing.T) { } // the worker page links back, by name _, body = fetch(t, srv, "/workers/orders?server=multi", false) - if !strings.Contains(body, `

Queues `) { + if !strings.Contains(body, `

  • `) || !strings.Contains(body, "From queue_ids") { t.Errorf("worker page lacks its queue:\n%s", body) } urls := []string{fakes[0].URL, fakes[1].URL, fakes[2].URL} - if status, body := fetch(t, srv, "/workers/orders?server=multi&instance="+fleettest.StaticLabels(urls...)[fakes[2].URL], false); status != http.StatusOK || strings.Contains(body, "wqueues") { + if status, body := fetch(t, srv, "/workers/orders?server=multi&instance="+fleettest.StaticLabels(urls...)[fakes[2].URL], false); status != http.StatusOK || strings.Contains(body, `class="sources"`) { t.Errorf("an older varmq's worker = %d, lists queues:\n%s", status, body) } @@ -823,6 +832,12 @@ func TestQueueNamesHostile(t *testing.T) { ids := []string{"redisq:" + evil + ":fifo:{" + evil + "}", "sqliteq:h:prio:" + evil, "memory:a:fifo:1", evil} pod := okPod("http://a", map[string]*fleet.Sample{evil: smp("Running", 1, 0, 1, 0)}) pod.Workers[evil].QueueIDs = ids + one := int64(1) + pod.Workers[evil].Strategy = evil + for _, id := range ids { + pod.Workers[evil].Sources = append(pod.Workers[evil].Sources, api.WorkerSource{QueueID: id, Kind: "queue", Priority: &one}) + } + pod.Workers[evil].Sources = append(pod.Workers[evil].Sources, api.WorkerSource{Kind: evil}) s := snap(viewT0, pod) var groups []fleet.QueueGroup for _, id := range ids { @@ -837,7 +852,7 @@ func TestQueueNamesHostile(t *testing.T) { t.Fatal("hostile job: no cancel button to check") } for name, body := range map[string]string{ - "list": run(t, "queues", "layout", Page{Server: "dev", owners: ownersOf(s), Queues: &fleet.QueuesView{Groups: groups}}), + "list": run(t, "queues", "layout", Page{Server: "dev", owners: ownersOf(s), qworkers: workerChips(s, "dev"), Queues: &fleet.QueuesView{Groups: groups}}), "page": run(t, "queue", "layout", Page{Server: "dev", owners: ownersOf(s), Jobs: jobs}), "worker": run(t, "detail", "layout", Page{Server: "dev", owners: ownersOf(s), Detail: &d}), "confirm": run(t, "confirm", "queue_confirm", QueueConfirmView{Action: "pause", Server: "dev", Queue: parseQueueID(ids[0]), Noun: instanceNoun}), @@ -848,6 +863,13 @@ func TestQueueNamesHostile(t *testing.T) { if strings.Contains(body, "alert") || !strings.Contains(body, "<script>") { t.Errorf("%s: a field rendered as markup", name) } + // the worker's sources and its chips on the queue list are drawn, escaped + if name == "worker" && (strings.Count(body, `class="srcn"`) != len(ids)+1 || !strings.Contains(body, `<script>`)) { + t.Errorf("worker: sources not drawn:\n%s", body) + } + if name == "list" && (strings.Count(body, `class="chip wchip"`) != len(ids) || !strings.Contains(body, `]*>([^<]+)`).FindAllStringSubmatch(body, -1) { + order = append(order, m[1]+" "+m[2]) + } + if want := []string{esc(redis) + " Redis: ", esc(sqlite) + " SQLite: ", esc(memory) + " In memory: ", esc(extra) + " Redis: "}; !slices.Equal(order, want) { + t.Errorf("source nodes = %q, want %q", order, want) + } + for _, want := range []string{ + `

    Sources 4, by priority; the strategy picks among them

    `, + `orders FIFO    p 1    `, + `p 5`, + `orders #1`, // in-memory: named by its worker + `
    ordersPriority: lowest number first
    `, + } { + if !strings.Contains(body, want) { + t.Errorf("orders page lacks %q", want) + } + } + if strings.Count(body, `class="prio"`) != 2 || strings.Contains(body, "Submitted") || strings.Contains(body, "srcnote") { + t.Errorf("orders: a priority chip on the unset one, a Submitted card or the queue_ids note:\n%s", body) + } + + _, body = fetch(t, srv, "/workers/emails?server=multi", false) + if !strings.Contains(body, "Pulls directly") || strings.Contains(body, `class="prio"`) || !strings.Contains(body, "what the worker pulls from") { + t.Errorf("one source:\n%s", body) + } + _, body = fetch(t, srv, "/workers/fanout?server=multi", false) + for _, want := range []string{ + "Round robin: each source in turn", + `
  • topic
    • `, + `
  • custom storage
    • `, + } { + if !strings.Contains(body, want) { + t.Errorf("fanout page lacks %q", want) + } + } + if strings.Contains(body, `class="prio"`) { + t.Error("fanout: a priority chip under round robin") + } + _, body = fetch(t, srv, "/workers/legacy?server=multi", false) + if !strings.Contains(body, "From queue_ids") || !strings.Contains(body, `legacy
    `) || strings.Count(body, `
  • Pods reporting: —

    `) { diff --git a/internal/web/sources.go b/internal/web/sources.go new file mode 100644 index 0000000..d193cf3 --- /dev/null +++ b/internal/web/sources.go @@ -0,0 +1,77 @@ +package web + +import ( + "cmp" + "math" + "strconv" + + "github.com/goptics/vartop/internal/api" +) + +// SourcesView is a worker's sources as its page draws them: Nodes, in pick +// order, fanning into the worker. Label says how it picks: the strategy +// in words with two or more sources, "Pulls directly" with one, "" when +// the server does not say (Legacy: varmq before API 1.5.0, the nodes read +// from queue_ids: no strategy, no priorities). +type SourcesView struct { + Nodes []SourceNode + Label string + Legacy bool +} + +// SourceNode is one source: a queue (Queue, URL to its page), a topic +// subscription, or storage with no Queue ID (Kind queue, no URL). Kind is +// as varmq sent it. Priority is "p N" when the worker picks by priority +// and N was set. +type SourceNode struct { + Kind string + Queue QueueName + URL string + Priority string +} + +// strategyWords says each strategy varmq sends in words. +var strategyWords = map[string]string{ + "priority": "Priority: lowest number first", + "round_robin": "Round robin: each source in turn", + "max_len": "Max length: longest queue first", + "min_len": "Min length: shortest non-empty first", +} + +// WorkerSources is the sources of the page's worker, nil with none to show +// (none reported, or a stale page: no current value). +func (p Page) WorkerSources() *SourcesView { + d := p.Detail + if d == nil || p.Stale != "" { + return nil + } + v := &SourcesView{} + srcs := d.Sources + if srcs == nil { + v.Legacy = true + for _, id := range d.QueueIDs { + srcs = append(srcs, api.WorkerSource{QueueID: id, Kind: "queue"}) + } + } + if len(srcs) == 0 { + return nil + } + for _, s := range srcs { + n := SourceNode{Kind: s.Kind} + if s.QueueID != "" { + n.Queue, n.URL = p.QName(s.QueueID), jobsURL(s.QueueID, p.Server, "", 0, defaultJobLimit) + } + if d.Strategy == "priority" && s.Priority != nil && *s.Priority != math.MaxInt64 { + n.Priority = "p " + strconv.FormatInt(*s.Priority, 10) + } + v.Nodes = append(v.Nodes, n) + } + switch { + case v.Legacy: // the server does not say how it picks + case len(v.Nodes) == 1: + v.Label = "Pulls directly" + default: + v.Label = cmp.Or(strategyWords[d.Strategy], d.Strategy) + } + return v +} diff --git a/internal/web/static/app.css b/internal/web/static/app.css index 891db68..4207b33 100644 --- a/internal/web/static/app.css +++ b/internal/web/static/app.css @@ -345,8 +345,43 @@ small.err { display: block; margin: 0; font-size: var(--fs-1); font-weight: 400; .qfull { display: flex; flex-wrap: wrap; align-items: center; gap: 4px 8px; margin: 6px 0 0; color: var(--subtle); font-size: var(--fs-1); } .qfull code { min-width: 0; overflow-wrap: anywhere; } .qfull .copy { padding: 1px 8px; font-size: var(--fs-1); } -.wqueues { display: flex; flex-wrap: wrap; align-items: center; gap: 6px 14px; margin: 0 0 16px; } -.wqueues .sub { font-size: var(--fs-2); } +/* worker page: its sources fan into it (no JS): a node per source in pick + order, edges into one bus, an arrow to the worker; stacked under 640px */ +.fan { + --edge: color-mix(in srgb, var(--subtle) 70%, transparent); + display: grid; grid-template-columns: minmax(0, max-content) 44px minmax(0, max-content); align-items: center; justify-content: start; + margin: 0 0 8px; padding: 14px 16px; background: var(--bg); border: 1px solid var(--line); border-radius: var(--r-3); +} +.srcs { margin: 0; padding: 0; list-style: none; min-width: 0; } +.srcs li { position: relative; display: flex; align-items: center; padding: 5px 0; } +.srcs li::after { content: ""; flex: 1 0 24px; border-top: 1.5px solid var(--edge); } +.srcs li::before { content: ""; position: absolute; right: 0; top: 0; bottom: 0; border-right: 1.5px solid var(--edge); } +.srcs li:first-child::before { top: 50%; } +.srcs li:last-child::before { bottom: 50%; } +.srcs li:only-child::before { display: none; } +.srcn { display: inline-flex; flex-wrap: wrap; align-items: center; gap: 4px 8px; min-width: 0; max-width: 100%; padding: 8px 12px; background: var(--panel); border: 1px solid var(--line-strong); border-radius: var(--r-3); color: var(--fg); text-decoration: none; } +a.srcn:hover, a.srcn:focus-visible { border-color: var(--accent); } +a.srcn:hover .qnm { color: var(--accent); text-decoration: underline; } +.srcn .qnm.none { color: var(--subtle); font-style: italic; } +.prio { padding: 0 6px; border: 1px solid var(--line); border-radius: 999px; background: var(--raised); color: var(--subtle); font-size: var(--fs-1); font-variant-numeric: tabular-nums; } +.fan .link { position: relative; border-top: 1.5px solid var(--edge); } +.fan .link::after { content: ""; position: absolute; right: -1px; top: -6px; border: 5px solid transparent; border-right: 0; border-left: 7px solid var(--edge); } +.wnode { display: flex; flex-direction: column; gap: 2px; min-width: 0; padding: 10px 14px; background: var(--panel); border: 1.5px solid var(--accent); border-radius: var(--r-3); } +.wnode .nm { display: flex; align-items: center; gap: 8px; font-weight: 600; overflow-wrap: anywhere; } +.wnode small { color: var(--subtle); font-size: var(--fs-1); } +.srcnote { margin: 0 0 16px; } +.sources { margin: 0 0 16px; } +@media (max-width: 640px) { + .fan { grid-template-columns: minmax(0, 1fr); justify-content: stretch; } + .srcs li { padding: 5px 0 5px 24px; } + .srcs li::after { position: absolute; left: 8px; top: 50%; width: 16px; } + .srcs li::before { left: 8px; right: auto; border-right: 0; border-left: 1.5px solid var(--edge); } + .srcs li:last-child::before { bottom: 0; } + .srcs li:only-child::before { display: block; top: 50%; bottom: 0; } + .fan .link { width: 0; height: 18px; margin-left: 8px; border-top: 0; border-left: 1.5px solid var(--edge); } + .fan .link::after { right: auto; top: auto; left: -6px; bottom: -1px; border: 5px solid transparent; border-bottom: 0; border-top: 7px solid var(--edge); } + .wnode { justify-self: start; max-width: 100%; } +} a.ql { color: var(--fg); text-decoration: none; } a.ql:hover .qnm { color: var(--accent); text-decoration: underline; } @@ -374,6 +409,10 @@ a.ql:hover .qnm { color: var(--accent); text-decoration: underline; } .queues .name .status { margin-left: 8px; font-family: var(--sans); font-size: var(--fs-1); font-weight: 500; white-space: nowrap; } .chip { display: inline-block; padding: 1px 8px; border: 1px solid var(--line); border-radius: 999px; background: var(--raised); color: var(--fg); font-size: var(--fs-1); } .queues .on .chip { margin: 2px 4px 2px 0; } +.queues .on .sub { font-size: var(--fs-1); } +a.wchip { text-decoration: none; } +a.wchip:hover { border-color: var(--accent); color: var(--accent); } +.readfrom { margin: 4px 0 0; color: var(--subtle); font-size: var(--fs-1); } .states { position: sticky; top: 0; z-index: 2; display: grid; grid-template-columns: repeat(5, minmax(0, 1fr)); margin: 0 0 12px; overflow: hidden; background: var(--panel); border: 1px solid var(--line); border-radius: var(--r-3); diff --git a/internal/web/templates/detail.html b/internal/web/templates/detail.html index ce01983..d01e7b5 100644 --- a/internal/web/templates/detail.html +++ b/internal/web/templates/detail.html @@ -1,21 +1,33 @@ +{{/* topic_icon: a topic subscription's 24-unit filled glyph: a broadcast */}} +{{define "topic_icon"}}{{end}} +{{/* prio: a source's priority chip (SourceNode), when it has one */}} +{{define "prio"}}{{with .Priority}}{{.}}{{end}}{{end}} {{define "live_body"}}{{with .Detail}}

    {{.Name}}

    {{if not $.Stale}}{{.Status}}{{if eq .Class "mixed"}}{{.StatusTip}}{{end}}{{end}}
    -{{if and .QueueIDs (not $.Stale)}}

    Queues{{range .QueueIDs}} {{template "qname" ($.QName .)}}{{end}}

    {{end}}
    {{if $.Stale}} -

    Submitted

    —

    Completed

    —

    Successful

    —

    Failed

    —

    Success rate

    —

    {{else}} -

    Submitted

    {{.Submitted}}

    Completed

    {{.Completed}}

    Successful

    {{.Successful}}

    Failed

    {{.Failed}}

    Success rate

    {{.SuccessRate}}

    {{end}}
    +{{with $.WorkerSources}}{{$n := len .Nodes}}
    +

    Sources {{if gt $n 1}}{{$n}}, by priority; the strategy picks among them{{else}}what the worker pulls from{{end}}

    +
    +
      {{range .Nodes}} +
    1. {{if .URL}}{{template "qname" .Queue}}{{template "prio" .}}{{else}}{{if eq .Kind "topic"}}{{template "topic_icon"}}topic{{else if eq .Kind "queue"}}custom storage{{else}}{{.Kind}}{{end}}{{template "prio" .}}{{end}}
      2. {{end}} +
    + +
    {{$.Detail.Name}}{{with .Label}}{{.}}{{end}}
    +
    +{{if .Legacy}}

    From queue_ids: this varmq (before API 1.5.0) does not say how the worker picks, or with what priorities.

    {{end}} +
    {{end}}
    {{$c := history .Completed1h historyWindow $.Interval 720 120}}{{$peak := $c.Peak}}{{with $.CompletedDay}}{{$c = history .Points .Span .Step 720 120}}{{$peak = .Peak}}{{end}}
    Completed /s peak {{tickLabel $peak}}/s{{template "range_switch" (rangeSwitch $ "completed")}}
    {{template "history" $c}} diff --git a/internal/web/templates/queue.html b/internal/web/templates/queue.html index 9649ad0..aca163d 100644 --- a/internal/web/templates/queue.html +++ b/internal/web/templates/queue.html @@ -10,7 +10,8 @@ {{with .Jobs}}{{$state := .State}}{{$q := $.QName .Queue.ID}}

    {{template "qname" $q}}

    {{.Queue.ID}}

    -

    {{with $q.NS}}Namespace {{.}} · {{end}}Reported by {{range .Queue.On}}{{.}} {{end}}

    + {{with $q.NS}}

    Namespace {{.}}

    {{end}} +

    Read from {{range $i, $l := .Queue.On}}{{if $i}}, {{end}}{{$l}}{{end}}

    {{template "qstate" .Ctl}}
    diff --git a/internal/web/templates/queues.html b/internal/web/templates/queues.html index 8da7f0b..a3f7e6c 100644 --- a/internal/web/templates/queues.html +++ b/internal/web/templates/queues.html @@ -10,7 +10,7 @@ Completed Failed Submitted (total) - Reported by + Workers {{range .Groups}}{{$q := $.QName .ID}} @@ -22,7 +22,7 @@ {{.Failed}}{{else}} {{.Err}}{{end}} {{with .Totals}}{{.Submitted}}{{else}}—{{end}} - {{range $i, $l := .On}}{{if $i}}, {{end}}{{$l}}{{end}} + {{range $i, $w := $.WorkersOn .ID}}{{if $i}}, {{end}}{{$w.Name}}{{else}}instances {{range $i, $l := .On}}{{if $i}}, {{end}}{{$l}}{{end}}{{end}} {{end}}