Anomaly intelligence for incident response

LABELS_CONTRACT.md

@@ -40,6 +40,20 @@ Both tools propagate the underlying time-series labels (`instance`, `pod`, `devi
 
 This is the reason `honor_labels: true` is required on the Prometheus scrape config for both tools' `/metrics` endpoints.
 
+### `day_of_week` (contracted when present, identical semantics in both tools)
+
+Emitted on recurrence-analysis metrics (`anomaly_recurrence_score` in promanomaly, incident-pattern metrics in promforecast). Represents the ISO 8601 day of week as a three-letter abbreviation: `Mon`, `Tue`, `Wed`, `Thu`, `Fri`, `Sat`, `Sun`.
+
+- Type: string, one of the seven values above.
+- Stability: label values are fixed (the seven abbreviations); adding or removing is a breaking change.
+
+### `hour_of_day` (contracted when present, identical semantics in both tools)
+
+Emitted alongside `day_of_week` on recurrence-analysis metrics. Represents the hour in UTC as a string (`"0"` through `"23"`).
+
+- Type: string, `"0"` through `"23"`.
+- Stability: same as `day_of_week`.
+
 ## Tool-private labels (NOT contracted)
 
 Either tool may add its own labels to its own output metrics without coordination, as long as they don't collide with the contracted set above.

detector/src/promanomaly/anomalies.py

@@ -18,13 +18,20 @@
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any
 
+from .detectors.cohort import cohort_key_for
+
 if TYPE_CHECKING:  # pragma: no cover - import only for type hints
     from .state import SnapshotStore
 
 # Canonical labels stamped by the detector; everything else on a sample
 # is a source label echoed from the query.
 _CANONICAL = frozenset({"id", "group", "detector", "detector_instance"})
 
+# Labels that are not part of a series' fleet identity: the detector axis
+# (so the same source series counted once regardless of how many detectors
+# scored it) and the cohort meta-label. Mirrors the density rollup's rule.
+_IDENTITY_DROP = frozenset({"detector", "detector_instance", "cohort_label"})
+
 
 @dataclass(frozen=True)
 class FiringSeries:
@@ -139,4 +146,198 @@ def collect_firing(
     return results
 
 
-__all__ = ["FiringSeries", "collect_firing"]
+@dataclass(frozen=True)
+class BlastRadiusRow:
+    """Co-firing rollup for one group or one cohort in the current run.
+
+    A point-in-time read off the current snapshot — not a stateful
+    correlation engine and not tied to any event identity. ``firing`` is
+    how many distinct members of the scope are over their threshold right
+    now, ``total`` is how many members the scope has at all, ``fraction``
+    is their ratio, and ``max_co_firing_seconds`` is the longest any of
+    the co-firing members has been continuously breached.
+    """
+
+    scope: str
+    group: str
+    firing: int
+    total: int
+    fraction: float
+    max_co_firing_seconds: float
+    cohort_label: str | None = None
+    cohort_key: dict[str, str] | None = None
+
+    def as_dict(self) -> dict[str, Any]:
+        out: dict[str, Any] = {
+            "scope": self.scope,
+            "group": self.group,
+            "firing": self.firing,
+            "total": self.total,
+            "fraction": self.fraction,
+            "max_co_firing_seconds": self.max_co_firing_seconds,
+        }
+        if self.cohort_label is not None:
+            out["cohort_label"] = self.cohort_label
+        if self.cohort_key is not None:
+            out["cohort_key"] = dict(self.cohort_key)
+        return out
+
+
+def _series_identity(labels: dict[str, str]) -> tuple[tuple[str, str], ...]:
+    """Fleet identity of a series — drops detector and cohort meta-labels.
+
+    The same source series scored by several detectors collapses to one
+    identity, matching the density rollup so the two never disagree on how
+    many members a scope has.
+    """
+    items = {k: v for k, v in labels.items() if k not in _IDENTITY_DROP}
+    return tuple(sorted(items.items()))
+
+
+def collect_blast_radius(store: SnapshotStore) -> list[BlastRadiusRow]:
+    """Compute per-group and per-cohort co-firing rollups for the current run.
+
+    Args:
+        store: the live snapshot store. Only the most recent snapshot per
+            group is read; no TSDB queries are issued.
+
+    For each group, and for each cohort within a group whose samples carry
+    a ``cohort_label`` axis, the rollup reports how many members are firing
+    now, the total member count, the firing fraction, and the longest
+    continuous breach across the co-firing members. Group-level counts are
+    taken from the already-emitted ``anomaly_active_series`` /
+    ``anomaly_density`` rollup samples when present (recovering the total as
+    ``round(active / fraction)``), and recomputed from
+    ``anomaly_outside_threshold`` samples otherwise. Series still warming up
+    are excluded from both counts, consistent with the density rollup.
+
+    Returns:
+        Rows sorted by fraction descending, then firing count descending,
+        then ``(group, cohort_key)`` for a stable order.
+    """
+    # Per group: identity sets for total / firing, the longest breach among
+    # firing members, and the group-level rollup samples if the runner
+    # already emitted them.
+    group_total: dict[str, set[tuple[tuple[str, str], ...]]] = {}
+    group_firing: dict[str, set[tuple[tuple[str, str], ...]]] = {}
+    group_warming: dict[str, set[tuple[tuple[str, str], ...]]] = {}
+    group_max_duration: dict[str, float] = {}
+    group_active: dict[str, float] = {}
+    group_density: dict[str, float] = {}
+
+    # Per cohort: keyed by (group, cohort_label, cohort_key) so two axes in
+    # the same group stay separate.
+    CohortId = tuple[str, str, tuple[tuple[str, str], ...]]
+    cohort_total: dict[CohortId, set[tuple[tuple[str, str], ...]]] = {}
+    cohort_firing: dict[CohortId, set[tuple[tuple[str, str], ...]]] = {}
+    cohort_warming: dict[CohortId, set[tuple[tuple[str, str], ...]]] = {}
+    cohort_max_duration: dict[CohortId, float] = {}
+
+    # Durations are joined on the full per-detector key so the longest
+    # breach is read from the same series that is firing.
+    durations: dict[tuple[tuple[str, str], ...], float] = {}
+
+    for snap in store.all_snapshots():
+        for sample in snap.samples:
+            labels = dict(sample.labels)
+            grp = labels.get("group", "")
+            metric = sample.metric
+            if metric == "anomaly_duration_seconds":
+                durations[_detector_key(labels)] = sample.value
+            elif metric == "anomaly_active_series" and set(labels) == {"group"}:
+                group_active[grp] = sample.value
+            elif metric == "anomaly_density" and set(labels) == {"group"}:
+                group_density[grp] = sample.value
+
+    for snap in store.all_snapshots():
+        for sample in snap.samples:
+            labels = dict(sample.labels)
+            if sample.metric == "anomaly_warming_up" and sample.value >= 1.0:
+                grp = labels.get("group", "")
+                ident = _series_identity(labels)
+                group_warming.setdefault(grp, set()).add(ident)
+                axis = labels.get("cohort_label")
+                if axis is not None:
+                    key = cohort_key_for(labels, axis)
+                    cohort_warming.setdefault((grp, axis, key), set()).add(ident)
+
+    for snap in store.all_snapshots():
+        for sample in snap.samples:
+            if sample.metric != "anomaly_outside_threshold":
+                continue
+            labels = dict(sample.labels)
+            grp = labels.get("group", "")
+            ident = _series_identity(labels)
+            group_total.setdefault(grp, set()).add(ident)
+            firing_now = sample.value >= 1.0
+            if firing_now:
+                group_firing.setdefault(grp, set()).add(ident)
+                breach = durations.get(_detector_key(labels), 0.0)
+                group_max_duration[grp] = max(group_max_duration.get(grp, 0.0), breach)
+            axis = labels.get("cohort_label")
+            if axis is not None:
+                key = cohort_key_for(labels, axis)
+                cid = (grp, axis, key)
+                cohort_total.setdefault(cid, set()).add(ident)
+                if firing_now:
+                    cohort_firing.setdefault(cid, set()).add(ident)
+                    breach = durations.get(_detector_key(labels), 0.0)
+                    cohort_max_duration[cid] = max(cohort_max_duration.get(cid, 0.0), breach)
+
+    rows: list[BlastRadiusRow] = []
+
+    groups = set(group_total) | set(group_active) | set(group_density)
+    for grp in groups:
+        warming = group_warming.get(grp, set())
+        firing = len(group_firing.get(grp, set()) - warming)
+        total = len((group_total.get(grp, set())) - warming)
+        # Prefer the runner's already-emitted rollup when it is present —
+        # that is the "built on the density rollup" intent.
+        if grp in group_active:
+            firing = round(group_active[grp])
+        density = group_density.get(grp)
+        if density is not None and density > 0:
+            total = round(group_active.get(grp, float(firing)) / density)
+        fraction = firing / total if total else 0.0
+        rows.append(
+            BlastRadiusRow(
+                scope="group",
+                group=grp,
+                firing=firing,
+                total=total,
+                fraction=fraction,
+                max_co_firing_seconds=group_max_duration.get(grp, 0.0),
+            )
+        )
+
+    for cid in cohort_total:
+        grp, axis, key = cid
+        warming = cohort_warming.get(cid, set())
+        firing = len(cohort_firing.get(cid, set()) - warming)
+        total = len(cohort_total[cid] - warming)
+        fraction = firing / total if total else 0.0
+        rows.append(
+            BlastRadiusRow(
+                scope="cohort",
+                group=grp,
+                cohort_label=axis,
+                cohort_key=dict(key),
+                firing=firing,
+                total=total,
+                fraction=fraction,
+                max_co_firing_seconds=cohort_max_duration.get(cid, 0.0),
+            )
+        )
+
+    rows.sort(
+        key=lambda r: (
+            -r.fraction,
+            -r.firing,
+            r.group,
+            tuple(sorted((r.cohort_key or {}).items())),
+        )
+    )
+    return rows
+
+
+__all__ = ["BlastRadiusRow", "FiringSeries", "collect_blast_radius", "collect_firing"]

detector/src/promanomaly/cli/__init__.py

@@ -80,6 +80,7 @@
 from ._probe import (
     stratified_summary_for_query as _stratified_summary_for_query,
 )
+from ._timeline import format_timeline_text, run_timeline
 from ._top import _print_top_text
 
 
@@ -584,6 +585,125 @@ async def _fetch() -> list[QuerySeries]:
         _print_analyze_text(findings)
 
 
+@cli.command(name="timeline")
+@click.option(
+    "--from",
+    "from_",
+    required=True,
+    help=(
+        "Range start. RFC3339 (e.g. 2024-01-15T00:00:00Z), unix seconds, "
+        "or a relative '-1h' / '-6h' offset from --to."
+    ),
+)
+@click.option(
+    "--to",
+    "to_",
+    default="now",
+    show_default=True,
+    help="Range end. RFC3339, unix seconds, or 'now' (default).",
+)
+@click.option(
+    "--group",
+    "groups",
+    multiple=True,
+    help="Restrict to these groups (repeatable).",
+)
+@click.option(
+    "--datasource-url",
+    default="http://localhost:8428/",
+    show_default=True,
+    help="PromQL-compatible datasource URL.",
+)
+@click.option(
+    "--step",
+    default="15s",
+    show_default=True,
+    help="Sample step for the range queries.",
+)
+@click.option(
+    "--output",
+    type=click.Choice(["text", "json"]),
+    default="text",
+    show_default=True,
+    help="Output format. 'json' is suitable for machine consumption.",
+)
+@click.option(
+    "--timeout",
+    default="30s",
+    show_default=True,
+    help="Datasource HTTP timeout.",
+)
+def timeline(
+    from_: str,
+    to_: str,
+    groups: tuple[str, ...],
+    datasource_url: str,
+    step: str,
+    output: str,
+    timeout: str,
+) -> None:
+    """Post-incident anomaly timeline over a historical window.
+
+    Queries the TSDB for ``anomaly_outside_threshold``,
+    ``anomaly_change_point_total``, and ``anomaly_severity`` history and
+    prints a time-ordered list of anomaly events — every firing start
+    and every change-point, with severity attached. Complements
+    ``analyze`` (raw-signal change-points) by operating on the *emitted
+    anomaly metrics*.
+    """
+    from ._timeline import METRIC_CHANGE_POINT, METRIC_OUTSIDE, METRIC_SEVERITY
+
+    try:
+        to_seconds = _parse_analyze_timestamp(to_, now=time.time())
+        from_seconds = _parse_analyze_timestamp(from_, now=to_seconds)
+    except ValueError as exc:
+        raise click.UsageError(str(exc)) from exc
+    if from_seconds >= to_seconds:
+        raise click.UsageError("--from must be before --to")
+
+    step_seconds = parse_duration(step)
+    window_seconds = to_seconds - from_seconds
+
+    # Build an optional group label matcher so only the requested
+    # groups' metrics are fetched.
+    group_filter = ""
+    if groups:
+        group_filter = '{group=~"' + "|".join(groups) + '"}'
+
+    metrics = [METRIC_OUTSIDE, METRIC_CHANGE_POINT, METRIC_SEVERITY]
+
+    async def _fetch() -> dict[str, list[QuerySeries]]:
+        source = PromQLSource(datasource_url, parse_duration(timeout))
+        result: dict[str, list[QuerySeries]] = {}
+        try:
+            await source.start()
+            for metric in metrics:
+                promql = f"{metric}{group_filter}"
+                qr = await source.range_query(
+                    promql=promql,
+                    end=to_seconds,
+                    window_seconds=window_seconds,
+                    step_seconds=step_seconds,
+                )
+                result[metric] = list(qr.series)
+        finally:
+            await source.close()
+        return result
+
+    try:
+        series_by_metric = asyncio.run(_fetch())
+    except SourceQueryError as exc:
+        click.echo(f"query failed: {exc}", err=True)
+        sys.exit(2)
+
+    events = run_timeline(series_by_metric)
+
+    if output == "json":
+        click.echo(json.dumps(events, indent=2, sort_keys=True))
+    else:
+        click.echo(format_timeline_text(events))
+
+
 @cli.command(name="calibrate-buckets")
 @click.option("--query", "promql", required=True, help="PromQL query to analyse.")
 @click.option(
@@ -1156,7 +1276,9 @@ def main() -> None:
     "_safe_float",
     "_stratified_summary_for_query",
     "cli",
+    "format_timeline_text",
     "main",
+    "run_timeline",
 ]