home-network-k3s-impl/platform/monitoring/tests/extract_rules.py

#!/usr/bin/env python3
"""Flatten this repo's PrometheusRule CRs into a plain Prometheus rule file.

`promtool test rules` consumes a rule file (`{groups: [...]}`), not a Kubernetes
PrometheusRule object, so the committed manifests cannot be fed to it directly. This
script is the (deliberately dumb) adapter: it walks the given YAML files, picks out every
`kind: PrometheusRule` document, and concatenates their `spec.groups` into one rule file.

Why an adapter instead of a second copy of the rules: a rule-unit test is only worth
anything if it tests the EXACT expression that ships. Copying expressions into a fixture
would let the fixture stay green while the deployed rule rots -- which is the class of bug
this harness exists to catch (a permanently-firing CNPG alert and a structurally inert
Velero alert both reached prod in 2026-09).

Flux `postBuild` placeholders (`${BASE_DOMAIN}`) are left verbatim: no expression in
platform/monitoring uses one, and leaving them alone keeps annotation text identical to
the file under test. If a placeholder ever appears inside an `expr`, that is refused
loudly rather than silently producing an untestable rule file.

Group names are made unique per source file so two files may legally reuse a name.

Usage:
    extract_rules.py --out generated.rules.yaml alerts.yaml [more.yaml ...]
"""

from __future__ import annotations

import argparse
import re
import sys
from pathlib import Path

try:
    import yaml
except ImportError:  # pragma: no cover - reported by the runner as a SKIP, never a pass
    print(
        "extract_rules.py: PyYAML is required (it ships transitively with the pinned "
        "yamllint used by ci/lint.sh: pip install 'yamllint==1.35.1')",
        file=sys.stderr,
    )
    raise SystemExit(4)

PLACEHOLDER = re.compile(r"\$\{[A-Za-z_][A-Za-z0-9_]*\}")


def extract(paths: list[Path]) -> dict:
    groups: list[dict] = []
    seen: set[str] = set()
    for path in paths:
        docs = [d for d in yaml.safe_load_all(path.read_text()) if d]
        rules_docs = [d for d in docs if d.get("kind") == "PrometheusRule"]
        if not rules_docs:
            raise SystemExit(f"{path}: contains no PrometheusRule documents")
        for doc in rules_docs:
            for group in doc.get("spec", {}).get("groups", []) or []:
                for rule in group.get("rules", []) or []:
                    expr = str(rule.get("expr", ""))
                    if PLACEHOLDER.search(expr):
                        raise SystemExit(
                            f"{path}: rule {rule.get('alert') or rule.get('record')!r} has a "
                            f"postBuild placeholder inside its expr; it cannot be unit-tested "
                            f"as written"
                        )
                # Namespace the group by source file so names cannot collide.
                group = dict(group)
                name = f"{path.stem}::{group['name']}"
                if name in seen:
                    raise SystemExit(f"duplicate rule group {name!r}")
                seen.add(name)
                group["name"] = name
                groups.append(group)
    if not groups:
        raise SystemExit("no rule groups found")
    return {"groups": groups}


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--out", required=True, type=Path, help="rule file to write")
    ap.add_argument("manifests", nargs="+", type=Path, help="PrometheusRule YAML files")
    args = ap.parse_args()

    out = extract(args.manifests)
    args.out.write_text(yaml.safe_dump(out, sort_keys=False, default_flow_style=False))
    n = sum(len(g.get("rules", []) or []) for g in out["groups"])
    print(
        f"extract_rules.py: {len(out['groups'])} group(s), {n} rule(s) -> {args.out}",
        file=sys.stderr,
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())