bvandeusenandClaude Opus 5.5
41d9d00216
Read from what Steward stored for Archives on 2026-10-14: the nightly Restic replication job had said "running" since the first scrape two days earlier, with no good copy yet and past both its late and missing times. Archives ranks "running" above lateness, so the widget called it healthy. - states blocks take warn_past / crit_past: a deadline metric per row (a *_timestamp_seconds). Once it has passed, a healthy row reads as a warning or a problem: "FullNightly-Restic is running, 11h 53m overdue". Quiet and already-failing rows keep their own reading. - The Archives template (version 2) uses the job's late_at / missing_at deadlines and shows its last good copy beside the next run. It says why section_up, the queue and the HTTP/audit counters are left out. - tests/plugins/prometheus/archives_live.py is the series as captured live, and test_archives_template.py renders the template over them. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
327 lines
13 KiB
Python
327 lines
13 KiB
Python
"""A widget template drawn from a target's newest values: what the
|
|
Prometheus dashboard widget shows. Rows in, a view out; nothing is fetched
|
|
here, so the widget route reads the database and a test passes rows.
|
|
|
|
The view follows the widget shape (System 154): a verdict naming the worst
|
|
thing, then each block with its problems first and the healthy rest after.
|
|
Every row carries a tone: crit, warn, ok, or dim for a state the template
|
|
calls quiet.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
|
|
from steward.core.formatting import human_duration
|
|
|
|
from .service import fold_states, label_pairs, shown_value
|
|
from .template_format import UNITS, Block, Limit, Template
|
|
|
|
TONE_ORDER = {"crit": 0, "warn": 1, "ok": 2, "dim": 3}
|
|
|
|
|
|
@dataclass
|
|
class Sample:
|
|
resource: str
|
|
metric: str
|
|
labels: str # as stored: "target=S,state=late"
|
|
pairs: dict[str, str]
|
|
value: float
|
|
at: datetime | None = None
|
|
|
|
|
|
@dataclass
|
|
class Row:
|
|
name: str
|
|
tone: str = "ok"
|
|
why: str = "" # what is wrong, in words: "is late", "below 10%"
|
|
text: str = "" # the value as shown, "—" when there is none
|
|
moment: datetime | None = None
|
|
bar: float | None = None # a share of `of`, 0 to 1
|
|
cells: list[dict] = field(default_factory=list)
|
|
resource: str = ""
|
|
metric: str = ""
|
|
history: list[float] | None = None
|
|
|
|
|
|
@dataclass
|
|
class BlockView:
|
|
kind: str
|
|
title: str
|
|
tone: str = "ok"
|
|
rows: list[Row] = field(default_factory=list)
|
|
columns: list[str] = field(default_factory=list)
|
|
note: str = "" # the block couldn't be drawn: no series, or not a state set
|
|
|
|
|
|
@dataclass
|
|
class View:
|
|
tone: str
|
|
verdict: str
|
|
blocks: list[BlockView]
|
|
wants: list[tuple[str, str]] = field(default_factory=list) # (resource, metric) for sparklines
|
|
note: str = ""
|
|
|
|
|
|
def samples(target: str, latest) -> dict[str, list[Sample]]:
|
|
"""A target's newest values by metric. `latest` is PluginMetricLatest
|
|
rows (resource_name, metric_name, value, recorded_at) for the target and
|
|
its "<target>:<labels>" children."""
|
|
out: dict[str, list[Sample]] = {}
|
|
for row in latest:
|
|
if row.resource_name == target:
|
|
labels = ""
|
|
elif row.resource_name.startswith(target + ":"):
|
|
labels = row.resource_name[len(target) + 1:]
|
|
else:
|
|
continue
|
|
out.setdefault(row.metric_name, []).append(Sample(
|
|
row.resource_name, row.metric_name, labels, dict(label_pairs(labels)), row.value,
|
|
getattr(row, "recorded_at", None)))
|
|
return out
|
|
|
|
|
|
def pick(index: dict[str, list[Sample]], metric: str, where: dict[str, str]) -> list[Sample]:
|
|
return [s for s in index.get(metric, []) if all(s.pairs.get(k) == v for k, v in where.items())]
|
|
|
|
|
|
def shown(metric: str, unit: str, value: float | None) -> dict:
|
|
"""{text, moment} for a value, in `unit` or else as its name says. A
|
|
count reads with thousands separated: 93,310."""
|
|
if unit == "count" and value is not None:
|
|
return {"text": f"{value:,.0f}" if value == int(value) else f"{value:,.2f}", "moment": None}
|
|
return shown_value(f"x{UNITS[unit]}" if unit else metric, value)
|
|
|
|
|
|
def _worst(tones) -> str:
|
|
tones = [t for t in tones if t != "dim"]
|
|
return min(tones, key=TONE_ORDER.__getitem__) if tones else "ok"
|
|
|
|
|
|
def _ordered(rows: list[Row]) -> list[Row]:
|
|
return sorted(rows, key=lambda r: (TONE_ORDER[r.tone], r.name.lower()))
|
|
|
|
|
|
# ── Blocks ───────────────────────────────────────────────────────────────────
|
|
|
|
def _same_row(index, metric: str, labels: str, name_label: str, name: str) -> list[Sample]:
|
|
"""The series of `metric` belonging to a state set's row: the same labels,
|
|
or else the same value of the label that names it."""
|
|
match = [s for s in index.get(metric, []) if s.labels == labels]
|
|
if not match and name:
|
|
match = [s for s in index.get(metric, []) if s.pairs.get(name_label) == name]
|
|
return match
|
|
|
|
|
|
def _overdue(spec: dict, index, labels: str, name: str, now: datetime) -> tuple[str, float] | None:
|
|
"""(tone, seconds past) when the row's crit_past or warn_past deadline has passed."""
|
|
for level in ("crit", "warn"):
|
|
metric = spec[f"{level}_past"]
|
|
if not metric:
|
|
continue
|
|
match = _same_row(index, metric, labels, spec["name"], name)
|
|
if len(match) == 1 and match[0].value > 0:
|
|
past = now.timestamp() - match[0].value
|
|
if past > 0:
|
|
return level, past
|
|
return None
|
|
|
|
|
|
def _states(block: Block, index, now) -> BlockView:
|
|
spec = block.spec
|
|
view = BlockView("states", block.title, columns=[c["label"] for c in spec["columns"]])
|
|
found = pick(index, spec["metric"], spec["where"])
|
|
folded = fold_states(spec["metric"], [{"labels": s.labels, "value": s.value, "resource": s.resource}
|
|
for s in found])
|
|
if not found:
|
|
view.note = f"Nothing reported for {spec['metric']}."
|
|
elif folded is None:
|
|
view.note = (f"{spec['metric']} isn't a state set: it should have one series per state, "
|
|
"the current one 1 and the rest 0.")
|
|
for f in folded or []:
|
|
pairs = dict(label_pairs(f["labels"]))
|
|
name = pairs.get(spec["name"], "") if spec["name"] else ""
|
|
current = [s for s in f["state"].split(", ") if s]
|
|
tones = []
|
|
for state in current:
|
|
if state in spec["healthy"]:
|
|
tones.append("ok")
|
|
elif state in spec["quiet"]:
|
|
tones.append("dim")
|
|
elif state in spec["warning"]:
|
|
tones.append("warn")
|
|
else:
|
|
tones.append("crit")
|
|
tone = (min(tones, key=TONE_ORDER.__getitem__) if tones else "warn")
|
|
why = "" if tone in ("ok", "dim") else (f"is {f['state']}" if current else "reports no state")
|
|
overdue = _overdue(spec, index, f["labels"], name, now) if tone == "ok" else None
|
|
if overdue:
|
|
tone, why = overdue[0], f"is {f['state']}, {human_duration(overdue[1])} overdue"
|
|
row = Row(name=name or f["labels"] or block.title, tone=tone, text=f["state"] or "—", why=why,
|
|
resource=f["resource"], metric=spec["metric"])
|
|
for col in spec["columns"]:
|
|
match = _same_row(index, col["metric"], f["labels"], spec["name"], name)
|
|
row.cells.append(shown(col["metric"], col["unit"], match[0].value if len(match) == 1 else None))
|
|
view.rows.append(row)
|
|
view.rows = _ordered(view.rows)
|
|
return view
|
|
|
|
|
|
def _limit_text(lim: Limit, metric: str, unit: str) -> str:
|
|
return f"{lim.value:g}%" if lim.share else (shown(metric, unit, lim.value)["text"] or f"{lim.value:g}")
|
|
|
|
|
|
def _judge(item: dict, value: float, of: float | None) -> tuple[str, str]:
|
|
for level in ("crit", "warn"):
|
|
for side in ("above", "below"):
|
|
lim = item[f"{level}_{side}"]
|
|
if lim is None:
|
|
continue
|
|
if lim.share:
|
|
if not of:
|
|
continue
|
|
x = value / of * 100
|
|
else:
|
|
x = value
|
|
if (x > lim.value) if side == "above" else (x < lim.value):
|
|
return level, f"{side} {_limit_text(lim, item['metric'], item['unit'])}"
|
|
return "ok", ""
|
|
|
|
|
|
def _rows_for(item: dict, index) -> list[tuple[str, Sample | None, str]]:
|
|
"""The series an item shows: one per value of its `name` label, or the one
|
|
series it names. (name, sample, problem)."""
|
|
found = pick(index, item["metric"], item["where"])
|
|
if item["name"]:
|
|
named = [(s.pairs.get(item["name"], "") or item["label"], s, "") for s in found]
|
|
return named or [(item["label"], None, "not reported")]
|
|
if len(found) == 1:
|
|
return [(item["label"], found[0], "")]
|
|
if not found:
|
|
return [(item["label"], None, "not reported")]
|
|
return [(item["label"], None, f"{len(found)} series match; narrow it with `where` or `name`")]
|
|
|
|
|
|
def _values(block: Block, index, now) -> tuple[BlockView, list]:
|
|
view, wants = BlockView("values", block.title), []
|
|
for item in block.spec["items"]:
|
|
for name, s, problem in _rows_for(item, index):
|
|
if s is None:
|
|
view.rows.append(Row(name=name, tone="warn", why=problem, text="—", metric=item["metric"]))
|
|
continue
|
|
row = Row(name=name, resource=s.resource, metric=s.metric, **shown(s.metric, item["unit"], s.value))
|
|
of = None
|
|
if item["of"]:
|
|
whole = [o for o in index.get(item["of"], []) if o.labels == s.labels]
|
|
if len(whole) == 1 and whole[0].value:
|
|
of = whole[0].value
|
|
row.bar = max(0.0, min(1.0, s.value / of))
|
|
row.text = f"{row.text} of {shown(item['of'], item['unit'], of)['text']}"
|
|
row.tone, row.why = _judge(item, s.value, of)
|
|
if item["sparkline"]:
|
|
wants.append((s.resource, s.metric))
|
|
view.rows.append(row)
|
|
view.rows = _ordered(view.rows)
|
|
return view, wants
|
|
|
|
|
|
def _times(block: Block, index, now) -> BlockView:
|
|
view = BlockView("times", block.title)
|
|
for item in block.spec["items"]:
|
|
for name, s, problem in _rows_for(item, index):
|
|
if s is None:
|
|
view.rows.append(Row(name=name, tone="warn", why=problem, text="—", metric=item["metric"]))
|
|
continue
|
|
got = shown_value("x_timestamp_seconds", s.value)
|
|
row = Row(name=name, resource=s.resource, metric=s.metric, **got)
|
|
if row.moment is None:
|
|
if item["warn_after"] or item["crit_after"]:
|
|
row.tone, row.why = "warn", "never"
|
|
else:
|
|
age = (now - row.moment).total_seconds()
|
|
for level in ("crit", "warn"):
|
|
after = item[f"{level}_after"]
|
|
if after and age > after:
|
|
row.tone, row.why = level, f"over {human_duration(after)} ago"
|
|
break
|
|
view.rows.append(row)
|
|
view.rows = _ordered(view.rows)
|
|
return view
|
|
|
|
|
|
def _table(block: Block, index, now) -> BlockView:
|
|
spec = block.spec
|
|
found = pick(index, spec["metric"], spec["where"])
|
|
rows: dict[str, dict[str, float]] = {}
|
|
columns: list[str] = []
|
|
for s in found:
|
|
r, c = s.pairs.get(spec["rows"]), s.pairs.get(spec["columns"])
|
|
if r is None or c is None:
|
|
continue
|
|
rows.setdefault(r, {})[c] = s.value
|
|
if c not in columns:
|
|
columns.append(c)
|
|
columns.sort()
|
|
view = BlockView("table", block.title, columns=columns)
|
|
if not rows:
|
|
view.note = (f"Nothing reported for {spec['metric']} with labels "
|
|
f"{spec['rows']} and {spec['columns']}.")
|
|
for name in sorted(rows, key=str.lower):
|
|
view.rows.append(Row(name=name, cells=[shown(spec["metric"], spec["unit"], rows[name].get(c))
|
|
for c in columns]))
|
|
return view
|
|
|
|
|
|
def _info(block: Block, index, now) -> BlockView:
|
|
spec = block.spec
|
|
view = BlockView("info", block.title)
|
|
found = sorted(pick(index, spec["metric"], spec["where"]), key=lambda s: s.labels)
|
|
if not found:
|
|
view.note = f"Nothing reported for {spec['metric']}."
|
|
for s in found:
|
|
keys = spec["labels"] or list(s.pairs)
|
|
view.rows.append(Row(name=" · ".join(f"{k} {s.pairs[k]}" for k in keys if s.pairs.get(k)),
|
|
resource=s.resource, metric=s.metric))
|
|
return view
|
|
|
|
|
|
_DRAW = {"states": _states, "times": _times, "table": _table, "info": _info}
|
|
|
|
|
|
def render(template: Template, target: str, latest, now: datetime,
|
|
history: dict[tuple[str, str], list[float]] | None = None) -> View:
|
|
"""The view of `template` over a target's newest values. `history` gives
|
|
the sparkline points the first pass asked for in `wants`."""
|
|
index = samples(target, latest)
|
|
blocks, wants = [], []
|
|
for block in template.blocks:
|
|
if block.kind == "values":
|
|
view, want = _values(block, index, now)
|
|
wants += want
|
|
else:
|
|
view = _DRAW[block.kind](block, index, now)
|
|
if history:
|
|
for row in view.rows:
|
|
row.history = history.get((row.resource, row.metric))
|
|
view.tone = "warn" if view.note else _worst(r.tone for r in view.rows)
|
|
blocks.append(view)
|
|
|
|
note = ""
|
|
if template.expects and not any(m.startswith(template.expects) for m in index):
|
|
note = (f"{target} reports no {template.expects}… metrics: is “{template.name}” "
|
|
"the template for it, or is the target filtering them out?")
|
|
problems = [(b, r) for b in blocks for r in b.rows if r.tone in ("crit", "warn")]
|
|
problems.sort(key=lambda p: TONE_ORDER[p[1].tone])
|
|
empty = [b for b in blocks if b.note]
|
|
tone = _worst([b.tone for b in blocks] + (["warn"] if note else []))
|
|
if note:
|
|
verdict = note
|
|
elif len(problems) + len(empty) == 0:
|
|
verdict = "All well."
|
|
else:
|
|
first = (f"{problems[0][0].title}: {problems[0][1].name} {problems[0][1].why}." if problems
|
|
else f"{empty[0].title}: {empty[0].note}")
|
|
count = len(problems) + len(empty)
|
|
verdict = first if count == 1 else f"{count} need a look. {first}"
|
|
return View(tone=tone, verdict=verdict, blocks=blocks, wants=wants, note=note)
|