feat(prometheus): fold state sets on the target page; forget series an endpoint drops (#5689)

#5689 "Prometheus target page: fold state sets, show epoch times, size
bytes", step 1 of milestone 508 "Prometheus widget templates".

- A state set (name ending in _state, a state label, every series 0 or 1;
  or the OpenMetrics form, the label named after the metric) reads as one
  row per thing it describes, naming the current state, the others on
  hover. fabledarchives_jobs{state} is a count and stays as it is.
- label_pairs / state_label / fold_states in service.py, for the widget
  templates to reuse.
- After a good scrape, series it no longer has lose their newest value
  (forget_gone): Archives' old build_info row read as current beside the
  new one. History stays; a failed scrape forgets nothing; alert rules keep
  their own last-sample time.
- Epoch times and _bytes sizes were already read by shown_value /
  metric_value.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

docs/plugins/prometheus.md

@@ -46,6 +46,10 @@ Each target also records four metrics about itself, under its own name:
 
 A value's unit is read from Prometheus naming: `_bytes` as bytes, `_seconds` as a span, `_timestamp_seconds` as a moment ("3h ago"), `_per_sec` as a rate.
 
+A **state set** is one series per possible state, the current one 1 and the rest 0, such as `fabledarchives_backup_target_state{target, state}`. The target page shows it as one row per target (or job), naming the current state; the other states are listed on hover. A metric is read as a state set when its name ends in `_state` and every series has a `state` label reading 0 or 1. The OpenMetrics form, where the label is named after the metric (`foo{foo="a"} 1`), counts too. Each state is still stored as its own series, so an alert rule can watch one state.
+
+**Only what the endpoint reports now is current.** When a scrape succeeds, any series it no longer has loses its newest value. That happens when the endpoint stopped reporting it, for example an old `build_info` after an upgrade, or when a filter or the cap now leaves it out. Its history stays on the chart until retention removes it. A failed scrape leaves the newest values as they were.
+
 Renaming a target moves its stored history to the new name. Deleting one removes its newest values; its history ages out with retention (Settings → Retention).
 
 ---

plugins/prometheus/routes.py

@@ -20,8 +20,8 @@ from steward.models.users import UserRole
 
 from .models import PrometheusTarget
 from .scraper import ScrapeError, scrape
-from .service import (OWN_METRICS, SOURCE, FormError, form_values, map_families, parse_form,
-                      patterns, target_state)
+from .service import (OWN_METRICS, SOURCE, FormError, fold_states, form_values, map_families,
+                      parse_form, patterns, target_state)
 
 prometheus_bp = Blueprint("prometheus", __name__, template_folder="templates")
 
@@ -256,7 +256,9 @@ async def test_target():
 
 def group_series(target_name: str, latest: list) -> list[dict]:
     """The target's stored series grouped by metric, each with its rows:
-    the labels that tell them apart, and the newest value."""
+    the labels that tell them apart, and the newest value. A state set is
+    folded to one row per thing it describes, naming the current state
+    (`states` set on the group)."""
     groups: dict[str, list] = {}
     for row in latest:
         if row.resource_name == target_name and row.metric_name in OWN_METRICS:
@@ -264,9 +266,18 @@ def group_series(target_name: str, latest: list) -> list[dict]:
         labels = row.resource_name[len(target_name) + 1:] if row.resource_name != target_name else ""
         groups.setdefault(row.metric_name, []).append({
             "labels": labels, "resource": row.resource_name, "at": row.recorded_at,
-            **shown_value(row.metric_name, row.value)})
-    return [{"metric": m, "rows": sorted(rows, key=lambda r: r["labels"])}
-            for m, rows in sorted(groups.items())]
+            "value": row.value, **shown_value(row.metric_name, row.value)})
+    out = []
+    for metric, rows in sorted(groups.items()):
+        folded = fold_states(metric, rows)
+        if folded is not None:
+            out.append({"metric": metric, "states": True, "rows": [
+                {"labels": f["labels"], "resource": f["resource"], "moment": None,
+                 "text": f["state"] or "none", "states": f["states"]} for f in folded]})
+        else:
+            out.append({"metric": metric, "states": False,
+                        "rows": sorted(rows, key=lambda r: r["labels"])})
+    return out
 
 
 @prometheus_bp.get("/targets/<target_id>")

plugins/prometheus/scheduler.py

@@ -83,18 +83,46 @@ async def scrape_due(app: "Quart", scrape=None) -> None:
     await record(app, targets, results, datetime.now(timezone.utc))
 
 
+async def forget_gone(session, target: str, kept: set[tuple[str, str]]) -> None:
+    """Drop the newest value of each series of `target` that its last scrape
+    didn't keep: the endpoint stopped reporting it (an old build_info after an
+    upgrade, a deleted job), or a filter or the cap now leaves it out. Pages,
+    widgets and query_metrics read the newest values, so they show what the
+    endpoint says now. History stays for charts and ages out with retention.
+    Alert rules keep their own last-sample time, so one on a gone series
+    still goes no_data."""
+    from sqlalchemy import delete, tuple_
+
+    from steward.models.metrics import PluginMetricLatest
+
+    from .service import SOURCE
+
+    L = PluginMetricLatest
+    stored = (await session.execute(select(L.resource_name, L.metric_name).where(
+        L.source_module == SOURCE,
+        (L.resource_name == target) | L.resource_name.startswith(target + ":", autoescape=True)))).all()
+    gone = [(r, m) for r, m in stored if (r, m) not in kept]
+    if gone:
+        await session.execute(delete(L).where(
+            L.source_module == SOURCE, tuple_(L.resource_name, L.metric_name).in_(gone)))
+
+
 async def record(app: "Quart", targets, results, now: datetime) -> None:
     """Each target's outcome on its row, and its kept series (counters as
-    rates) plus its own is_up / scrape_ms / series counts as metrics."""
+    rates) plus its own is_up / scrape_ms / series counts as metrics. After a
+    good scrape, the series it no longer has are forgotten (forget_gone); a
+    failed scrape says nothing about them, so they stay."""
     from steward.core.alerts import MetricBatch
 
     from .models import PrometheusTarget
     from .scraper import ScrapeError
-    from .service import (IS_UP, SCRAPE_MS, SERIES_CAPPED, SERIES_KEPT, SOURCE, map_families)
+    from .service import (IS_UP, OWN_METRICS, SCRAPE_MS, SERIES_CAPPED, SERIES_KEPT, SOURCE,
+                          map_families)
 
     async with app.db_sessionmaker() as session:
         async with session.begin():
             batch = MetricBatch()
+            answered: list[tuple[str, set[tuple[str, str]]]] = []
             for target, (result, ms) in zip(targets, results):
                 row = await session.get(PrometheusTarget, target.id)
                 if row is None:          # deleted while it was being scraped
@@ -114,6 +142,9 @@ async def record(app: "Quart", targets, results, now: datetime) -> None:
                 row.last_kept = len(mapped.series)
                 row.last_dropped = mapped.dropped
                 row.last_capped = mapped.capped
+                # A counter's first scrape gives no rate yet; it is still kept.
+                answered.append((row.name, {(x.resource, x.metric) for x in mapped.series}
+                                 | {(row.name, m) for m in OWN_METRICS}))
                 for series in mapped.series:
                     value = series.value
                     if series.counter:
@@ -128,3 +159,5 @@ async def record(app: "Quart", targets, results, now: datetime) -> None:
                     batch.add(source_module=SOURCE, resource_name=row.name,
                               metric_name=metric_name, value=value, recorded_at=now)
             await batch.record(session)
+            for name, kept in answered:
+                await forget_gone(session, name, kept)

plugins/prometheus/service.py

@@ -60,6 +60,67 @@ def rate_name(sample_name: str) -> str:
     return base + "_per_sec"
 
 
+def label_pairs(labels: str) -> list[tuple[str, str]]:
+    """resource_for's labels read back: "target=S,state=late" →
+    [("target", "S"), ("state", "late")]. A piece without "=" was part of the
+    value before it, a value that held a comma."""
+    pairs: list[tuple[str, str]] = []
+    for piece in labels.split(",") if labels else []:
+        key, eq, value = piece.partition("=")
+        if eq and key:
+            pairs.append((key, value))
+        elif pairs:
+            pairs[-1] = (pairs[-1][0], pairs[-1][1] + "," + piece)
+    return pairs
+
+
+def state_label(metric: str, pairs: list[tuple[str, str]]) -> str | None:
+    """Which label names a state set's state: `state`, or the metric's own
+    name as OpenMetrics writes a stateset (foo{foo="a"} 1)."""
+    keys = {k for k, _ in pairs}
+    for key in ("state", metric):
+        if key in keys:
+            return key
+    return None
+
+
+def fold_states(metric: str, rows: list[dict]) -> list[dict] | None:
+    """A state set's series folded to one row per thing it describes.
+
+    A state set is one series per possible state, the current one 1 and the
+    rest 0: Archives' fabledarchives_backup_target_state{target, state}. Each
+    row in `rows` has `labels` (resource_for's form), `value` and `resource`.
+    A metric is folded when its name ends in _state (or its state label is
+    its own name) and every series carries the state label and reads 0 or 1;
+    anything else is None, and shown series by series. The name matters: a
+    count split by a `state` label (fabledarchives_jobs{state="waiting"})
+    often reads 0 or 1 too, and is not a state set.
+
+    Each folded row: `labels` without the state, `state` (the one set to 1;
+    several are joined, none is ""), `states` (every state, in order) and
+    `resource` (the current state's series, else the first)."""
+    if not rows:
+        return None
+    sets: dict[str, list[tuple[str, dict]]] = {}
+    for row in rows:
+        pairs = label_pairs(row["labels"])
+        key = state_label(metric, pairs)
+        if key is None or row["value"] not in (0, 1):
+            return None
+        if key == "state" and not metric.endswith("_state"):
+            return None
+        rest = ",".join(f"{k}={v}" for k, v in pairs if k != key)
+        sets.setdefault(rest, []).append((dict(pairs)[key], row))
+    folded = []
+    for rest, members in sorted(sets.items()):
+        members.sort(key=lambda m: m[0])
+        current = [(s, r) for s, r in members if r["value"] == 1]
+        folded.append({"labels": rest, "state": ", ".join(s for s, _ in current),
+                       "states": [s for s, _ in members],
+                       "resource": (current or members)[0][1]["resource"]})
+    return folded
+
+
 @dataclass
 class Series:
     resource: str

plugins/prometheus/templates/prometheus/target.html

@@ -1,7 +1,8 @@
 {# plugins/prometheus/templates/prometheus/target.html — one scrape target
    (milestone 506): what its newest scrape found, then every series it
    stored, grouped by metric, each with its labels and newest value and a
-   link to its chart. #}
+   link to its chart. A state set reads as one row per thing it describes,
+   naming the current state (#5689). #}
 {% extends "base.html" %}
 {% from "_macros.html" import crumbs, page_header, status_badge, stat_tile %}
 {% block title %}Prometheus — {{ target.name }} — Steward{% endblock %}
@@ -54,10 +55,12 @@
     {% for g in groups %}
       {% for r in g.rows %}
       <tr>
-        {% if loop.first %}<td class="mono" rowspan="{{ g.rows|length }}" style="vertical-align:top;overflow-wrap:anywhere;">{{ g.metric }}</td>{% endif %}
+        {% if loop.first %}<td class="mono" rowspan="{{ g.rows|length }}" style="vertical-align:top;overflow-wrap:anywhere;">{{ g.metric }}
+          {%- if g.states %}<div class="cell-sub">a state set: the current state is shown</div>{% endif %}</td>{% endif %}
         <td class="mono" style="overflow-wrap:anywhere;font-size:0.82rem;">{{ r.labels or "—" }}</td>
         <td style="text-align:right;white-space:nowrap;">
-          <a href="/plugins/prometheus/targets/{{ target.id }}/series?resource={{ r.resource|urlencode }}&metric={{ g.metric|urlencode }}">{% if r.moment %}{{ r.moment|ago }}{% else %}{{ r.text }}{% endif %}</a>
+          <a href="/plugins/prometheus/targets/{{ target.id }}/series?resource={{ r.resource|urlencode }}&metric={{ g.metric|urlencode }}"
+             {%- if r.states %} title="One of: {{ r.states|join(', ') }}"{% endif %}>{% if r.moment %}{{ r.moment|ago }}{% else %}{{ r.text }}{% endif %}</a>
         </td>
       </tr>
       {% endfor %}

tests/integration/test_prometheus_scrape.py

@@ -1,6 +1,8 @@
 """Integration: a Prometheus scrape writes each target's outcome to its row,
 its kept series to the metric tables (a counter as a rate), and its own
-is_up, which an alert rule can watch (#5668). Requires STEWARD_DATABASE_URL."""
+is_up, which an alert rule can watch (#5668). A series a good scrape no
+longer reports loses its newest value (#5689); a failed scrape keeps them.
+Requires STEWARD_DATABASE_URL."""
 from __future__ import annotations
 
 import asyncio
@@ -14,7 +16,7 @@ pytestmark = pytest.mark.integration
 
 TEXT = """\
 # TYPE fa_backup_target_state gauge
-fa_backup_target_state{target="Steward",state="late"} 0
+fa_backup_target_state{target="Steward",state="STATE"} 1
 # TYPE fa_requests_total counter
 fa_requests_total COUNT
 # TYPE noise gauge
@@ -57,13 +59,14 @@ def test_a_scrape_records_series_rates_and_a_down_target_fires_its_rule(app):
     tag = uuid.uuid4().hex[:8]
     up, down = f"arch-{tag}", f"gone-{tag}"
     counts = iter([1_000, 1_011])
+    states = iter(["late", "fresh"])
     tokens = []
 
     async def fake_scrape(url, token, verify_tls):
         tokens.append(token)
         if "gone" in url:
             raise ScrapeError("Couldn't connect: connection refused.")
-        return prometheus.parse(TEXT.replace("COUNT", str(next(counts))))
+        return prometheus.parse(TEXT.replace("COUNT", str(next(counts))).replace("STATE", next(states)))
 
     async def latest(name_prefix):
         async with app.db_sessionmaker() as s:
@@ -117,7 +120,10 @@ def test_a_scrape_records_series_rates_and_a_down_target_fires_its_rule(app):
 
     first, second, rows, state, down_latest = asyncio.run(go())
     assert "scrape-token" in tokens
-    assert first[(f"{up}:target=Steward,state=late", "fa_backup_target_state")] == 0.0
+    assert first[(f"{up}:target=Steward,state=late", "fa_backup_target_state")] == 1.0
+    # The second scrape no longer reports "late": its newest value is forgotten.
+    assert (f"{up}:target=Steward,state=late", "fa_backup_target_state") not in second
+    assert second[(f"{up}:target=Steward,state=fresh", "fa_backup_target_state")] == 1.0
     assert (up, "fa_requests_per_sec") not in first                  # a counter needs two scrapes
     assert first[(up, "is_up")] == 1.0 and first[(up, "series_kept")] == 2.0
     assert not any(m == "noise" for _, m in first)                   # not in the keep list

tests/plugins/prometheus/test_pages.py

@@ -60,7 +60,11 @@ def test_series_are_grouped_by_metric_with_their_labels_and_units():
     assert [g["metric"] for g in groups] == ["fa_disk_free_bytes", "fa_last_backup_timestamp_seconds", "fa_state"]
     assert groups[0]["rows"][0]["labels"] == "" and groups[0]["rows"][0]["text"] == "2.0 GiB"
     assert groups[1]["rows"][0]["moment"] == datetime.fromtimestamp(stamp, timezone.utc)
-    assert [r["labels"] for r in groups[2]["rows"]] == ["target=S,state=fresh", "target=S,state=late"]
+    # A state set reads as one row naming the current state.
+    assert groups[2]["states"] and not groups[0]["states"]
+    assert [(r["labels"], r["text"], r["states"]) for r in groups[2]["rows"]] == [
+        ("target=S", "fresh", ["fresh", "late"])]
+    assert groups[2]["rows"][0]["resource"] == "A:target=S,state=fresh"
     assert shown_value("x_timestamp_seconds", 0) == {"text": "never", "moment": None}
     assert shown_value("jobs_per_sec", 0.25)["text"] == "0.25/s"
     assert shown_value("job_seconds", 90)["text"] == "1m 30s"
@@ -69,9 +73,9 @@ def test_series_are_grouped_by_metric_with_their_labels_and_units():
 
 async def test_the_target_page_leads_with_its_state_then_its_series(app):
     from quart import session
-    groups = [{"metric": "fa_state", "rows": [
-        {"labels": "target=S,state=late", "resource": "archives:target=S,state=late", "at": None,
-         "text": "0", "moment": None}]}]
+    groups = [{"metric": "fa_state", "states": True, "rows": [
+        {"labels": "target=S", "resource": "archives:target=S,state=late", "moment": None,
+         "text": "late", "states": ["fresh", "late"]}]}]
     async with app.test_request_context("/plugins/prometheus/targets/id-archives"):
         session["user_role"] = "viewer"
         page = await render_plugin_template(app, "prometheus", "prometheus/target.html",
@@ -81,6 +85,7 @@ async def test_the_target_page_leads_with_its_state_then_its_series(app):
     assert page.index("Answering") < page.index("Series stored") < page.index("fa_state")
     assert "3 series weren't stored" in page and "Keeping fabledarchives_*" in page
     assert "resource=archives%3Atarget%3DS%2Cstate%3Dlate" in page and "metric=fa_state" in page
+    assert 'title="One of: fresh, late">late</a>' in page and "a state set" in page
     assert 'href="/plugins/prometheus/">Prometheus</a>' in page          # its breadcrumb
     assert "/edit" not in page
 

tests/plugins/prometheus/test_service.py

@@ -8,8 +8,8 @@ import pytest
 from plugins.prometheus.models import PrometheusTarget
 from plugins.prometheus.scheduler import is_due
 from plugins.prometheus.scraper import ScrapeError, scrape
-from plugins.prometheus.service import (FormError, map_families, parse_form, patterns,
-                                        rate_name, resource_for, target_state, wanted)
+from plugins.prometheus.service import (FormError, fold_states, label_pairs, map_families, parse_form,
+                                        patterns, rate_name, resource_for, target_state, wanted)
 from steward.core import prometheus
 from steward.core.counter_rates import CounterRates
 
@@ -75,6 +75,38 @@ def test_names():
     assert patterns("a_*, b_*\n\n a_* \nc") == ["a_*", "b_*", "c"]
 
 
+def test_labels_read_back_as_stored_even_with_a_comma_in_a_value():
+    labels = resource_for("A", (("job", "Nightly"), ("schedule", "0 3 * * 1,3,5"), ("state", "ok")))
+    assert label_pairs(labels[2:]) == [("job", "Nightly"), ("schedule", "0 3 * * 1,3,5"), ("state", "ok")]
+    assert label_pairs("") == []
+
+
+def _rows(metric_rows):
+    return [{"labels": labels, "value": value, "resource": f"A:{labels}"} for labels, value in metric_rows]
+
+
+def test_a_state_set_folds_to_its_current_state_per_thing():
+    rows = _rows([("target=S,state=late", 0), ("target=S,state=fresh", 1), ("target=S,state=missing", 0),
+                  ("target=T,state=fresh", 0), ("target=T,state=late", 1), ("target=T,state=missing", 1),
+                  ("target=U,state=fresh", 0), ("target=U,state=late", 0)])
+    folded = fold_states("fa_backup_target_state", rows)
+    assert [(f["labels"], f["state"]) for f in folded] == [
+        ("target=S", "fresh"), ("target=T", "late, missing"), ("target=U", "")]
+    assert folded[0]["states"] == ["fresh", "late", "missing"]
+    assert folded[0]["resource"] == "A:target=S,state=fresh"            # the current one's chart
+    assert folded[2]["resource"] == "A:target=U,state=fresh"            # none current: the first
+    # OpenMetrics names a stateset's label after the metric itself.
+    assert fold_states("fa_mode", _rows([("fa_mode=a", 1), ("fa_mode=b", 0)]))[0]["state"] == "a"
+
+
+def test_only_a_state_set_is_folded():
+    # A count split by a state label reads 0 or 1 often enough; its name says it is a count.
+    assert fold_states("fa_jobs", _rows([("kind=scan,state=waiting", 1), ("kind=scan,state=running", 0)])) is None
+    assert fold_states("fa_state", _rows([("state=a", 1), ("state=b", 2)])) is None      # not 0/1
+    assert fold_states("fa_state", _rows([("state=a", 1), ("target=S", 0)])) is None     # a row without one
+    assert fold_states("fa_state", []) is None
+
+
 def test_a_counter_becomes_a_rate_and_a_reset_is_skipped():
     rates = CounterRates()
     t0 = datetime(2026, 10, 12, 12, 0, tzinfo=timezone.utc)