Metrics Cardinality skill for Codex

Gives the Codex app a display name, a one-line description and a starter prompt for this skill.

interface:
  display_name: "Metrics Cardinality"
  short_description: "Find the label that is eating your metrics bill"
  default_prompt: "$metrics-cardinality Prometheus memory doubled this month, find which metrics and labels caused it"

#!/usr/bin/env python3
"""Report metric cardinality from a Prometheus exposition dump or a TSDB status JSON.

Usage:
  curl -s localhost:8080/metrics > metrics.txt
  python3 scripts/cardinality_report.py metrics.txt [--top 15] [--max-series 1000]

  curl -s 'http://prometheus:9090/api/v1/status/tsdb?limit=50' > tsdb.json
  python3 scripts/cardinality_report.py tsdb.json

For exposition text it prints series per metric, distinct values per label per metric,
and labels whose values look unbounded (IDs, UUIDs, IPs, long paths, emails).
Exits 1 if any metric exceeds --max-series or a suspicious label is found, else 0.
Standard library only.
"""
import argparse
import json
import re
import sys
from collections import defaultdict

LINE = re.compile(r'^([a-zA-Z_:][a-zA-Z0-9_:]*)(\{(.*)\})?\s+\S+')
LABEL = re.compile(r'([a-zA-Z_][a-zA-Z0-9_]*)="((?:[^"\\]|\\.)*)"')
SUSPICIOUS = [
    ("uuid", re.compile(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", re.I)),
    ("long hex id", re.compile(r"\b[0-9a-f]{16,}\b", re.I)),
    ("ip address", re.compile(r"\b\d{1,3}(\.\d{1,3}){3}\b")),
    ("email", re.compile(r"[^@\s]+@[^@\s]+\.[a-z]{2,}", re.I)),
    ("numeric id in path", re.compile(r"/\d{2,}(/|$)")),
]
SUFFIXES = ("_bucket", "_sum", "_count", "_created", "_total")


def family(name, types):
    for s in SUFFIXES:
        if name.endswith(s) and name[: -len(s)] in types:
            return name[: -len(s)]
    return name


def report_exposition(text, top, max_series):
    types = {}
    series = defaultdict(int)
    values = defaultdict(lambda: defaultdict(set))
    for line in text.splitlines():
        if line.startswith("# TYPE "):
            parts = line.split()
            if len(parts) >= 4:
                types[parts[2]] = parts[3]
            continue
        if not line or line.startswith("#"):
            continue
        m = LINE.match(line)
        if not m:
            continue
        fam = family(m.group(1), types)
        series[fam] += 1
        for k, v in LABEL.findall(m.group(3) or ""):
            if k != "le" and k != "quantile":
                values[fam][k].add(v)

    total = sum(series.values())
    print(f"total series in dump: {total}, metric families: {len(series)}\n")
    print(f"{'series':>8}  {'type':10} metric  (labels: distinct values)")
    problems = 0
    for fam, n in sorted(series.items(), key=lambda kv: -kv[1])[:top]:
        labels = ", ".join(f"{k}:{len(v)}" for k, v in sorted(values[fam].items(), key=lambda kv: -len(kv[1])))
        flag = "  <-- over budget" if n > max_series else ""
        print(f"{n:>8}  {types.get(fam, '?'):10} {fam}  ({labels}){flag}")
        if n > max_series:
            problems += 1

    findings = []
    for fam, labels in values.items():
        for k, vals in labels.items():
            hits, sample = set(), None
            for v in sorted(vals):
                for name, rx in SUSPICIOUS:
                    if rx.search(v):
                        hits.add(name)
                        sample = sample or v[:60]
            if hits:
                findings.append(f"{fam}{{{k}}}: {len(vals)} values, looks like {', '.join(sorted(hits))} (e.g. {sample!r})")
    if findings:
        print("\nlabels with unbounded-looking values:")
        for f in sorted(findings):
            print(f"  {f}")
        problems += len(findings)
    print("\nnote: one dump is one target; multiply by replicas and other label sets for the server-wide count.")
    return 1 if problems else 0


def report_tsdb(data, top, max_series):
    d = data.get("data", data)
    head = d.get("headStats", {})
    if head:
        print(f"head series: {head.get('numSeries')}, chunks: {head.get('chunkCount')}\n")
    problems = 0
    sections = [
        ("seriesCountByMetricName", "series by metric"),
        ("labelValueCountByLabelName", "distinct values by label name"),
        ("seriesCountByLabelValuePair", "series by label=value pair"),
        ("memoryInBytesByLabelName", "memory bytes by label name"),
    ]
    for key, title in sections:
        rows = d.get(key) or []
        if not rows:
            continue
        print(title)
        for r in rows[:top]:
            flag = ""
            if key == "seriesCountByMetricName" and r.get("value", 0) > max_series:
                flag = "  <-- over budget"
                problems += 1
            print(f"  {r.get('value'):>10}  {r.get('name')}{flag}")
        print()
    return 1 if problems else 0


def main():
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("path", help="exposition text or TSDB status JSON; - for stdin")
    ap.add_argument("--top", type=int, default=15)
    ap.add_argument("--max-series", type=int, default=1000, help="per-metric budget for the over-budget flag")
    a = ap.parse_args()
    try:
        text = sys.stdin.read() if a.path == "-" else open(a.path, encoding="utf-8", errors="replace").read()
    except OSError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2
    if text.lstrip().startswith("{"):
        try:
            return report_tsdb(json.loads(text), a.top, a.max_series)
        except json.JSONDecodeError as exc:
            print(f"error: invalid JSON: {exc}", file=sys.stderr)
            return 2
    return report_exposition(text, a.top, a.max_series)


if __name__ == "__main__":
    sys.exit(main())