diff --git a/LABELS_CONTRACT.md b/LABELS_CONTRACT.md index 5677406..bdceed8 100644 --- a/LABELS_CONTRACT.md +++ b/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. diff --git a/detector/src/promanomaly/anomalies.py b/detector/src/promanomaly/anomalies.py index 1df4753..b494aaa 100644 --- a/detector/src/promanomaly/anomalies.py +++ b/detector/src/promanomaly/anomalies.py @@ -18,6 +18,8 @@ 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 @@ -25,6 +27,11 @@ # 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"] diff --git a/detector/src/promanomaly/cli/__init__.py b/detector/src/promanomaly/cli/__init__.py index 8a6d6bb..983c606 100644 --- a/detector/src/promanomaly/cli/__init__.py +++ b/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", ] diff --git a/detector/src/promanomaly/cli/_cost.py b/detector/src/promanomaly/cli/_cost.py index daecf51..ee3dc6f 100644 --- a/detector/src/promanomaly/cli/_cost.py +++ b/detector/src/promanomaly/cli/_cost.py @@ -83,6 +83,7 @@ class GroupCostEstimate: short_queries_per_refresh: int = 0 discovery_probe_queries_per_refresh: int = 0 stratified_baseline_queries_per_day: float = 0.0 + recurrence_queries_per_day: float = 0.0 memory_bytes: float = 0.0 cpu_cores: float = 0.0 queries: list[QueryCostEstimate] = field(default_factory=list) @@ -192,6 +193,12 @@ def estimate_cost(cfg: Config) -> list[GroupCostEstimate]: gest.stratified_baseline_queries_per_day += qest.stratified_baseline_queries_per_day gest.memory_bytes += qest.memory_bytes gest.cpu_cores += qest.cpu_cores + # Recurrence analysis: one TSDB range query per refresh_interval + # (not per group refresh), amortised to queries-per-day. + if group.recurrence.enabled: + recurrence_interval = group.recurrence.refresh_interval_seconds + if recurrence_interval > 0: + gest.recurrence_queries_per_day = round(86400.0 / recurrence_interval, 2) groups.append(gest) return groups @@ -210,6 +217,7 @@ def run_estimate_cost(cfg: Config, *, strict: bool) -> int: short_total = 0 probe_total = 0 stratified_total = 0.0 + recurrence_total = 0.0 has_discovery = False for gest in groups: projected_total += gest.projected_series @@ -218,6 +226,7 @@ def run_estimate_cost(cfg: Config, *, strict: bool) -> int: short_total += gest.short_queries_per_refresh probe_total += gest.discovery_probe_queries_per_refresh stratified_total += gest.stratified_baseline_queries_per_day + recurrence_total += gest.recurrence_queries_per_day group_has_discovery = any(q.is_discovery for q in gest.queries) has_discovery = has_discovery or group_has_discovery entry: dict[str, Any] = { @@ -229,6 +238,7 @@ def run_estimate_cost(cfg: Config, *, strict: bool) -> int: "stratified_baseline_queries_per_day": round( gest.stratified_baseline_queries_per_day, 2 ), + "recurrence_queries_per_day": gest.recurrence_queries_per_day, "estimated_memory_bytes": int(gest.memory_bytes), "estimated_cpu_cores": round(gest.cpu_cores, 3), } @@ -249,6 +259,7 @@ def run_estimate_cost(cfg: Config, *, strict: bool) -> int: "short_queries_per_refresh": short_total, "discovery_probe_queries_per_refresh": probe_total, "stratified_baseline_queries_per_day": round(stratified_total, 2), + "recurrence_queries_per_day": round(recurrence_total, 2), "estimated_memory_bytes": int(total_memory), "estimated_cpu_cores": round(total_cpu, 3), "refresh_interval_seconds": cfg.server.refresh_interval_seconds, diff --git a/detector/src/promanomaly/cli/_timeline.py b/detector/src/promanomaly/cli/_timeline.py new file mode 100644 index 0000000..51b01a8 --- /dev/null +++ b/detector/src/promanomaly/cli/_timeline.py @@ -0,0 +1,190 @@ +"""Helpers for the ``timeline`` CLI command. + +Pure functions that turn fetched anomaly metrics into a time-ordered list of +anomaly events, kept free of I/O so they can be unit-tested directly. + +The timeline reads three emitted anomaly metrics over a window: + +* ``anomaly_outside_threshold`` — a 0/1 gauge; each 0->1 transition per series + marks the start of an anomaly. +* ``anomaly_change_point_total`` — a counter; each positive increase between + consecutive samples marks a change-point firing. +* ``anomaly_severity`` — a gauge whose nearest sample at or before an event is + attached to that event when available. +""" + +from __future__ import annotations + +from typing import Any + +from ..source import QuerySeries +from ._analyze import _format_iso + +__all__ = ["format_timeline_text", "run_timeline"] + +# Metric names the timeline understands. +METRIC_OUTSIDE = "anomaly_outside_threshold" +METRIC_CHANGE_POINT = "anomaly_change_point_total" +METRIC_SEVERITY = "anomaly_severity" + +# Labels copied onto every event, in display order. +_EVENT_LABELS = ("id", "group", "detector", "detector_instance") + + +def _identity_key(labels: dict[str, str]) -> tuple[tuple[str, str], ...]: + """Build a hashable key for matching series across metrics. + + Severity is matched to anomaly/change-point events by the shared + identifying labels rather than by every label, so a severity series with + extra labels still lines up with its event source. + """ + return tuple(sorted((k, labels[k]) for k in _EVENT_LABELS if k in labels)) + + +def _event_labels(labels: dict[str, str]) -> dict[str, str]: + """Extract the contracted event labels present on a series.""" + return {k: labels[k] for k in _EVENT_LABELS if k in labels} + + +def _samples(series: QuerySeries) -> list[tuple[float, float]]: + """Return ``(timestamp, value)`` pairs sorted by timestamp. + + Rows whose timestamp or value is missing are skipped so a gap in the + underlying data cannot manufacture a spurious transition. + """ + df = series.samples + pairs: list[tuple[float, float]] = [] + for ts, y in zip(df["timestamp"], df["y"], strict=True): + if ts is None or y is None: + continue + try: + fts = float(ts) + fy = float(y) + except (TypeError, ValueError): + continue + if fts != fts or fy != fy: # skip NaN + continue + pairs.append((fts, fy)) + pairs.sort(key=lambda p: p[0]) + return pairs + + +class _SeverityLookup: + """Nearest-at-or-before severity lookup keyed by identifying labels.""" + + def __init__(self, series_list: list[QuerySeries]) -> None: + self._by_key: dict[tuple[tuple[str, str], ...], list[tuple[float, float]]] = {} + for series in series_list: + key = _identity_key(series.labels) + samples = _samples(series) + if not samples: + continue + self._by_key.setdefault(key, []).extend(samples) + for samples in self._by_key.values(): + samples.sort(key=lambda p: p[0]) + + def at(self, labels: dict[str, str], when: float) -> float | None: + """Return the severity sampled at or just before ``when``, if any.""" + samples = self._by_key.get(_identity_key(labels)) + if not samples: + return None + best: float | None = None + for ts, value in samples: + if ts <= when: + best = value + else: + break + return best + + +def _anomaly_events( + series_list: list[QuerySeries], + severity: _SeverityLookup, +) -> list[dict[str, Any]]: + """Emit an event at each 0->1 transition of the outside-threshold gauge.""" + events: list[dict[str, Any]] = [] + for series in series_list: + labels = _event_labels(series.labels) + previous: float | None = None + for ts, value in _samples(series): + crossed = value >= 0.5 + if crossed and (previous is None or previous < 0.5): + events.append(_make_event("anomaly", ts, labels, severity)) + previous = value + return events + + +def _change_point_events( + series_list: list[QuerySeries], + severity: _SeverityLookup, +) -> list[dict[str, Any]]: + """Emit an event at each positive increase of the change-point counter.""" + events: list[dict[str, Any]] = [] + for series in series_list: + labels = _event_labels(series.labels) + previous: float | None = None + for ts, value in _samples(series): + if previous is not None and value > previous: + events.append(_make_event("change_point", ts, labels, severity)) + previous = value + return events + + +def _make_event( + kind: str, + when: float, + labels: dict[str, str], + severity: _SeverityLookup, +) -> dict[str, Any]: + """Assemble a single timeline event, attaching severity when available.""" + event: dict[str, Any] = { + "time": when, + "time_rfc3339": _format_iso(when), + "kind": kind, + "id": labels.get("id", ""), + "group": labels.get("group", ""), + "detector": labels.get("detector", ""), + "detector_instance": labels.get("detector_instance", ""), + } + value = severity.at(labels, when) + if value is not None: + event["severity"] = value + return event + + +def run_timeline(series_by_metric: dict[str, list[QuerySeries]]) -> list[dict[str, Any]]: + """Compute a time-ordered list of anomaly events from fetched series. + + Args: + series_by_metric: Fetched series keyed by metric name. Recognised keys + are ``anomaly_outside_threshold``, ``anomaly_change_point_total``, + and ``anomaly_severity``; missing keys are treated as empty. + + Returns: + Events sorted by ascending time. Ties break by kind then ``id`` so the + ordering is deterministic. + """ + severity = _SeverityLookup(series_by_metric.get(METRIC_SEVERITY, [])) + events: list[dict[str, Any]] = [] + events.extend(_anomaly_events(series_by_metric.get(METRIC_OUTSIDE, []), severity)) + events.extend(_change_point_events(series_by_metric.get(METRIC_CHANGE_POINT, []), severity)) + events.sort(key=lambda e: (e["time"], e["kind"], e["id"])) + return events + + +def format_timeline_text(events: list[dict[str, Any]]) -> str: + """Render timeline events as a human-readable table.""" + if not events: + return "No anomaly events in the requested window." + lines = [f"{'time':<24} {'kind':<12} {'severity':>8} labels"] + for event in events: + labels = ", ".join( + f"{key}={event[key]}" + for key in ("id", "group", "detector", "detector_instance") + if event.get(key) + ) + severity = event.get("severity") + severity_text = f"{severity:>8.2f}" if isinstance(severity, (int, float)) else f"{'-':>8}" + ts = event["time_rfc3339"] + lines.append(f"{ts:<24} {event['kind']:<12} {severity_text} {labels}") + return "\n".join(lines) diff --git a/detector/src/promanomaly/cli/_top.py b/detector/src/promanomaly/cli/_top.py index aa38148..17dd746 100644 --- a/detector/src/promanomaly/cli/_top.py +++ b/detector/src/promanomaly/cli/_top.py @@ -35,11 +35,39 @@ def _print_lint_hints(payload: dict[str, Any]) -> None: click.echo("") +def _print_blast_radius(payload: dict[str, Any]) -> None: + rows = payload.get("blast_radius") or [] + if not rows: + return + click.echo("") + click.echo("blast radius:") + header = ( + f" {'SCOPE':<8} {'GROUP':<16} {'FIRING':>7}" + f" {'TOTAL':>7} {'FRAC':>6} {'MAX DUR':>7} COHORT" + ) + click.echo(header) + for r in rows: + cohort = "" + if r.get("cohort_label"): + key_repr = ",".join(f"{k}={v}" for k, v in sorted((r.get("cohort_key") or {}).items())) + cohort = f"{r['cohort_label']} {{{key_repr}}}" if key_repr else r["cohort_label"] + click.echo( + f" {r.get('scope', ''):<8} " + f"{r.get('group', ''):<16} " + f"{r.get('firing', 0):>7} " + f"{r.get('total', 0):>7} " + f"{r.get('fraction', 0.0):>5.0%} " + f"{_format_duration(r.get('max_co_firing_seconds', 0.0)):>7} " + f"{cohort}" + ) + + def _print_top_text(payload: dict[str, Any]) -> None: _print_lint_hints(payload) anomalies = payload.get("anomalies", []) if not anomalies: click.echo("no firing anomalies") + _print_blast_radius(payload) return header = f"{'SEV':>5} {'SCORE':>8} {'DUR':>6} {'TYPE':<16} {'GROUP':<16} ID / LABELS" click.echo(header) @@ -63,6 +91,7 @@ def _print_top_text(payload: dict[str, Any]) -> None: f"{a.get('group', ''):<16} " f"{id_repr}" ) + _print_blast_radius(payload) __all__ = ["_print_top_text"] diff --git a/detector/src/promanomaly/config.py b/detector/src/promanomaly/config.py index dafcea9..fce7956 100644 --- a/detector/src/promanomaly/config.py +++ b/detector/src/promanomaly/config.py @@ -774,6 +774,35 @@ class EnsembleConfig(_ModelBase): AutoSelect = bool | Literal["explicit_plus_best"] +class RecurrenceConfig(_ModelBase): + """Opt-in recurrence-pattern analysis for a group. + + When enabled, the runner queries the TSDB for historical + ``anomaly_outside_threshold`` and emits a 7x24 day-of-week x + hour-of-day firing-rate heatmap as ``anomaly_recurrence_score``. + The analysis runs at most once per ``refresh_interval`` (default + ``1h``) — not on every group refresh — because the 4-week lookback + makes frequent re-queries wasteful. + """ + + enabled: bool = False + lookback: Duration = "4w" + step: Duration = "1h" + refresh_interval: Duration = "1h" + + @property + def lookback_seconds(self) -> float: + return parse_duration(self.lookback) + + @property + def step_seconds(self) -> float: + return parse_duration(self.step) + + @property + def refresh_interval_seconds(self) -> float: + return parse_duration(self.refresh_interval) + + class GroupConfig(_ModelBase): name: PromName priority: int = 1 @@ -781,6 +810,7 @@ class GroupConfig(_ModelBase): ensemble: EnsembleConfig | None = None auto_select: AutoSelect = False auto_select_interval: Duration = "24h" + recurrence: RecurrenceConfig = Field(default_factory=RecurrenceConfig) @property def auto_select_interval_seconds(self) -> float: diff --git a/detector/src/promanomaly/exporter.py b/detector/src/promanomaly/exporter.py index f34858a..8a820b7 100644 --- a/detector/src/promanomaly/exporter.py +++ b/detector/src/promanomaly/exporter.py @@ -200,6 +200,15 @@ def validate_label_name(name: str) -> str: "per-detector metrics down to the winner." ), ), + "anomaly_recurrence_score": ( + "gauge", + ( + "Stratified firing-rate of anomaly_outside_threshold by " + "day_of_week x hour_of_day over a configurable lookback. " + "A high value at (Tue, 9) means this series fires often on " + "Tuesdays at 09:00 UTC. Opt-in per group via recurrence.enabled." + ), + ), } diff --git a/detector/src/promanomaly/recurrence.py b/detector/src/promanomaly/recurrence.py new file mode 100644 index 0000000..633ad9c --- /dev/null +++ b/detector/src/promanomaly/recurrence.py @@ -0,0 +1,126 @@ +"""Anomaly recurrence pattern detection. + +Computes a stratified firing-rate over day-of-week x hour-of-day buckets +by querying the detector's own emitted ``anomaly_outside_threshold`` +history from the TSDB. The result is a 7x24 heatmap of how often a +given ``(id, group)`` series has been anomalous at each (day, hour) +combination — an answer to "is this a recurring pattern?" without any +persisted state. + +The analysis is opt-in per group via ``recurrence.enabled: true`` in the +group config. +""" + +from __future__ import annotations + +import datetime +import math +from typing import TYPE_CHECKING + +from .state.snapshot import Sample + +if TYPE_CHECKING: + from .source import QuerySeries + +# Day-of-week names used as label values (ISO 8601: Monday=0). +_DOW_NAMES = ("Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun") + + +def compute_recurrence_samples( + series_list: list[QuerySeries], + *, + group: str, +) -> list[Sample]: + """Turn historical outside-threshold series into recurrence-score samples. + + For each distinct ``(id, group)`` identity in *series_list*, the + rolling-window samples are bucketed by (day_of_week, hour_of_day) in + UTC, and the firing rate (fraction of samples where value >= 0.5) per + bucket is emitted as ``anomaly_recurrence_score``. + + When multiple detectors score the same ``id``, the TSDB returns one + ``anomaly_outside_threshold`` series per detector. This function + deduplicates by ``(id, timestamp)`` — at each point in time, it takes + the max value across detectors (any-detector-fires semantics), so the + recurrence rate reflects how often the *series* was anomalous, not how + many detectors agreed. + + Only buckets with at least one observation are emitted so cardinality + stays bounded at ``<= 168 * number_of_ids``. + + Args: + series_list: Historical ``anomaly_outside_threshold`` series from + the TSDB. The caller should use ``max by (id, group)`` in the + PromQL query when possible; this function handles the residual + case where the raw per-detector series arrive. + group: The group name (used to match the ``group`` label). + + Returns: + A list of :class:`Sample` objects, one per non-empty bucket. + """ + # Step 1: Deduplicate across detectors. For each (id, timestamp), + # take the max value so that if *any* detector fires, the point counts + # as firing — matching the ``max`` ensemble semantics. + # + # Key: (series_id, timestamp) -> max(value) + deduped: dict[tuple[str, float], float] = {} + + for series in series_list: + series_id = series.labels.get("id", "") + series_group = series.labels.get("group", "") + if series_group != group: + continue + + df = series.samples + for ts_val, y_val in zip(df["timestamp"], df["y"], strict=True): + try: + ts = float(ts_val) + y = float(y_val) + except (TypeError, ValueError): + continue + if math.isnan(ts) or math.isnan(y): + continue + + point_key = (series_id, ts) + prev = deduped.get(point_key) + if prev is None or y > prev: + deduped[point_key] = y + + # Step 2: Bucket by (id, dow, hour) and compute firing rates. + BucketKey = tuple[str, int, int] # (id, dow, hour) + firing: dict[BucketKey, int] = {} + total: dict[BucketKey, int] = {} + + for (series_id, ts), value in deduped.items(): + dt = datetime.datetime.fromtimestamp(ts, tz=datetime.UTC) + dow = dt.weekday() # Monday=0 + hour = dt.hour + + key: BucketKey = (series_id, dow, hour) + total[key] = total.get(key, 0) + 1 + if value >= 0.5: + firing[key] = firing.get(key, 0) + 1 + + out: list[Sample] = [] + for key in sorted(total): + series_id, dow, hour = key + t = total[key] + if t == 0: + continue + rate = firing.get(key, 0) / t + labels = tuple( + sorted( + { + "id": series_id, + "group": group, + "day_of_week": _DOW_NAMES[dow], + "hour_of_day": str(hour), + }.items() + ) + ) + out.append(Sample(metric="anomaly_recurrence_score", labels=labels, value=rate)) + + return out + + +__all__ = ["compute_recurrence_samples"] diff --git a/detector/src/promanomaly/runner.py b/detector/src/promanomaly/runner.py index 1a9b5bb..da0ae16 100644 --- a/detector/src/promanomaly/runner.py +++ b/detector/src/promanomaly/runner.py @@ -63,6 +63,7 @@ effective_threshold, merge_params, ) +from .recurrence import compute_recurrence_samples from .selector import select_winner from .self_observability import estimate_samples_bytes from .severity import compute_severity @@ -248,6 +249,12 @@ def __init__( # this facade; the runner delegates to it rather than owning # the heavy cache state directly. self._stratified = StratifiedFetcher(source=source, config=config) + # Tracks the last time recurrence analysis ran per group so it + # doesn't execute on every group refresh (the 4-week lookback + # makes frequent re-queries wasteful). Between refreshes, the + # last-good samples are carried forward via _recurrence_cache. + self._recurrence_last_run: dict[str, float] = {} + self._recurrence_cache: dict[str, list[Sample]] = {} # Surface stratified-config issues once at boot — but only on # single-replica deployments. In HA mode every follower would # otherwise log the same warning, multiplying the noise by the @@ -306,6 +313,8 @@ def replace_config(self, new_config: Config) -> None: for removed in old_names - new_names: self._store.remove_group(removed) self._group_locks.pop(removed, None) + self._recurrence_cache.pop(removed, None) + self._recurrence_last_run.pop(removed, None) for added in new_names - old_names: self._group_locks[added] = asyncio.Lock() # Bound the rolling detector-success ledger to live groups so a @@ -486,6 +495,30 @@ async def _run_group_inner(self, group: GroupConfig) -> _GroupInnerResult: # other snapshot metric. out.samples.extend(self._density_samples(group, out.samples)) + # Recurrence pattern analysis. Opt-in per group; queries the + # TSDB for historical anomaly_outside_threshold and emits a 7x24 + # firing-rate heatmap. Time-gated by recurrence.refresh_interval + # (default 1h) so the 4-week lookback query doesn't fire on every + # 1-minute group refresh. Between refreshes, the last-good + # recurrence samples are carried forward in the snapshot via + # _recurrence_cache. + if group.recurrence.enabled: + now = time.time() + last = self._recurrence_last_run.get(group.name, 0.0) + interval = group.recurrence.refresh_interval_seconds + if now - last >= interval: + try: + recurrence_samples = await self._recurrence_samples(group) + self._recurrence_cache[group.name] = recurrence_samples + self._recurrence_last_run[group.name] = now + except Exception as exc: + logger.warning( + "recurrence_analysis_failed", + group=group.name, + error=str(exc), + ) + out.samples.extend(self._recurrence_cache.get(group.name, [])) + return out async def _process_discovered_query( @@ -1881,6 +1914,23 @@ def _series_key(label_dict: dict[str, str]) -> tuple[tuple[str, str], ...]: ) return out + async def _recurrence_samples(self, group: GroupConfig) -> list[Sample]: + """Query TSDB for historical outside-threshold and compute recurrence scores.""" + # ``max by (id, group)`` collapses the per-detector series so the + # TSDB returns one value per (id, group, timestamp) — avoiding the + # multi-detector inflation where N detectors scoring the same id + # would otherwise contribute N observations per time step. + promql = f'max by (id, group) (anomaly_outside_threshold{{group="{group.name}"}})' + result = await self._source.range_query( + promql=promql, + end=time.time(), + window_seconds=group.recurrence.lookback_seconds, + step_seconds=group.recurrence.step_seconds, + ) + if not result.series: + return [] + return compute_recurrence_samples(result.series, group=group.name) + def _update_operational_metrics(self, result: GroupRunResult) -> None: if result.succeeded: self._ops.last_run_timestamp_seconds.labels(group=result.group).set(time.time()) diff --git a/detector/src/promanomaly/server.py b/detector/src/promanomaly/server.py index 4e84451..143f0a3 100644 --- a/detector/src/promanomaly/server.py +++ b/detector/src/promanomaly/server.py @@ -15,7 +15,7 @@ from fastapi import FastAPI, HTTPException, Request, Response from fastapi.responses import JSONResponse, PlainTextResponse -from .anomalies import collect_firing +from .anomalies import collect_blast_radius, collect_firing from .config import Config, ReloadAuthConfig from .detectors import list_meta from .inspect import InspectError, inspect_series @@ -104,11 +104,15 @@ async def debug_anomalies( min_severity=min_severity, limit=max(0, limit), ) + blast = collect_blast_radius(application._store) + if group is not None: + blast = [r for r in blast if r.group == group] body: dict[str, Any] = { "count": len(firing), "min_severity": min_severity, "group": group, "anomalies": [f.as_dict() for f in firing], + "blast_radius": [r.as_dict() for r in blast], } # Opt-in metadata lint (off by default so the no-extra-TSDB-query # contract holds for the common scrape path). ``promanomaly top`` diff --git a/detector/tests/conftest.py b/detector/tests/conftest.py index 3bd24e6..808e4e1 100644 --- a/detector/tests/conftest.py +++ b/detector/tests/conftest.py @@ -8,6 +8,7 @@ import pandas as pd import pytest +from click.testing import CliRunner from promanomaly.source import PromQLSource, QueryResult, QuerySeries @@ -18,6 +19,49 @@ class StubResponse: error: Exception | None = None +class _QueryResult: + """Lightweight stand-in for :class:`QueryResult` used by CLI stubs.""" + + def __init__(self, series: list[QuerySeries]) -> None: + self.series = series + self.truncated = False + + +class _CLIStubSource: + """Minimal async-compatible source stub for CLI command tests. + + The CLI commands construct a :class:`PromQLSource` internally, so + tests monkeypatch the ``PromQLSource`` symbol in the CLI module with + a factory that returns this stub. The ``calls`` list records every + range query the CLI issued so assertions can inspect the PromQL. + """ + + def __init__(self, responder: Any) -> None: + self._responder = responder + self.calls: list[dict[str, Any]] = [] + + async def start(self) -> None: + return None + + async def close(self) -> None: + return None + + async def range_query( + self, + promql: str, + end: float, + window_seconds: float, + step_seconds: float, + max_series: int | None = None, + ) -> Any: + self.calls.append( + {"promql": promql, "end": end, "window": window_seconds, "step": step_seconds} + ) + return self._responder( + promql=promql, end=end, window_seconds=window_seconds, step_seconds=step_seconds + ) + + class StubSource(PromQLSource): """Drop-in replacement for the live PromQL client used in unit tests.""" @@ -77,5 +121,24 @@ def stub_source() -> StubSource: return StubSource() +@pytest.fixture +def cli_stub_source() -> Any: + """Return a callable ``(monkeypatch, responder)`` that patches the CLI + module's ``PromQLSource`` and returns a stub with ``.calls``.""" + import promanomaly.cli as cli_module + + def _factory(monkeypatch: pytest.MonkeyPatch, responder: Any) -> _CLIStubSource: + stub = _CLIStubSource(responder) + monkeypatch.setattr(cli_module, "PromQLSource", lambda *_a, **_kw: stub) + return stub + + return _factory + + +@pytest.fixture +def cli_runner() -> CliRunner: + return CliRunner() + + def make_series(labels: dict[str, str], samples: pd.DataFrame) -> QuerySeries: return QuerySeries(labels=labels, samples=samples) diff --git a/detector/tests/test_blast_radius.py b/detector/tests/test_blast_radius.py new file mode 100644 index 0000000..c435fcc --- /dev/null +++ b/detector/tests/test_blast_radius.py @@ -0,0 +1,210 @@ +"""Tests for the blast-radius rollup — collect_blast_radius + /debug/anomalies.""" + +from __future__ import annotations + +import time +from pathlib import Path +from typing import Any +from unittest.mock import patch + +from fastapi.testclient import TestClient + +from promanomaly.anomalies import collect_blast_radius +from promanomaly.config import CURRENT_API_VERSION, Config +from promanomaly.main import Application +from promanomaly.state import GroupSnapshot, Sample, SnapshotStore +from tests.conftest import StubSource + + +def _store_with(group: str, samples: list[Sample]) -> SnapshotStore: + store = SnapshotStore() + store.write(GroupSnapshot(group=group, timestamp=time.time(), samples=samples)) + return store + + +def _sample(metric: str, value: float, **labels: str) -> Sample: + return Sample(metric=metric, value=value, labels=tuple(sorted(labels.items()))) + + +def _row(rows: list[Any], scope: str, **match: str) -> Any: + for row in rows: + if row.scope != scope: + continue + if all(getattr(row, k, None) == v for k, v in match.items()): + return row + raise AssertionError(f"no {scope} row matching {match} in {[r.as_dict() for r in rows]}") + + +def test_group_rollup_firing_total_fraction_and_duration() -> None: + # Three members; two firing, one clear. One firing member has been + # breached for 120s, the other 40s, so the max is 120. + samples = [ + _sample("anomaly_outside_threshold", 1.0, id="a", group="g", detector="MAD"), + _sample("anomaly_duration_seconds", 120.0, id="a", group="g", detector="MAD"), + _sample("anomaly_outside_threshold", 1.0, id="b", group="g", detector="MAD"), + _sample("anomaly_duration_seconds", 40.0, id="b", group="g", detector="MAD"), + _sample("anomaly_outside_threshold", 0.0, id="c", group="g", detector="MAD"), + _sample("anomaly_duration_seconds", 0.0, id="c", group="g", detector="MAD"), + ] + rows = collect_blast_radius(_store_with("g", samples)) + row = _row(rows, "group", group="g") + assert row.firing == 2 + assert row.total == 3 + assert row.fraction == 2 / 3 + assert row.max_co_firing_seconds == 120.0 + + +def test_group_rollup_prefers_density_samples_when_present() -> None: + # The runner-emitted rollup is authoritative: active=2 over density 0.5 + # recovers a total of 4, even though only one outside_threshold sample + # is in the snapshot. + samples = [ + _sample("anomaly_active_series", 2.0, group="g"), + _sample("anomaly_density", 0.5, group="g"), + _sample("anomaly_outside_threshold", 1.0, id="a", group="g", detector="MAD"), + ] + rows = collect_blast_radius(_store_with("g", samples)) + row = _row(rows, "group", group="g") + assert row.firing == 2 + assert row.total == 4 + assert row.fraction == 0.5 + + +def test_warming_up_members_excluded() -> None: + samples = [ + _sample("anomaly_outside_threshold", 1.0, id="a", group="g", detector="MAD"), + _sample("anomaly_outside_threshold", 0.0, id="b", group="g", detector="MAD"), + _sample("anomaly_warming_up", 1.0, id="b", group="g", detector="MAD"), + ] + rows = collect_blast_radius(_store_with("g", samples)) + row = _row(rows, "group", group="g") + assert row.firing == 1 + assert row.total == 1 + assert row.fraction == 1.0 + + +def test_cohort_rollup_groups_by_cohort_identity() -> None: + # Cohort axis is "instance": three nodes in the same cohort (same job), + # two firing. cohort_key is the shared non-axis labels. + common = {"group": "g", "detector": "Cohort", "cohort_label": "instance", "job": "web"} + samples = [ + _sample("anomaly_outside_threshold", 1.0, id="x", instance="n1", **common), + _sample("anomaly_duration_seconds", 200.0, id="x", instance="n1", **common), + _sample("anomaly_outside_threshold", 1.0, id="x", instance="n2", **common), + _sample("anomaly_duration_seconds", 90.0, id="x", instance="n2", **common), + _sample("anomaly_outside_threshold", 0.0, id="x", instance="n3", **common), + ] + rows = collect_blast_radius(_store_with("g", samples)) + cohort = _row(rows, "cohort", group="g", cohort_label="instance") + assert cohort.firing == 2 + assert cohort.total == 3 + assert cohort.fraction == 2 / 3 + assert cohort.max_co_firing_seconds == 200.0 + # cohort_key excludes the axis label and the meta-labels. + assert "instance" not in cohort.cohort_key + assert cohort.cohort_key.get("job") == "web" + assert cohort.cohort_key.get("id") == "x" + + +def test_rows_sorted_by_fraction_then_firing() -> None: + samples = [ + # group g1: 1 of 4 firing -> fraction 0.25 + _sample("anomaly_outside_threshold", 1.0, id="a", group="g1", detector="MAD"), + _sample("anomaly_outside_threshold", 0.0, id="b", group="g1", detector="MAD"), + _sample("anomaly_outside_threshold", 0.0, id="c", group="g1", detector="MAD"), + _sample("anomaly_outside_threshold", 0.0, id="d", group="g1", detector="MAD"), + ] + store = SnapshotStore() + store.write(GroupSnapshot(group="g1", timestamp=time.time(), samples=samples)) + store.write( + GroupSnapshot( + group="g2", + timestamp=time.time(), + samples=[ + _sample("anomaly_outside_threshold", 1.0, id="a", group="g2", detector="MAD"), + _sample("anomaly_outside_threshold", 1.0, id="b", group="g2", detector="MAD"), + ], + ), + ) + rows = [r for r in collect_blast_radius(store) if r.scope == "group"] + # g2 fully firing (1.0) sorts ahead of g1 (0.25). + assert rows[0].group == "g2" + assert rows[1].group == "g1" + + +def _build_test_app() -> tuple[Any, Application]: + import tempfile + + import yaml + + raw: dict[str, Any] = { + "apiVersion": CURRENT_API_VERSION, + "datasource": {"url": "http://stub"}, + "groups": [ + { + "name": "g", + "queries": [ + {"id": "q", "promql": "up", "detectors": [{"name": "MAD"}]}, + ], + } + ], + } + tmp_path = Path(tempfile.mkdtemp()) + config_path = tmp_path / "config.yaml" + config_path.write_text(yaml.dump(raw)) + stub = StubSource() + with patch("promanomaly.main.PromQLSource", return_value=stub): + application = Application(Config.model_validate(raw), config_path) + return application.build_app(), application + + +def test_debug_anomalies_includes_blast_radius() -> None: + samples = [ + _sample("anomaly_outside_threshold", 1.0, id="a", group="g", detector="MAD"), + _sample("anomaly_severity", 0.8, id="a", group="g", detector="MAD"), + _sample("anomaly_outside_threshold", 0.0, id="b", group="g", detector="MAD"), + ] + store = _store_with("g", samples) + app, application = _build_test_app() + application._store = store + client = TestClient(app) + resp = client.get("/debug/anomalies") + assert resp.status_code == 200 + body = resp.json() + assert "blast_radius" in body + rows = body["blast_radius"] + group_rows = [r for r in rows if r["scope"] == "group"] + assert len(group_rows) == 1 + assert group_rows[0]["group"] == "g" + assert group_rows[0]["firing"] == 1 + assert group_rows[0]["total"] == 2 + assert group_rows[0]["fraction"] == 0.5 + + +def test_debug_anomalies_group_param_filters_blast_radius() -> None: + store = SnapshotStore() + store.write( + GroupSnapshot( + group="g", + timestamp=time.time(), + samples=[ + _sample("anomaly_outside_threshold", 1.0, id="a", group="g", detector="MAD"), + ], + ), + ) + store.write( + GroupSnapshot( + group="other", + timestamp=time.time(), + samples=[ + _sample("anomaly_outside_threshold", 1.0, id="a", group="other", detector="MAD"), + ], + ), + ) + app, application = _build_test_app() + application._store = store + client = TestClient(app) + resp = client.get("/debug/anomalies?group=g") + assert resp.status_code == 200 + rows = resp.json()["blast_radius"] + assert {r["group"] for r in rows} == {"g"} diff --git a/detector/tests/test_cli_timeline.py b/detector/tests/test_cli_timeline.py new file mode 100644 index 0000000..1718b1c --- /dev/null +++ b/detector/tests/test_cli_timeline.py @@ -0,0 +1,193 @@ +"""Tests for the ``timeline`` CLI command.""" + +from __future__ import annotations + +import json +from typing import Any + +import pandas as pd + +from promanomaly.cli import cli +from promanomaly.cli._timeline import ( + METRIC_CHANGE_POINT, + METRIC_OUTSIDE, + METRIC_SEVERITY, + run_timeline, +) +from promanomaly.source import QuerySeries + + +def _series( + values: list[float], + *, + labels: dict[str, str] | None = None, + start: float = 0.0, + step: float = 60.0, +) -> QuerySeries: + df = pd.DataFrame( + {"timestamp": [start + float(i) * step for i in range(len(values))], "y": values} + ) + return QuerySeries(labels=labels or {}, samples=df) + + +def _responder_for(series_by_metric: dict[str, list[QuerySeries]]) -> Any: + """Return a responder branching on the metric name in the query.""" + + def _responder(*, promql: str, end: float, window_seconds: float, step_seconds: float) -> Any: + from tests.conftest import _QueryResult + + for metric, series in series_by_metric.items(): + if promql.startswith(metric): + return _QueryResult(list(series)) + return _QueryResult([]) + + return _responder + + +def test_timeline_text_output(cli_stub_source: Any, cli_runner: Any, monkeypatch: Any) -> None: + series_by_metric = { + METRIC_OUTSIDE: [ + _series( + [0.0, 0.0, 1.0, 1.0], + labels={"id": "errors", "group": "g1", "detector": "MAD"}, + ) + ], + METRIC_CHANGE_POINT: [ + _series( + [0.0, 0.0, 0.0, 1.0], + labels={"id": "errors", "group": "g1", "detector": "CUSUM"}, + ) + ], + } + cli_stub_source(monkeypatch, _responder_for(series_by_metric)) + + result = cli_runner.invoke( + cli, + ["timeline", "--from", "-1h", "--to", "now"], + ) + + assert result.exit_code == 0 + output = result.output + assert "anomaly" in output + assert "change_point" in output + # The anomaly starts at t=120s and the change-point fires at t=180s, so the + # anomaly line must appear before the change-point line. + assert output.index("anomaly") < output.index("change_point") + + +def test_timeline_json_output(cli_stub_source: Any, cli_runner: Any, monkeypatch: Any) -> None: + series_by_metric = { + METRIC_OUTSIDE: [ + _series( + [0.0, 1.0], + labels={ + "id": "errors", + "group": "g1", + "detector": "MAD", + "detector_instance": "short", + }, + ) + ], + METRIC_SEVERITY: [ + _series( + [2.0, 4.5], + labels={ + "id": "errors", + "group": "g1", + "detector": "MAD", + "detector_instance": "short", + }, + ) + ], + } + cli_stub_source(monkeypatch, _responder_for(series_by_metric)) + + result = cli_runner.invoke( + cli, + ["timeline", "--from", "-1h", "--output", "json"], + ) + + assert result.exit_code == 0 + payload = json.loads(result.output) + assert isinstance(payload, list) + assert len(payload) == 1 + event = payload[0] + assert event["kind"] == "anomaly" + assert event["id"] == "errors" + assert event["group"] == "g1" + assert event["detector"] == "MAD" + assert event["detector_instance"] == "short" + assert event["severity"] == 4.5 + assert "time" in event + assert "time_rfc3339" in event + + +def test_timeline_group_filter(cli_stub_source: Any, cli_runner: Any, monkeypatch: Any) -> None: + series_by_metric = { + METRIC_OUTSIDE: [ + _series([0.0, 1.0], labels={"id": "errors", "group": "g1", "detector": "MAD"}) + ], + } + stub = cli_stub_source(monkeypatch, _responder_for(series_by_metric)) + + result = cli_runner.invoke( + cli, + ["timeline", "--from", "-1h", "--group", "g1", "--group", "g2"], + ) + + assert result.exit_code == 0 + # Every issued query carries the repeated-group label matcher. + assert stub.calls + for call in stub.calls: + assert '{group=~"g1|g2"}' in call["promql"] + + +def test_timeline_source_error_exits( + cli_stub_source: Any, cli_runner: Any, monkeypatch: Any +) -> None: + def _responder(*, promql: str, end: float, window_seconds: float, step_seconds: float) -> Any: + from promanomaly.source import SourceQueryError + + raise SourceQueryError("boom", "test error") + + cli_stub_source(monkeypatch, _responder) + + result = cli_runner.invoke( + cli, + ["timeline", "--from", "-1h"], + ) + + assert result.exit_code == 2 + + +def test_timeline_from_after_to(cli_stub_source: Any, cli_runner: Any, monkeypatch: Any) -> None: + cli_stub_source(monkeypatch, _responder_for({})) + + result = cli_runner.invoke( + cli, + ["timeline", "--from", "now", "--to", "-1h"], + ) + + assert result.exit_code != 0 + + +def test_run_timeline_events() -> None: + labels = {"id": "errors", "group": "g1", "detector": "MAD"} + series_by_metric = { + METRIC_OUTSIDE: [_series([0.0, 0.0, 1.0, 1.0, 0.0], labels=labels)], + METRIC_CHANGE_POINT: [_series([0.0, 0.0, 0.0, 2.0, 2.0], labels=labels)], + METRIC_SEVERITY: [_series([1.0, 1.5, 3.0, 3.5, 2.0], labels=labels)], + } + + events = run_timeline(series_by_metric) + + assert len(events) == 2 + anomaly, change_point = events + assert anomaly["kind"] == "anomaly" + assert anomaly["time"] == 120.0 + assert anomaly["severity"] == 3.0 + assert change_point["kind"] == "change_point" + assert change_point["time"] == 180.0 + assert change_point["severity"] == 3.5 + # Sorted ascending by time. + assert anomaly["time"] < change_point["time"] diff --git a/detector/tests/test_recurrence.py b/detector/tests/test_recurrence.py new file mode 100644 index 0000000..f3d6bab --- /dev/null +++ b/detector/tests/test_recurrence.py @@ -0,0 +1,260 @@ +"""Tests for anomaly recurrence pattern detection.""" + +from __future__ import annotations + +import datetime +from typing import Any + +import pandas as pd +import pytest + +from promanomaly.config import CURRENT_API_VERSION, Config +from promanomaly.exporter import OperationalMetrics +from promanomaly.recurrence import compute_recurrence_samples +from promanomaly.runner import Runner +from promanomaly.source import QuerySeries +from promanomaly.state import SnapshotStore +from tests.conftest import StubResponse, StubSource + + +def _outside_series( + firing_timestamps: list[float], + clear_timestamps: list[float], + labels: dict[str, str], +) -> QuerySeries: + """Build an anomaly_outside_threshold series with known firing times.""" + timestamps = sorted(firing_timestamps + clear_timestamps) + values = [1.0 if t in firing_timestamps else 0.0 for t in timestamps] + df = pd.DataFrame({"timestamp": timestamps, "y": values}) + return QuerySeries(labels=labels, samples=df) + + +def _ts(year: int, month: int, day: int, hour: int) -> float: + """UTC timestamp for a given date+hour.""" + return datetime.datetime(year, month, day, hour, tzinfo=datetime.UTC).timestamp() + + +def test_recurrence_basic_buckets() -> None: + """Firing on Tuesday at 10:00 UTC produces a non-zero score for (Tue, 10).""" + # Two Tuesdays: one fires at 10:00, the other is clear at 10:00. + labels = {"id": "errors", "group": "g1"} + tue1_10 = _ts(2025, 5, 6, 10) # Tuesday + tue2_10 = _ts(2025, 5, 13, 10) # next Tuesday + + series = _outside_series( + firing_timestamps=[tue1_10], + clear_timestamps=[tue2_10], + labels=labels, + ) + + samples = compute_recurrence_samples([series], group="g1") + + # Find the Tue/10 bucket. + tue_10 = [ + s + for s in samples + if dict(s.labels).get("day_of_week") == "Tue" and dict(s.labels).get("hour_of_day") == "10" + ] + assert len(tue_10) == 1 + assert tue_10[0].value == 0.5 # 1 firing out of 2 samples + + +def test_recurrence_empty_series_emits_nothing() -> None: + samples = compute_recurrence_samples([], group="g1") + assert samples == [] + + +def test_recurrence_all_clear_emits_zero_scores() -> None: + labels = {"id": "errors", "group": "g1"} + mon_9 = _ts(2025, 5, 5, 9) # Monday + mon_10 = _ts(2025, 5, 5, 10) + + series = _outside_series( + firing_timestamps=[], + clear_timestamps=[mon_9, mon_10], + labels=labels, + ) + samples = compute_recurrence_samples([series], group="g1") + assert all(s.value == 0.0 for s in samples) + + +def test_recurrence_filters_by_group() -> None: + labels = {"id": "errors", "group": "other"} + mon_9 = _ts(2025, 5, 5, 9) + + series = _outside_series( + firing_timestamps=[mon_9], + clear_timestamps=[], + labels=labels, + ) + samples = compute_recurrence_samples([series], group="g1") + assert samples == [] + + +def test_recurrence_multiple_ids() -> None: + labels_a = {"id": "a", "group": "g"} + labels_b = {"id": "b", "group": "g"} + mon_9 = _ts(2025, 5, 5, 9) + + series_a = _outside_series(firing_timestamps=[mon_9], clear_timestamps=[], labels=labels_a) + series_b = _outside_series(firing_timestamps=[], clear_timestamps=[mon_9], labels=labels_b) + + samples = compute_recurrence_samples([series_a, series_b], group="g") + + ids = {dict(s.labels)["id"] for s in samples} + assert "a" in ids + assert "b" in ids + + a_samples = [s for s in samples if dict(s.labels)["id"] == "a"] + assert all(s.value == 1.0 for s in a_samples) + b_samples = [s for s in samples if dict(s.labels)["id"] == "b"] + assert all(s.value == 0.0 for s in b_samples) + + +def test_recurrence_deduplicates_across_detectors() -> None: + """Multiple detectors scoring the same id should not inflate the count.""" + labels_mad = {"id": "errors", "group": "g1", "detector": "MAD"} + labels_hampel = {"id": "errors", "group": "g1", "detector": "Hampel"} + mon_9 = _ts(2025, 5, 5, 9) + + # Both detectors fire at the same timestamp. + series_mad = _outside_series(firing_timestamps=[mon_9], clear_timestamps=[], labels=labels_mad) + series_hampel = _outside_series( + firing_timestamps=[mon_9], clear_timestamps=[], labels=labels_hampel + ) + + samples = compute_recurrence_samples([series_mad, series_hampel], group="g1") + + mon_9_buckets = [ + s + for s in samples + if dict(s.labels).get("day_of_week") == "Mon" and dict(s.labels).get("hour_of_day") == "9" + ] + assert len(mon_9_buckets) == 1 + # One observation, one firing -> rate 1.0, not inflated to 2/2. + assert mon_9_buckets[0].value == 1.0 + + +def test_recurrence_dedup_any_detector_fires() -> None: + """If one detector fires and another doesn't, the point counts as firing.""" + labels_mad = {"id": "errors", "group": "g1", "detector": "MAD"} + labels_hampel = {"id": "errors", "group": "g1", "detector": "Hampel"} + mon_9 = _ts(2025, 5, 5, 9) + + series_mad = _outside_series(firing_timestamps=[mon_9], clear_timestamps=[], labels=labels_mad) + series_hampel = _outside_series( + firing_timestamps=[], clear_timestamps=[mon_9], labels=labels_hampel + ) + + samples = compute_recurrence_samples([series_mad, series_hampel], group="g1") + + mon_9_buckets = [ + s + for s in samples + if dict(s.labels).get("day_of_week") == "Mon" and dict(s.labels).get("hour_of_day") == "9" + ] + assert len(mon_9_buckets) == 1 + # max(1.0, 0.0) = 1.0 -> firing. One observation -> rate 1.0. + assert mon_9_buckets[0].value == 1.0 + + +def test_recurrence_labels_include_required_fields() -> None: + labels = {"id": "errors", "group": "g1"} + mon_9 = _ts(2025, 5, 5, 9) + series = _outside_series(firing_timestamps=[mon_9], clear_timestamps=[], labels=labels) + samples = compute_recurrence_samples([series], group="g1") + assert len(samples) > 0 + for sample in samples: + label_dict = dict(sample.labels) + assert "id" in label_dict + assert "group" in label_dict + assert "day_of_week" in label_dict + assert "hour_of_day" in label_dict + assert sample.metric == "anomaly_recurrence_score" + + +@pytest.mark.asyncio +async def test_runner_emits_recurrence_when_enabled(stub_source: StubSource) -> None: + """When recurrence.enabled is true, the runner emits recurrence scores.""" + mon_9 = _ts(2025, 5, 5, 9) + + def responder(promql: str) -> StubResponse: + if "anomaly_outside_threshold" in promql: + return StubResponse( + series=[ + _outside_series( + firing_timestamps=[mon_9], + clear_timestamps=[], + labels={"id": "q", "group": "g"}, + ) + ] + ) + n = 120 + ts = [float(i) for i in range(n)] + ys = [5.0 + (10.0 if i == n - 1 else 0.0) for i in range(n)] + df = pd.DataFrame({"timestamp": ts, "y": ys}) + return StubResponse(series=[QuerySeries(labels={"k": "v"}, samples=df)]) + + stub_source.respond(responder) + + raw: dict[str, Any] = { + "apiVersion": CURRENT_API_VERSION, + "datasource": {"url": "http://stub"}, + "defaults": {"window": "15m", "step": "1s", "min_points": 5}, + "groups": [ + { + "name": "g", + "recurrence": {"enabled": True, "lookback": "4w", "step": "1h"}, + "queries": [ + {"id": "q", "promql": "up", "detectors": [{"name": "MAD"}]}, + ], + } + ], + } + cfg = Config.model_validate(raw) + store = SnapshotStore() + runner = Runner(cfg, stub_source, store, OperationalMetrics()) + result = await runner.run_group("g") + + assert result.succeeded + recurrence = [s for s in result.samples if s.metric == "anomaly_recurrence_score"] + assert len(recurrence) > 0 + # All emitted samples carry the required labels. + for sample in recurrence: + label_dict = dict(sample.labels) + assert label_dict.get("group") == "g" + assert "id" in label_dict + assert "day_of_week" in label_dict + assert "hour_of_day" in label_dict + + +@pytest.mark.asyncio +async def test_runner_skips_recurrence_when_disabled(stub_source: StubSource) -> None: + n = 120 + ts = [float(i) for i in range(n)] + ys = [5.0 + (10.0 if i == n - 1 else 0.0) for i in range(n)] + df = pd.DataFrame({"timestamp": ts, "y": ys}) + stub_source.respond(lambda _: StubResponse(series=[QuerySeries(labels={"k": "v"}, samples=df)])) + + raw: dict[str, Any] = { + "apiVersion": CURRENT_API_VERSION, + "datasource": {"url": "http://stub"}, + "defaults": {"window": "15m", "step": "1s", "min_points": 5}, + "groups": [ + { + "name": "g", + # recurrence not enabled (default) + "queries": [ + {"id": "q", "promql": "up", "detectors": [{"name": "MAD"}]}, + ], + } + ], + } + cfg = Config.model_validate(raw) + store = SnapshotStore() + runner = Runner(cfg, stub_source, store, OperationalMetrics()) + result = await runner.run_group("g") + + assert result.succeeded + recurrence = [s for s in result.samples if s.metric == "anomaly_recurrence_score"] + assert len(recurrence) == 0 diff --git a/docs/cli.md b/docs/cli.md index 5157f9a..b84b9a7 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -18,6 +18,7 @@ promanomaly ships with a small, focused set of CLI commands for validation, test | `top` | List the currently-firing series, ranked by severity | | `warmup` | Report which series are still warming up, with an ETA to readiness | | `diagnose` | Flag mis-tuned detectors (never-fires, fires-often, stuck warming, empty) | +| `timeline` | Post-incident anomaly timeline over a historical window | | `adapter` | Run the Kubernetes external/custom metrics adapter | | `adapter-validate` | Validate an adapter config file (schema only) | @@ -253,6 +254,30 @@ promanomaly diagnose --target http://localhost:9092 Pure analysis, no persisted state. Pairs with `backtest` (precision/recall on injected anomalies) and `calibrate-buckets` (stratified bucketing) to close the tuning loop. Parity with `promforecast diagnose`. +## `timeline` + +Post-incident anomaly timeline. 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` (which finds change-points in a *raw signal*) by operating on the *emitted anomaly metrics* — the retro view. + +```bash +promanomaly timeline --from -6h --to now +promanomaly timeline --from 2025-05-29T08:00:00Z --to 2025-05-29T12:00:00Z --group api --group web +promanomaly timeline --from -24h --output json +``` + +**Flags** + +| Flag | Default | Description | +|--------------------|----------------------------|-------------| +| `--from ` | (required) | Range start (RFC3339, unix seconds, or relative e.g. `-6h`) | +| `--to ` | `now` | Range end | +| `--group ` | all | Restrict to these groups (repeatable) | +| `--datasource-url` | `http://localhost:8428/` | PromQL-compatible datasource URL | +| `--step` | `15s` | Sample step for the range queries | +| `--output` | `text` | `text` (human table) or `json` (machine-readable list of events) | +| `--timeout` | `30s` | Datasource HTTP timeout | + +Each event carries: `time`, `time_rfc3339`, `kind` (`anomaly` or `change_point`), `id`, `group`, `detector`, `detector_instance`, and `severity` (when available). Events are sorted by ascending time; ties break by kind then `id` for a deterministic order. + ## `adapter` Run the Kubernetes external/custom metrics adapter — a separate, stateless process from the detector that re-serves the anomaly metrics already in the TSDB through `external.metrics.k8s.io` / `custom.metrics.k8s.io`, so HPA and KEDA can consume anomaly signal. It is normally deployed via the [`promanomaly-metrics-adapter`](../charts/promanomaly-metrics-adapter/) chart, which generates this config and wires the serving TLS + APIService registrations. See [adapter.md](adapter.md) for the full surface and the autoscaling guard-rails.