bvandeusenandClaude Opus 5.5
4d95c5d0c2
#5690 "Widget template format: YAML schema, validator and renderer", step 2 of milestone 508 "Prometheus widget templates". - template_format.check(text) -> Template, or TemplateError with every problem, each with its line and what to change. Blocks: states, values (of -> share and bar, thresholds as numbers or shares, sparkline), times (warn_after / crit_after), table (two labels pivoted), info. Anchors/aliases refused, 64 KB cap, YAML's on/off/yes/no flagged. - template_view.render(template, target, latest, now, history) -> View: a verdict naming the worst problem, then each block with its problems first (crit, warn, ok, then quiet states). Missing series show as "—" and say so; a template whose `expects` prefix the target lacks says so. - shown_value moves to service.py so the renderer can use it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
303 lines
12 KiB
Python
303 lines
12 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 _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")
|
|
row = Row(name=name or f["labels"] or block.title, tone=tone, text=f["state"] or "—",
|
|
why="" if tone in ("ok", "dim") else (f"is {f['state']}" if current else "reports no state"),
|
|
resource=f["resource"], metric=spec["metric"])
|
|
for col in spec["columns"]:
|
|
match = [s for s in index.get(col["metric"], []) if s.labels == f["labels"]]
|
|
if not match and name:
|
|
match = [s for s in index.get(col["metric"], []) if s.pairs.get(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)
|