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>
458 lines
19 KiB
Python
458 lines
19 KiB
Python
"""Prometheus widget templates: the YAML a person writes, read and checked.
|
|
|
|
A template says which of a target's series a dashboard widget shows and
|
|
how. It is data only: it names series and lays them out in blocks, and
|
|
Steward draws them (template_view.py). It can't add a visual, a colour or a
|
|
formula; a new block is a Steward release (System 181's ruling).
|
|
|
|
name: Fabled Archives
|
|
version: 1
|
|
expects: fabledarchives_
|
|
blocks:
|
|
- states:
|
|
title: Backup targets
|
|
metric: fabledarchives_backup_target_state
|
|
name: target
|
|
healthy: [fresh]
|
|
quiet: [disabled, waiting]
|
|
warning: [late]
|
|
- values:
|
|
items:
|
|
- {label: Disk free, metric: fabledarchives_data_disk_free_bytes,
|
|
of: fabledarchives_data_disk_size_bytes, warn_below: 10%}
|
|
|
|
check(text) returns the Template, or every problem found, each with its
|
|
line and a sentence saying what to change.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
|
|
import yaml
|
|
|
|
FORMAT = 1
|
|
MAX_BYTES = 64 * 1024
|
|
MAX_BLOCKS = 20
|
|
MAX_ITEMS = 30
|
|
|
|
_METRIC = re.compile(r"^[a-zA-Z_:][a-zA-Z0-9_:]*$")
|
|
_LABEL = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]*$")
|
|
_DURATION = re.compile(r"^(\d+(?:\.\d+)?)\s*([smhd]?)$")
|
|
_SECONDS = {"": 1, "s": 1, "m": 60, "h": 3600, "d": 86400}
|
|
|
|
# What a unit shows a value as; the name is the suffix metric_value reads.
|
|
UNITS = {"bytes": "_bytes", "seconds": "_secs", "percent": "_pct", "per_sec": "_per_sec",
|
|
"count": "", "timestamp": "_timestamp_seconds"}
|
|
|
|
|
|
@dataclass
|
|
class Problem:
|
|
line: int
|
|
text: str
|
|
|
|
def __str__(self) -> str:
|
|
return f"Line {self.line}: {self.text}" if self.line else self.text
|
|
|
|
|
|
@dataclass
|
|
class Block:
|
|
kind: str # states | values | times | table | info
|
|
title: str
|
|
line: int
|
|
spec: dict = field(default_factory=dict)
|
|
|
|
|
|
@dataclass
|
|
class Template:
|
|
name: str
|
|
version: int
|
|
description: str = ""
|
|
expects: str = ""
|
|
blocks: list[Block] = field(default_factory=list)
|
|
text: str = ""
|
|
|
|
|
|
class TemplateError(ValueError):
|
|
def __init__(self, problems: list[Problem]):
|
|
self.problems = problems
|
|
super().__init__("; ".join(str(p) for p in problems))
|
|
|
|
|
|
# ── YAML with line numbers ───────────────────────────────────────────────────
|
|
|
|
class _Map(dict):
|
|
line = 0
|
|
lines: dict
|
|
|
|
|
|
class _List(list):
|
|
line = 0
|
|
lines: list
|
|
|
|
|
|
class _Loader(yaml.SafeLoader):
|
|
pass
|
|
|
|
|
|
def _construct_map(loader, node):
|
|
loader.flatten_mapping(node)
|
|
out = _Map()
|
|
out.line, out.lines = node.start_mark.line + 1, {}
|
|
for key_node, value_node in node.value:
|
|
key = loader.construct_object(key_node, deep=True)
|
|
if not isinstance(key, str):
|
|
raise yaml.constructor.ConstructorError(
|
|
None, None, "a key here should be a word, like `metric`", key_node.start_mark)
|
|
if key in out:
|
|
raise yaml.constructor.ConstructorError(
|
|
None, None, f"`{key}` is given twice", key_node.start_mark)
|
|
out[key] = loader.construct_object(value_node, deep=True)
|
|
out.lines[key] = key_node.start_mark.line + 1
|
|
return out
|
|
|
|
|
|
def _construct_list(loader, node):
|
|
out = _List(loader.construct_object(n, deep=True) for n in node.value)
|
|
out.line, out.lines = node.start_mark.line + 1, [n.start_mark.line + 1 for n in node.value]
|
|
return out
|
|
|
|
|
|
_Loader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, _construct_map)
|
|
_Loader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_SEQUENCE_TAG, _construct_list)
|
|
|
|
|
|
def _load(text: str) -> tuple[object, list[Problem]]:
|
|
if len(text.encode()) > MAX_BYTES:
|
|
return None, [Problem(0, f"A template is at most {MAX_BYTES // 1024} KB.")]
|
|
try:
|
|
# An alias can make a small file expand without end; a template has no use for one.
|
|
for event in yaml.parse(text, Loader=yaml.SafeLoader):
|
|
if isinstance(event, yaml.AliasEvent):
|
|
return None, [Problem(event.start_mark.line + 1,
|
|
"Anchors and aliases (&name, *name) aren't allowed; write the value out.")]
|
|
return yaml.load(text, Loader=_Loader), []
|
|
except yaml.constructor.ConstructorError as exc: # ours: a key twice, or not a word
|
|
mark = exc.problem_mark
|
|
return None, [Problem(mark.line + 1 if mark else 0, exc.problem[:1].upper() + exc.problem[1:] + ".")]
|
|
except yaml.MarkedYAMLError as exc:
|
|
mark = exc.problem_mark or exc.context_mark
|
|
return None, [Problem(mark.line + 1 if mark else 0, f"This isn't valid YAML: {exc.problem}.")]
|
|
except yaml.YAMLError as exc:
|
|
return None, [Problem(0, f"This isn't valid YAML: {exc}.")]
|
|
|
|
|
|
# ── Values ───────────────────────────────────────────────────────────────────
|
|
|
|
def duration(value) -> float | None:
|
|
"""Seconds from 90, "90s", "15m", "26h" or "2d"; None if it isn't one."""
|
|
if isinstance(value, bool):
|
|
return None
|
|
if isinstance(value, (int, float)):
|
|
return float(value) if value >= 0 else None
|
|
m = _DURATION.match(str(value).strip().lower())
|
|
return float(m.group(1)) * _SECONDS[m.group(2)] if m else None
|
|
|
|
|
|
@dataclass
|
|
class Limit:
|
|
"""A threshold: a number, or a share of the item's `of` ("10%")."""
|
|
value: float
|
|
share: bool = False
|
|
|
|
|
|
def limit(value) -> Limit | None:
|
|
if isinstance(value, bool):
|
|
return None
|
|
if isinstance(value, (int, float)):
|
|
return Limit(float(value))
|
|
text = str(value).strip()
|
|
if text.endswith("%"):
|
|
try:
|
|
return Limit(float(text[:-1]), share=True)
|
|
except ValueError:
|
|
return None
|
|
return None
|
|
|
|
|
|
# ── The checker ──────────────────────────────────────────────────────────────
|
|
|
|
class _Check:
|
|
def __init__(self):
|
|
self.problems: list[Problem] = []
|
|
|
|
def add(self, line: int, text: str) -> None:
|
|
self.problems.append(Problem(line, text))
|
|
|
|
def keys(self, m: _Map, allowed: tuple, required: tuple, where: str) -> bool:
|
|
ok = True
|
|
for key in m:
|
|
if key not in allowed:
|
|
self.add(m.lines[key], f"{where} has no `{key}`. It takes: {', '.join(allowed)}.")
|
|
ok = False
|
|
for key in required:
|
|
if key not in m:
|
|
self.add(m.line, f"{where} needs `{key}`.")
|
|
ok = False
|
|
return ok
|
|
|
|
def text(self, m: _Map, key: str, where: str, default: str = "") -> str:
|
|
if key not in m:
|
|
return default
|
|
value = m[key]
|
|
if isinstance(value, (dict, list)) or value is None:
|
|
self.add(m.lines[key], f"`{key}` in {where} should be text.")
|
|
return default
|
|
return str(value)
|
|
|
|
def metric(self, m: _Map, key: str, where: str) -> str:
|
|
value = m.get(key)
|
|
if not isinstance(value, str) or not _METRIC.match(value):
|
|
self.add(m.lines.get(key, m.line),
|
|
f"`{key}` in {where} should be a metric name as Steward stores it, like "
|
|
"fabledarchives_files (a counter is stored as <name>_per_sec).")
|
|
return ""
|
|
return value
|
|
|
|
def label(self, m: _Map, key: str, where: str) -> str:
|
|
if key not in m:
|
|
return ""
|
|
value = m[key]
|
|
if not isinstance(value, str) or not _LABEL.match(value):
|
|
self.add(m.lines[key], f"`{key}` in {where} should be a label name, like target.")
|
|
return ""
|
|
return value
|
|
|
|
def words(self, m: _Map, key: str, where: str) -> list[str]:
|
|
if key not in m:
|
|
return []
|
|
value = m[key]
|
|
if not isinstance(value, list) or not all(isinstance(v, (str, int)) for v in value):
|
|
self.add(m.lines[key], f"`{key}` in {where} should be a list, like [fresh, waiting].")
|
|
return []
|
|
if any(isinstance(v, bool) for v in value):
|
|
self.add(m.lines[key], f"`{key}` in {where} has a word YAML reads as true or false "
|
|
"(on, off, yes, no); put it in quotes, like 'off'.")
|
|
return []
|
|
return [str(v) for v in value]
|
|
|
|
def where(self, m: _Map, ctx: str) -> dict[str, str]:
|
|
if "where" not in m:
|
|
return {}
|
|
value = m["where"]
|
|
if not isinstance(value, dict) or not all(
|
|
isinstance(k, str) and _LABEL.match(k) and isinstance(v, (str, int, float))
|
|
and not isinstance(v, bool) for k, v in value.items()):
|
|
self.add(m.lines["where"], f"`where` in {ctx} should map labels to values, like "
|
|
"{kind: files}. Quote a value YAML would read as true or false.")
|
|
return {}
|
|
return {k: str(v) for k, v in value.items()}
|
|
|
|
def unit(self, m: _Map, ctx: str) -> str:
|
|
if "unit" not in m:
|
|
return ""
|
|
value = m["unit"]
|
|
if value not in UNITS:
|
|
self.add(m.lines["unit"], f"`unit` in {ctx} is one of: {', '.join(UNITS)}.")
|
|
return ""
|
|
return value
|
|
|
|
def span(self, m: _Map, key: str, ctx: str) -> float | None:
|
|
if key not in m:
|
|
return None
|
|
value = duration(m[key])
|
|
if value is None:
|
|
self.add(m.lines[key], f"`{key}` in {ctx} should be a span, like 26h, 90m or 2d.")
|
|
return value
|
|
|
|
def items(self, m: _Map, ctx: str) -> list[_Map]:
|
|
value = m.get("items")
|
|
if not isinstance(value, list) or not value:
|
|
self.add(m.lines.get("items", m.line), f"`items` in {ctx} should be a list of one or more items.")
|
|
return []
|
|
if len(value) > MAX_ITEMS:
|
|
self.add(m.lines["items"], f"{ctx} has {len(value)} items; at most {MAX_ITEMS}.")
|
|
out = []
|
|
for i, item in enumerate(value):
|
|
if not isinstance(item, dict):
|
|
self.add(value.lines[i], f"Each item in {ctx} should be a set of keys, like "
|
|
"{label: Disk free, metric: …}.")
|
|
else:
|
|
out.append(item)
|
|
return out
|
|
|
|
|
|
# Each block: the keys it takes, those it needs, and its title when none is given.
|
|
_BLOCKS = {
|
|
"states": (("title", "metric", "where", "name", "healthy", "quiet", "warning", "warn_past",
|
|
"crit_past", "columns"),
|
|
("metric", "healthy"), "States"),
|
|
"values": (("title", "items"), ("items",), "Figures"),
|
|
"times": (("title", "items"), ("items",), "Times"),
|
|
"table": (("title", "metric", "where", "rows", "columns", "unit"), ("metric", "rows", "columns"),
|
|
"Table"),
|
|
"info": (("title", "metric", "where", "labels"), ("metric",), "About"),
|
|
}
|
|
_VALUE_ITEM = ("label", "metric", "where", "name", "of", "unit", "warn_above", "crit_above",
|
|
"warn_below", "crit_below", "sparkline")
|
|
_TIME_ITEM = ("label", "metric", "where", "name", "warn_after", "crit_after")
|
|
_COLUMN = ("label", "metric", "unit")
|
|
|
|
|
|
def _states(c: _Check, m: _Map, ctx: str) -> dict:
|
|
spec = {"metric": c.metric(m, "metric", ctx), "where": c.where(m, ctx), "name": c.label(m, "name", ctx),
|
|
"healthy": c.words(m, "healthy", ctx), "quiet": c.words(m, "quiet", ctx),
|
|
"warning": c.words(m, "warning", ctx), "columns": [],
|
|
# A deadline per row (a *_timestamp_seconds): once it has passed, a
|
|
# healthy row reads as a warning or a problem. A job that says
|
|
# "running" for days is still overdue.
|
|
"warn_past": c.metric(m, "warn_past", ctx) if "warn_past" in m else "",
|
|
"crit_past": c.metric(m, "crit_past", ctx) if "crit_past" in m else ""}
|
|
if "healthy" in m and not spec["healthy"] and isinstance(m["healthy"], list):
|
|
c.add(m.lines["healthy"], f"`healthy` in {ctx} needs at least one state.")
|
|
seen: dict[str, str] = {}
|
|
for key in ("healthy", "quiet", "warning"):
|
|
for state in spec[key]:
|
|
if state in seen:
|
|
c.add(m.lines[key], f"“{state}” is in both `{seen[state]}` and `{key}` in {ctx}.")
|
|
seen[state] = key
|
|
if "columns" in m:
|
|
cols = m["columns"]
|
|
if not isinstance(cols, list):
|
|
c.add(m.lines["columns"], f"`columns` in {ctx} should be a list of {{label, metric}}.")
|
|
else:
|
|
for i, col in enumerate(cols):
|
|
where = f"a column of {ctx}"
|
|
if not isinstance(col, dict):
|
|
c.add(cols.lines[i], f"Each of `columns` in {ctx} should be {{label, metric}}.")
|
|
continue
|
|
if c.keys(col, _COLUMN, ("label", "metric"), where):
|
|
spec["columns"].append({"label": c.text(col, "label", where),
|
|
"metric": c.metric(col, "metric", where),
|
|
"unit": c.unit(col, where)})
|
|
return spec
|
|
|
|
|
|
def _value_item(c: _Check, item: _Map, ctx: str) -> dict | None:
|
|
where = f"an item of {ctx}"
|
|
if not c.keys(item, _VALUE_ITEM, ("label", "metric"), where):
|
|
return None
|
|
spec = {"label": c.text(item, "label", where), "metric": c.metric(item, "metric", where),
|
|
"where": c.where(item, where), "name": c.label(item, "name", where),
|
|
"of": c.metric(item, "of", where) if "of" in item else "", "unit": c.unit(item, where),
|
|
"sparkline": False}
|
|
if "sparkline" in item:
|
|
if not isinstance(item["sparkline"], bool):
|
|
c.add(item.lines["sparkline"], f"`sparkline` in {where} is true or false.")
|
|
spec["sparkline"] = item["sparkline"] is True
|
|
for key in ("warn_above", "crit_above", "warn_below", "crit_below"):
|
|
spec[key] = None
|
|
if key in item:
|
|
lim = limit(item[key])
|
|
if lim is None:
|
|
c.add(item.lines[key], f"`{key}` in {where} should be a number, or a share like 10%.")
|
|
elif lim.share and not spec["of"]:
|
|
c.add(item.lines[key], f"`{key}` in {where} is a share ({item[key]}), which needs `of`: "
|
|
"the metric it is a share of.")
|
|
spec[key] = lim
|
|
return spec
|
|
|
|
|
|
def _time_item(c: _Check, item: _Map, ctx: str) -> dict | None:
|
|
where = f"an item of {ctx}"
|
|
if not c.keys(item, _TIME_ITEM, ("label", "metric"), where):
|
|
return None
|
|
spec = {"label": c.text(item, "label", where), "metric": c.metric(item, "metric", where),
|
|
"where": c.where(item, where), "name": c.label(item, "name", where),
|
|
"warn_after": c.span(item, "warn_after", where), "crit_after": c.span(item, "crit_after", where)}
|
|
if spec["warn_after"] and spec["crit_after"] and spec["crit_after"] < spec["warn_after"]:
|
|
c.add(item.lines["crit_after"], f"`crit_after` in {where} should be later than `warn_after`.")
|
|
return spec
|
|
|
|
|
|
def _block(c: _Check, raw, line: int, n: int) -> Block | None:
|
|
if not isinstance(raw, dict) or len(raw) != 1:
|
|
c.add(line, f"Block {n} should be one kind with its settings under it, like "
|
|
f"`- states: {{metric: …}}`. The kinds are: {', '.join(_BLOCKS)}.")
|
|
return None
|
|
kind, body = next(iter(raw.items()))
|
|
if kind not in _BLOCKS:
|
|
c.add(line, f"Block {n} is “{kind}”; the kinds are: {', '.join(_BLOCKS)}.")
|
|
return None
|
|
if not isinstance(body, dict):
|
|
c.add(line, f"Block {n} ({kind}) should have its settings under it, like `metric: …`.")
|
|
return None
|
|
allowed, required, default_title = _BLOCKS[kind]
|
|
ctx = f"block {n} ({kind})"
|
|
if not c.keys(body, allowed, required, ctx.capitalize()):
|
|
return None
|
|
title = c.text(body, "title", ctx, default_title)
|
|
if kind == "states":
|
|
spec = _states(c, body, ctx)
|
|
elif kind in ("values", "times"):
|
|
reader = _value_item if kind == "values" else _time_item
|
|
spec = {"items": [s for s in (reader(c, i, ctx) for i in c.items(body, ctx)) if s]}
|
|
elif kind == "table":
|
|
spec = {"metric": c.metric(body, "metric", ctx), "where": c.where(body, ctx),
|
|
"rows": c.label(body, "rows", ctx), "columns": c.label(body, "columns", ctx),
|
|
"unit": c.unit(body, ctx)}
|
|
else:
|
|
spec = {"metric": c.metric(body, "metric", ctx), "where": c.where(body, ctx),
|
|
"labels": c.words(body, "labels", ctx)}
|
|
return Block(kind, title, line, spec)
|
|
|
|
|
|
_TOP = ("name", "description", "version", "format", "expects", "blocks")
|
|
|
|
|
|
def check(text: str) -> Template:
|
|
"""The template `text` describes. Raises TemplateError with every problem
|
|
found, each with its line, in the order they appear."""
|
|
raw, problems = _load(text)
|
|
if problems:
|
|
raise TemplateError(problems)
|
|
c = _Check()
|
|
if not isinstance(raw, dict):
|
|
raise TemplateError([Problem(1, "A template starts with its keys: name, version and blocks.")])
|
|
c.keys(raw, _TOP, ("name", "version", "blocks"), "A template")
|
|
name = c.text(raw, "name", "the template").strip()
|
|
if "name" in raw and not name:
|
|
c.add(raw.lines["name"], "Give the template a name.")
|
|
if len(name) > 80:
|
|
c.add(raw.lines["name"], "A name is at most 80 characters.")
|
|
version = raw.get("version")
|
|
if "version" in raw and (isinstance(version, bool) or not isinstance(version, int) or version < 1):
|
|
c.add(raw.lines["version"], "`version` is the template's own version: a whole number from 1. "
|
|
"Raise it when you change a template you have shared.")
|
|
if "format" in raw and raw["format"] != FORMAT:
|
|
c.add(raw.lines["format"], f"This Steward reads template format {FORMAT}.")
|
|
expects = c.text(raw, "expects", "the template")
|
|
if expects and not re.match(r"^[a-zA-Z_:][a-zA-Z0-9_:]*$", expects):
|
|
c.add(raw.lines["expects"], "`expects` should be the start of the metric names it reads, "
|
|
"like fabledarchives_.")
|
|
blocks: list[Block] = []
|
|
raw_blocks = raw.get("blocks")
|
|
if "blocks" in raw:
|
|
if not isinstance(raw_blocks, list) or not raw_blocks:
|
|
c.add(raw.lines["blocks"], "`blocks` should be a list of one or more blocks.")
|
|
else:
|
|
if len(raw_blocks) > MAX_BLOCKS:
|
|
c.add(raw.lines["blocks"], f"A template has at most {MAX_BLOCKS} blocks.")
|
|
for i, b in enumerate(raw_blocks):
|
|
block = _block(c, b, raw_blocks.lines[i], i + 1)
|
|
if block:
|
|
blocks.append(block)
|
|
if c.problems:
|
|
raise TemplateError(sorted(c.problems, key=lambda p: p.line))
|
|
return Template(name=name, version=version, description=c.text(raw, "description", "the template"),
|
|
expects=expects, blocks=blocks, text=text)
|
|
|
|
|
|
def problems(text: str) -> list[Problem]:
|
|
"""Every problem with `text`; empty when it is a template."""
|
|
try:
|
|
check(text)
|
|
except TemplateError as exc:
|
|
return exc.problems
|
|
return []
|