diff --git a/Makefile b/Makefile index 25ff5e0..05deaf6 100644 --- a/Makefile +++ b/Makefile @@ -132,9 +132,13 @@ labels-contract-check: ## Diff LABELS_CONTRACT.md against promforecast's copy. dev-up: ## Start the minimum dev stack: VM, Prometheus, Redis, node-exporter, detector. docker compose -f docker-compose.dev.yml up -d --build +.PHONY: dev-up-load +dev-up-load: ## Start the full demo stack: minimum + http-demo + loadgen + Grafana. + docker compose -f docker-compose.dev.yml --profile load up -d --build + .PHONY: dev-down -dev-down: ## Stop the local dev stack. - docker compose -f docker-compose.dev.yml down -v +dev-down: ## Stop the local dev stack (any profile). + docker compose -f docker-compose.dev.yml --profile load down -v .PHONY: load-test load-test: ## Synthetic high-cardinality run for safety-control validation. diff --git a/community/promanomaly-prophet/tests/test_detector.py b/community/promanomaly-prophet/tests/test_detector.py index 3fe7c24..8d530c6 100644 --- a/community/promanomaly-prophet/tests/test_detector.py +++ b/community/promanomaly-prophet/tests/test_detector.py @@ -42,9 +42,7 @@ def test_fit_score_normalises_residual_by_band(stub_prophet: type) -> None: assert bool(row["is_outside"]) is False -def test_params_forwarded_to_prophet( - monkeypatch: pytest.MonkeyPatch, stub_prophet: type -) -> None: +def test_params_forwarded_to_prophet(monkeypatch: pytest.MonkeyPatch, stub_prophet: type) -> None: # Sanity check: custom params reach the Prophet constructor. We # wrap the stub with a capturing subclass that records the kwargs # it was called with — simpler than threading state through the diff --git a/dashboards/grafana/promanomaly-change-points.json b/dashboards/grafana/promanomaly-change-points.json index 5911e3f..bf73577 100644 --- a/dashboards/grafana/promanomaly-change-points.json +++ b/dashboards/grafana/promanomaly-change-points.json @@ -1,9 +1,16 @@ { "annotations": { "list": [ + { + "name": "datasource", + "type": "datasource", + "label": "Datasource", + "query": "prometheus", + "current": {"text": "VictoriaMetrics", "value": "victoriametrics"} + }, { "name": "change-points", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "enable": true, "iconColor": "red", "expr": "changes(anomaly_change_point_total{group=~\"$group\", id=~\"$id\"}[5m]) > 0", @@ -21,7 +28,7 @@ { "name": "group", "type": "query", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "query": "label_values(anomaly_change_point_total, group)", "refresh": 2, "multi": true, @@ -30,7 +37,7 @@ { "name": "id", "type": "query", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "query": "label_values(anomaly_change_point_total{group=~\"$group\"}, id)", "refresh": 2, "multi": true, @@ -39,7 +46,7 @@ { "name": "detector", "type": "query", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "query": "label_values(anomaly_change_point_total{group=~\"$group\", id=~\"$id\"}, detector)", "refresh": 2, "multi": true, @@ -52,7 +59,7 @@ "id": 1, "type": "stat", "title": "Change-points in the last 24h", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "sum(increase(anomaly_change_point_total{group=~\"$group\", id=~\"$id\", detector=~\"$detector\"}[24h]))", @@ -66,7 +73,7 @@ "id": 2, "type": "stat", "title": "Series with recent change-points", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "count(increase(anomaly_change_point_total{group=~\"$group\", id=~\"$id\", detector=~\"$detector\"}[1h]) > 0)", @@ -80,7 +87,7 @@ "id": 3, "type": "stat", "title": "Active change-point detectors", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "count(count by (detector) (anomaly_change_point_total{group=~\"$group\", detector=~\"$detector\"}))", @@ -93,7 +100,7 @@ "id": 4, "type": "stat", "title": "Change-points last 5m", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "sum(increase(anomaly_change_point_total{group=~\"$group\", id=~\"$id\", detector=~\"$detector\"}[5m]))", @@ -108,7 +115,7 @@ "type": "timeseries", "title": "Change-point rate by detector", "description": "Rate of change-point firings per detector. Each spike is a regime shift the detector flagged.", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "sum by (detector) (rate(anomaly_change_point_total{group=~\"$group\", id=~\"$id\", detector=~\"$detector\"}[5m]))", @@ -122,7 +129,7 @@ "type": "timeseries", "title": "anomaly_score (change-point detectors)", "description": "Underlying score driving change-point counters. For BOCPD the score is a posterior probability; for CUSUM it is a sigma multiplier.", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "anomaly_score{group=~\"$group\", id=~\"$id\", detector=~\"BOCPD|CUSUM\"}", @@ -136,7 +143,7 @@ "type": "table", "title": "Top-N change-points (last 1h)", "description": "Series with the most change-point firings in the last hour. Click through to drill down into the underlying signal.", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "topk(20, increase(anomaly_change_point_total{group=~\"$group\", id=~\"$id\", detector=~\"$detector\"}[1h]))", @@ -151,7 +158,7 @@ "type": "timeseries", "title": "Source signal with change-point overlay", "description": "Underlying baseline overlaid for context. Annotations (top of dashboard) mark change-point firings.", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "anomaly_baseline{group=~\"$group\", id=~\"$id\", detector=~\"$detector\"}", diff --git a/dashboards/grafana/promanomaly-cohort.json b/dashboards/grafana/promanomaly-cohort.json index 6bcc147..dbfda8a 100644 --- a/dashboards/grafana/promanomaly-cohort.json +++ b/dashboards/grafana/promanomaly-cohort.json @@ -6,10 +6,17 @@ "tags": ["promanomaly", "anomaly-detection", "cohort"], "templating": { "list": [ + { + "name": "datasource", + "type": "datasource", + "label": "Datasource", + "query": "prometheus", + "current": {"text": "VictoriaMetrics", "value": "victoriametrics"} + }, { "name": "group", "type": "query", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "query": "label_values(anomaly_score{detector=\"Cohort\"}, group)", "refresh": 2, "multi": true, @@ -18,7 +25,7 @@ { "name": "id", "type": "query", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "query": "label_values(anomaly_score{detector=\"Cohort\", group=~\"$group\"}, id)", "refresh": 2, "multi": true, @@ -31,7 +38,7 @@ "id": 1, "type": "stat", "title": "Cohort members currently diverging", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "count(anomaly_outside_threshold{detector=\"Cohort\", group=~\"$group\", id=~\"$id\"} == 1)", @@ -45,7 +52,7 @@ "id": 2, "type": "stat", "title": "Active cohorts", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "count(count by (id, group) (anomaly_score{detector=\"Cohort\", group=~\"$group\", id=~\"$id\"}))", @@ -59,7 +66,7 @@ "id": 3, "type": "stat", "title": "Max divergence score (60s)", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "max(max_over_time(anomaly_score{detector=\"Cohort\", group=~\"$group\", id=~\"$id\"}[1m]))", @@ -73,7 +80,7 @@ "id": 4, "type": "table", "title": "Top-N currently diverging cohort members", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "topk(10, anomaly_score{detector=\"Cohort\", group=~\"$group\", id=~\"$id\"})", @@ -88,7 +95,7 @@ "type": "timeseries", "title": "Cohort divergence score over time", "description": "Per-member divergence from the cohort baseline. A member rising above the rest is the 'one bad node in the fleet' case the Cohort detector targets.", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "anomaly_score{detector=\"Cohort\", group=~\"$group\", id=~\"$id\"}", @@ -102,7 +109,7 @@ "type": "timeseries", "title": "Cohort baseline vs. individual members", "description": "anomaly_baseline carries the cross-member median for the Cohort detector. Overlay the source signal to see where each member sits relative to the fleet.", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "anomaly_baseline{detector=\"Cohort\", group=~\"$group\", id=~\"$id\"}", @@ -116,7 +123,7 @@ "type": "timeseries", "title": "Outside-threshold transitions (sustained divergence)", "description": "anomaly_outside_threshold transitions for Cohort. Used by the AnomalyOutsideThreshold reference alert with for: 5m flap suppression.", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "anomaly_outside_threshold{detector=\"Cohort\", group=~\"$group\", id=~\"$id\"}", diff --git a/dashboards/grafana/promanomaly-confidence.json b/dashboards/grafana/promanomaly-confidence.json index c7722af..038bbbc 100644 --- a/dashboards/grafana/promanomaly-confidence.json +++ b/dashboards/grafana/promanomaly-confidence.json @@ -6,10 +6,17 @@ "tags": ["promanomaly", "anomaly-detection", "confidence"], "templating": { "list": [ + { + "name": "datasource", + "type": "datasource", + "label": "Datasource", + "query": "prometheus", + "current": {"text": "VictoriaMetrics", "value": "victoriametrics"} + }, { "name": "group", "type": "query", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "query": "label_values(anomaly_confidence_score, group)", "refresh": 2, "multi": true, @@ -18,7 +25,7 @@ { "name": "id", "type": "query", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "query": "label_values(anomaly_confidence_score{group=~\"$group\"}, id)", "refresh": 2, "multi": true, @@ -31,7 +38,7 @@ "id": 1, "type": "timeseries", "title": "Confidence score per detector", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "anomaly_confidence_score{group=~\"$group\", id=~\"$id\"}", @@ -58,7 +65,7 @@ "id": 2, "type": "timeseries", "title": "Baseline stability per detector", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "anomaly_baseline_stability{group=~\"$group\", id=~\"$id\"}", @@ -85,7 +92,7 @@ "id": 3, "type": "table", "title": "Low-confidence firing right now", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "topk(20, anomaly_confidence_score{group=~\"$group\"} < 0.5 and on (id, group, detector) anomaly_outside_threshold == 1)", @@ -99,7 +106,7 @@ "id": 4, "type": "stat", "title": "High-confidence anomalies", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "count(anomaly_outside_threshold{group=~\"$group\"} == 1 and on (id, group, detector) anomaly_confidence_score >= 0.7)", @@ -112,7 +119,7 @@ "id": 5, "type": "stat", "title": "Low-confidence anomalies", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "count(anomaly_outside_threshold{group=~\"$group\"} == 1 and on (id, group, detector) anomaly_confidence_score < 0.5)", @@ -125,7 +132,7 @@ "id": 6, "type": "stat", "title": "Series with no calibration yet", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "count(anomaly_score{group=~\"$group\"}) - count(anomaly_confidence_score{group=~\"$group\"})", diff --git a/dashboards/grafana/promanomaly-detector-comparison.json b/dashboards/grafana/promanomaly-detector-comparison.json index e179804..7d7f5a5 100644 --- a/dashboards/grafana/promanomaly-detector-comparison.json +++ b/dashboards/grafana/promanomaly-detector-comparison.json @@ -6,10 +6,17 @@ "tags": ["promanomaly", "anomaly-detection", "detectors"], "templating": { "list": [ + { + "name": "datasource", + "type": "datasource", + "label": "Datasource", + "query": "prometheus", + "current": {"text": "VictoriaMetrics", "value": "victoriametrics"} + }, { "name": "group", "type": "query", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "query": "label_values(anomaly_score, group)", "refresh": 2, "multi": false, @@ -18,7 +25,7 @@ { "name": "id", "type": "query", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "query": "label_values(anomaly_score{group=~\"$group\"}, id)", "refresh": 2, "multi": false, @@ -31,7 +38,7 @@ "id": 1, "type": "timeseries", "title": "anomaly_score by detector", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "anomaly_score{group=~\"$group\", id=~\"$id\"}", @@ -44,7 +51,7 @@ "id": 2, "type": "timeseries", "title": "anomaly_outside_threshold by detector", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "anomaly_outside_threshold{group=~\"$group\", id=~\"$id\"}", @@ -60,7 +67,7 @@ "id": 3, "type": "timeseries", "title": "Composite (ensemble) verdict", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "anomaly_composite_score{group=~\"$group\", id=~\"$id\"}", @@ -77,7 +84,7 @@ "id": 4, "type": "timeseries", "title": "anomaly_duration_seconds by detector", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "anomaly_duration_seconds{group=~\"$group\", id=~\"$id\"}", @@ -90,7 +97,7 @@ "id": 5, "type": "timeseries", "title": "anomaly_confidence_score by detector", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "anomaly_confidence_score{group=~\"$group\", id=~\"$id\"}", @@ -106,7 +113,7 @@ "id": 6, "type": "table", "title": "Current per-detector verdict", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "anomaly_score{group=~\"$group\", id=~\"$id\"}", diff --git a/dashboards/grafana/promanomaly-overview.json b/dashboards/grafana/promanomaly-overview.json index 3a2f645..e2ad68d 100644 --- a/dashboards/grafana/promanomaly-overview.json +++ b/dashboards/grafana/promanomaly-overview.json @@ -6,10 +6,17 @@ "tags": ["promanomaly", "anomaly-detection"], "templating": { "list": [ + { + "name": "datasource", + "type": "datasource", + "label": "Datasource", + "query": "prometheus", + "current": {"text": "VictoriaMetrics", "value": "victoriametrics"} + }, { "name": "group", "type": "query", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "query": "label_values(anomaly_score, group)", "refresh": 2, "multi": true, @@ -18,7 +25,7 @@ { "name": "id", "type": "query", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "query": "label_values(anomaly_score{group=~\"$group\"}, id)", "refresh": 2, "multi": true, @@ -31,7 +38,7 @@ "id": 1, "type": "timeseries", "title": "anomaly_score", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "anomaly_score{group=~\"$group\", id=~\"$id\"}", @@ -44,7 +51,7 @@ "id": 2, "type": "timeseries", "title": "anomaly_baseline vs. actual", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "anomaly_baseline{group=~\"$group\", id=~\"$id\"}", @@ -57,7 +64,7 @@ "id": 3, "type": "table", "title": "Top-N current anomalies", - "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"}, + "datasource": {"type": "prometheus", "uid": "${datasource}"}, "targets": [ { "expr": "topk(10, anomaly_score{group=~\"$group\"})", diff --git a/detector/src/promanomaly/cache.py b/detector/src/promanomaly/cache.py index 7a7d4ae..f9a99c0 100644 --- a/detector/src/promanomaly/cache.py +++ b/detector/src/promanomaly/cache.py @@ -60,6 +60,12 @@ class TTLCache[K, V]: interface. """ + # In-process: ``get``/``put`` are lock-protected memory ops with no + # network round-trip, so the source layer calls them inline on the + # event loop. The Redis backend overrides this to True so the source + # offloads its blocking socket I/O to a worker thread. + is_blocking_io: bool = False + def __init__(self, *, max_entries: int, ttl_seconds: float) -> None: if max_entries < 1: raise ValueError("max_entries must be >= 1") @@ -133,6 +139,12 @@ class RedisQueryCache[K, V]: the TTL. """ + # ``get``/``put`` do blocking Redis socket I/O, so the source layer + # must offload them to a worker thread rather than running them on the + # asyncio event loop (where a slow Redis would stall every group's + # detection run and the /metrics handler). + is_blocking_io: bool = True + # Size lookup is throttled to once per this many seconds. Without # the throttle, ``__len__`` issues a SCAN over every cache key on # every ``/metrics`` render; with a default 15s scrape interval, diff --git a/detector/src/promanomaly/exporter.py b/detector/src/promanomaly/exporter.py index f475190..c91f85b 100644 --- a/detector/src/promanomaly/exporter.py +++ b/detector/src/promanomaly/exporter.py @@ -139,6 +139,18 @@ def validate_label_name(name: str) -> str: "handling absorbs that." ), ), + "anomaly_best_detector": ( + "gauge", + ( + "1 for the detector auto-select picked as best for a series; " + "emitted only for the winner under auto_select. An info-style " + "marker on its own series, so the winner can flip between " + "auto_select_interval cycles without churning the label set of " + "anomaly_score/anomaly_outside_threshold/etc. Join with " + "`* on (id, group, detector, ...) group_left` to filter the " + "per-detector metrics down to the winner." + ), + ), } diff --git a/detector/src/promanomaly/runner.py b/detector/src/promanomaly/runner.py index fbc4ead..b4e759a 100644 --- a/detector/src/promanomaly/runner.py +++ b/detector/src/promanomaly/runner.py @@ -19,6 +19,7 @@ import asyncio import concurrent.futures +import contextvars import os import time from dataclasses import dataclass, field @@ -125,6 +126,34 @@ class GroupRunResult: queries_succeeded: int = 0 +# Per-detector compute time captured during a single group run, flushed +# onto the operational gauge at the end of that run. A ContextVar rather +# than an instance attribute because the scheduler fires every group as +# its own concurrent asyncio task against one shared Runner; an instance +# dict would be reset and mutated by all groups at once, cross-attributing +# detector durations to groups that never ran them (e.g. a MAD-only group +# reporting a Hampel duration). Each group run gets a fresh task-local dict +# via ``_DETECT_DURATIONS.set`` so the timings stay scoped to their group. +_DETECT_DURATIONS: contextvars.ContextVar[dict[str, float]] = contextvars.ContextVar( + "promanomaly_detect_durations" +) + + +def _current_detect_durations() -> dict[str, float]: + """Return the current run's duration accumulator, creating one if unset. + + The real pipeline always calls :meth:`Runner.run_group`, which seeds the + accumulator before any detector runs; the fallback only matters for tests + that exercise :meth:`Runner._run_detector` in isolation. + """ + try: + return _DETECT_DURATIONS.get() + except LookupError: + durations: dict[str, float] = {} + _DETECT_DURATIONS.set(durations) + return durations + + class Runner: """Owns the detection pipeline shared across all groups. @@ -163,9 +192,9 @@ def __init__( # reset would defeat the cross-group cap. Updated when a group # finishes (success path); failures leave the old count in place. self._last_series_count: dict[str, int] = {} - # Per-detector compute time captured during the current run; - # flushed onto the operational gauge at the end of the run. - self._detect_durations: dict[str, float] = {} + # Per-detector compute time is tracked in the ``_DETECT_DURATIONS`` + # ContextVar (task-local per group run), not on the instance — see + # the module-level note for why a shared dict races across groups. # Stratified baseline cache + sliding-window fetch live behind # this facade; the runner delegates to it rather than owning # the heavy cache state directly. @@ -266,7 +295,10 @@ async def run_group(self, group_name: str) -> GroupRunResult: async with self._group_locks[group_name]: self._store.mark_attempted(group_name) run_start = time.monotonic() - self._detect_durations = {} + # Seed a fresh, task-local duration accumulator for this run. + # Isolated per group because each group runs in its own + # concurrent scheduler task (see ``_DETECT_DURATIONS`` note). + _DETECT_DURATIONS.set({}) result = GroupRunResult(group=group_name) source_failed = False try: @@ -1133,17 +1165,14 @@ async def _run_detector(self, plan: _DetectorPlan, series: QuerySeries) -> pd.Se plan.params, ) started = time.monotonic() + durations = _current_detect_durations() try: df = await asyncio.wait_for(future, timeout=timeout) except TimeoutError as exc: future.cancel() - self._detect_durations[plan.name] = self._detect_durations.get(plan.name, 0.0) + ( - time.monotonic() - started - ) + durations[plan.name] = durations.get(plan.name, 0.0) + (time.monotonic() - started) raise concurrent.futures.TimeoutError("detector timeout") from exc - self._detect_durations[plan.name] = self._detect_durations.get(plan.name, 0.0) + ( - time.monotonic() - started - ) + durations[plan.name] = durations.get(plan.name, 0.0) + (time.monotonic() - started) if df.empty: raise RuntimeError(f"detector {plan.name} returned no rows") return df.iloc[-1] @@ -1222,11 +1251,15 @@ def _format_detector_samples( emitted Sample list. """ extras: dict[str, str] = {} - if mark_as_winner: - # Opt-in label: only emitted when this detector is the - # auto-select winner for this series. Opt-in labels are - # non-breaking per the project's API contract. - extras["best_detector"] = plan.name + # The auto-select winner is NOT marked by adding a ``best_detector`` + # label to this detector's samples: the winner can flip every + # ``auto_select_interval``, and a label that appears/disappears + # churns the series identity of anomaly_score / outside_threshold / + # the change-point counter — breaking rate()/increase() on the + # counter and resetting ``for:`` timers on the outside_threshold + # alert. Instead the winner is published as a standalone + # ``anomaly_best_detector{...} = 1`` info gauge below, so the + # existing per-detector series keep a stable label set. if cohort_label_extra is not None: # Opt-in label emitted by the Cohort detector exclusively # (set by the runner when a cohort-aware plan scored this @@ -1274,6 +1307,14 @@ def _format_detector_samples( value=1.0 if outside else 0.0, ), ] + if mark_as_winner: + # Standalone info gauge naming the auto-select winner. Shares + # the per-detector label set (id, group, detector, + # detector_instance, source labels) so a PromQL join filters + # the other metrics down to the winner, while keeping the + # winner marker off those metrics' own label sets (see the + # note where ``extras`` is built). + samples.append(Sample(metric="anomaly_best_detector", labels=labels, value=1.0)) if defaults.emit_baseline: samples.append(Sample(metric="anomaly_baseline", labels=labels, value=baseline)) if defaults.emit_duration: @@ -1299,6 +1340,11 @@ def _format_detector_samples( detector_instance=plan.instance, fired=fired, ) + # ``labels`` is now a stable set for this (id, group, detector) + # series — the auto-select winner marker lives on the separate + # ``anomaly_best_detector`` gauge rather than as a label here, + # so this monotonic counter no longer splits across winner / + # non-winner variants and rate()/increase() stay continuous. samples.append( Sample( metric="anomaly_change_point_total", @@ -1555,7 +1601,10 @@ def _update_operational_metrics(self, result: GroupRunResult) -> None: self._ops.source_failure_streak.labels(group=result.group).set( float(self._store.source_failure_streak(result.group)) ) - for detector_name, elapsed in self._detect_durations.items(): + # Read the task-local accumulator seeded at the top of this run; + # ``_update_operational_metrics`` runs in the same task as + # ``run_group`` so the ContextVar resolves to this group's dict. + for detector_name, elapsed in _DETECT_DURATIONS.get({}).items(): self._ops.detect_duration_seconds.labels( group=result.group, detector=detector_name ).set(elapsed) diff --git a/detector/src/promanomaly/source.py b/detector/src/promanomaly/source.py index 7a53f5c..5c7c99f 100644 --- a/detector/src/promanomaly/source.py +++ b/detector/src/promanomaly/source.py @@ -9,6 +9,7 @@ from __future__ import annotations +import asyncio import time from dataclasses import dataclass, field from typing import Any @@ -180,7 +181,7 @@ async def range_query( int(max_series or 0), bucket, ) - cached = self._cache.get(cache_key, time.monotonic()) + cached = await self._cache_get(cache_key) if cached is not None: return cached @@ -195,12 +196,30 @@ async def range_query( "end": f"{end:.3f}", "step": f"{step_seconds:.3f}", } - try: - response = await self._client.get("/api/v1/query_range", params=params) - except httpx.TimeoutException as exc: - raise SourceQueryError("timeout", str(exc)) from exc - except httpx.HTTPError as exc: - raise SourceQueryError("transport_error", str(exc)) from exc + # One transparent retry on a transient transport error. The common + # case is a keepalive connection the TSDB (or an intermediary) + # closed while idle between scrapes: the next request on that pooled + # connection fails immediately and a fresh connection succeeds. A + # single retry absorbs that blip instead of degrading the whole + # group to serve_stale and tripping AnomalySourceFailing on a + # harmless hiccup. Timeouts are NOT retried — under a slow TSDB a + # retry only compounds load, and the per-query timeout already + # bounds the wait. HTTP error envelopes (4xx/5xx) are real server + # responses handled below, not transport failures, so they never + # reach this retry. + response: httpx.Response | None = None + transport_exc: httpx.HTTPError | None = None + for _attempt in range(2): + try: + response = await self._client.get("/api/v1/query_range", params=params) + break + except httpx.TimeoutException as exc: + raise SourceQueryError("timeout", _exc_detail(exc)) from exc + except httpx.HTTPError as exc: + transport_exc = exc + if response is None: + assert transport_exc is not None + raise SourceQueryError("transport_error", _exc_detail(transport_exc)) from transport_exc if response.status_code >= 500: raise SourceQueryError("server_error", f"HTTP {response.status_code}") @@ -239,9 +258,45 @@ async def range_query( query_result = QueryResult(series=series_list, truncated=truncated) if self._cache is not None and cache_key is not None: - self._cache.put(cache_key, query_result, time.monotonic()) + await self._cache_put(cache_key, query_result) return query_result + async def _cache_get(self, cache_key: tuple[str, float, float, int, int]) -> QueryResult | None: + """Read from the cache without blocking the event loop. + + The in-process :class:`TTLCache` is a lock-guarded memory lookup and + runs inline; the Redis backend (``is_blocking_io=True``) does socket + I/O, so it is offloaded to a worker thread to keep the loop free for + other groups' detection runs. + """ + assert self._cache is not None + if getattr(self._cache, "is_blocking_io", False): + return await asyncio.to_thread(self._cache.get, cache_key, time.monotonic()) + return self._cache.get(cache_key, time.monotonic()) + + async def _cache_put( + self, cache_key: tuple[str, float, float, int, int], value: QueryResult + ) -> None: + """Write to the cache without blocking the event loop (see ``_cache_get``).""" + assert self._cache is not None + if getattr(self._cache, "is_blocking_io", False): + await asyncio.to_thread(self._cache.put, cache_key, value, time.monotonic()) + return + self._cache.put(cache_key, value, time.monotonic()) + + +def _exc_detail(exc: Exception) -> str: + """Human-readable detail for a transport exception. + + httpx transport errors (``RemoteProtocolError`` for a server-closed + connection, bare ``ConnectError``/``ReadError``) frequently stringify to + an empty message, which left the ``source_query_failed`` log line and the + ``anomaly_source_failure_total`` counter with no indication of *what* + failed. Fall back to the exception class name so the failure is always + diagnosable. + """ + return str(exc) or type(exc).__name__ + def _build_samples_frame(samples: list[list[Any]]) -> pd.DataFrame: """Vectorised parse of Prometheus ``[[ts, "value"], ...]`` payloads. diff --git a/detector/tests/test_auto_select_and_confidence.py b/detector/tests/test_auto_select_and_confidence.py index 3e879c8..c7d967d 100644 --- a/detector/tests/test_auto_select_and_confidence.py +++ b/detector/tests/test_auto_select_and_confidence.py @@ -111,8 +111,15 @@ async def test_auto_select_emits_only_winner_by_default(stub_source: StubSource) # only one winner per series. assert len(score_samples) == 1 labels = dict(score_samples[0].labels) - assert labels["best_detector"] == labels["detector"] + # The winner marker is no longer a label on anomaly_score — that would + # churn the series identity each time the winner flips. It is published + # as a standalone anomaly_best_detector=1 info gauge instead. + assert "best_detector" not in labels assert labels["detector"] in {"MAD", "Hampel"} + best = [s for s in result.samples if s.metric == "anomaly_best_detector"] + assert len(best) == 1 + assert best[0].value == 1.0 + assert dict(best[0].labels)["detector"] == labels["detector"] @pytest.mark.asyncio @@ -128,13 +135,17 @@ async def test_auto_select_explicit_plus_best_keeps_all(stub_source: StubSource) result = await runner.run_group("g1") score_samples = [s for s in result.samples if s.metric == "anomaly_score"] - # explicit_plus_best emits both detectors, tags the winner with - # best_detector= but does not tag the loser. + # explicit_plus_best emits both detectors' scores... assert len(score_samples) == 2 - winners = [s for s in score_samples if "best_detector" in dict(s.labels)] - losers = [s for s in score_samples if "best_detector" not in dict(s.labels)] - assert len(winners) == 1 - assert len(losers) == 1 + # ...with NO best_detector label on either — the winner marker lives on + # a standalone anomaly_best_detector info gauge so the score series keep + # a stable identity across winner flips. + assert all("best_detector" not in dict(s.labels) for s in score_samples) + best = [s for s in result.samples if s.metric == "anomaly_best_detector"] + assert len(best) == 1 + winner_detector = dict(best[0].labels)["detector"] + assert winner_detector in {"MAD", "Hampel"} + assert winner_detector in {dict(s.labels)["detector"] for s in score_samples} @pytest.mark.asyncio diff --git a/detector/tests/test_change_point_metrics.py b/detector/tests/test_change_point_metrics.py index 22a67f5..9dc2ff6 100644 --- a/detector/tests/test_change_point_metrics.py +++ b/detector/tests/test_change_point_metrics.py @@ -184,6 +184,57 @@ async def test_emit_change_points_false_suppresses_counter(stub_source: StubSour assert "anomaly_change_point_total" not in metrics +@pytest.mark.asyncio +async def test_no_metric_carries_best_detector_label(stub_source: StubSource) -> None: + """Regression: the auto-select winner marker ``best_detector`` must NOT + ride any per-detector metric — not the gauges, and especially not the + ``anomaly_change_point_total`` counter. + + The marker flips between detectors every ``auto_select_interval``. As a + label it churns series identity: on the counter that breaks + rate()/increase() and group-level sums; on outside_threshold it resets + ``for:`` alert timers. The winner is instead published as a standalone + ``anomaly_best_detector=1`` info gauge, so every per-detector series + keeps a stable label set.""" + df = step_change(n=120, step=5.0, noise=0.1, change_at=119) + stub_source.respond(lambda _: StubResponse(series=[make_series({"instance": "host-a"}, df)])) + + raw = { + "apiVersion": CURRENT_API_VERSION, + "datasource": {"url": "http://stub"}, + "defaults": { + "window": "15m", + "step": "15s", + "min_points": 30, + "warmup_policy": "emit_warming_up", + "emit_change_points": True, + "alert_thresholds": {"score": 3.0}, + }, + "groups": [ + { + "name": "g1", + # auto_select marks the single detector as the winner. + "auto_select": True, + "queries": [{"id": "m", "promql": "up", "detectors": [{"name": "BOCPD"}]}], + } + ], + } + cfg = Config.model_validate(raw) + runner = Runner(cfg, stub_source, SnapshotStore(), OperationalMetrics()) + result = await runner.run_group("g1") + + # No per-detector metric carries best_detector as a label... + for s in result.samples: + if s.metric != "anomaly_best_detector": + assert "best_detector" not in s.label_dict, s.metric + cp = next(s for s in result.samples if s.metric == "anomaly_change_point_total") + assert cp.label_dict["detector"] == "BOCPD" + # ...the winner is named by a single anomaly_best_detector info gauge. + best = next(s for s in result.samples if s.metric == "anomaly_best_detector") + assert best.value == 1.0 + assert best.label_dict["detector"] == "BOCPD" + + def test_exporter_renders_counter_type_declaration() -> None: """The exposition format must declare TYPE counter for the metric so Prometheus client libraries handle counter-reset detection correctly.""" diff --git a/detector/tests/test_safety_caps.py b/detector/tests/test_safety_caps.py index 6ef8c5b..bf3abbf 100644 --- a/detector/tests/test_safety_caps.py +++ b/detector/tests/test_safety_caps.py @@ -2,6 +2,8 @@ from __future__ import annotations +import asyncio + import pytest from promanomaly.config import CURRENT_API_VERSION, Config @@ -112,3 +114,62 @@ async def test_detect_duration_seconds_gauge_populated(stub_source: StubSource) ) assert elapsed is not None assert elapsed >= 0.0 + + +@pytest.mark.asyncio +async def test_detect_duration_not_cross_attributed_across_concurrent_groups( + stub_source: StubSource, +) -> None: + """Regression: detect-duration timings must stay scoped to the group + that ran the detector. + + The scheduler fires every group as its own concurrent asyncio task + against one shared Runner. A previous implementation accumulated + durations in a single instance dict, so concurrent runs cross- + attributed each other's detectors — a MAD-only group would publish an + IQR duration and vice versa. The accumulator is now task-local + (``_DETECT_DURATIONS``), so each group emits only its own detectors. + """ + cfg = Config.model_validate( + { + "apiVersion": CURRENT_API_VERSION, + "datasource": {"url": "http://stub"}, + "defaults": { + "window": "5m", + "step": "15s", + "min_points": 5, + "warmup_policy": "emit_warming_up", + "alert_thresholds": {"score": 3.0}, + }, + "groups": [ + { + "name": "ga", + "queries": [{"id": "ma", "promql": "up_a", "detectors": [{"name": "MAD"}]}], + }, + { + "name": "gb", + "queries": [{"id": "mb", "promql": "up_b", "detectors": [{"name": "IQR"}]}], + }, + ], + } + ) + stub_source.respond(lambda _: StubResponse(series=[make_series({}, clean_baseline(n=120))])) + + ops = OperationalMetrics() + runner = Runner(cfg, stub_source, SnapshotStore(), ops) + + # Run both groups concurrently, as the scheduler does. + await asyncio.gather(runner.run_group("ga"), runner.run_group("gb")) + + def sample(group: str, detector: str) -> float | None: + return ops.registry.get_sample_value( + "anomaly_detect_duration_seconds", + {"group": group, "detector": detector}, + ) + + # Each group reports its own detector... + assert sample("ga", "MAD") is not None + assert sample("gb", "IQR") is not None + # ...and never the other group's. + assert sample("ga", "IQR") is None + assert sample("gb", "MAD") is None diff --git a/detector/tests/test_source_retry.py b/detector/tests/test_source_retry.py new file mode 100644 index 0000000..cfddcfc --- /dev/null +++ b/detector/tests/test_source_retry.py @@ -0,0 +1,94 @@ +"""Transient-transport-error handling in :class:`PromQLSource`. + +A keepalive connection the TSDB closed while idle fails the next request on +that pooled connection and succeeds on a fresh one. The source retries such +transport errors exactly once so a harmless hiccup doesn't degrade the whole +group to serve_stale (and trip AnomalySourceFailing). Timeouts are not +retried. Transport errors that stringify empty still produce a diagnosable +detail (the exception class name). +""" + +from __future__ import annotations + +import httpx +import pytest + +from promanomaly.source import PromQLSource, SourceQueryError + +_OK_PAYLOAD = { + "status": "success", + "data": { + "resultType": "matrix", + "result": [{"metric": {"instance": "h"}, "values": [[0.0, "1.0"], [15.0, "2.0"]]}], + }, +} + + +def _source_with_handler(handler) -> PromQLSource: # type: ignore[no-untyped-def] + src = PromQLSource(base_url="http://stub", timeout=1.0) + src._client = httpx.AsyncClient(transport=httpx.MockTransport(handler), base_url="http://stub") + return src + + +@pytest.mark.asyncio +async def test_transient_transport_error_is_retried_once() -> None: + calls = {"n": 0} + + def handler(request: httpx.Request) -> httpx.Response: + calls["n"] += 1 + if calls["n"] == 1: + # Simulate a server-closed keepalive connection. + raise httpx.RemoteProtocolError("Server disconnected", request=request) + return httpx.Response(200, json=_OK_PAYLOAD) + + src = _source_with_handler(handler) + try: + result = await src.range_query( + promql="up", end=100.0, window_seconds=60.0, step_seconds=15.0 + ) + finally: + await src.close() + + assert calls["n"] == 2, "first attempt fails, retry succeeds" + assert result.series and result.series[0].labels == {"instance": "h"} + + +@pytest.mark.asyncio +async def test_transport_error_gives_up_after_one_retry() -> None: + calls = {"n": 0} + + def handler(request: httpx.Request) -> httpx.Response: + calls["n"] += 1 + # Empty-message error: detail must still be diagnosable. + raise httpx.ConnectError("", request=request) + + src = _source_with_handler(handler) + try: + with pytest.raises(SourceQueryError) as excinfo: + await src.range_query(promql="up", end=100.0, window_seconds=60.0, step_seconds=15.0) + finally: + await src.close() + + assert calls["n"] == 2, "exactly one retry, then give up" + assert excinfo.value.reason == "transport_error" + # Empty str(exc) falls back to the class name rather than an empty detail. + assert excinfo.value.message == "ConnectError" + + +@pytest.mark.asyncio +async def test_timeout_is_not_retried() -> None: + calls = {"n": 0} + + def handler(request: httpx.Request) -> httpx.Response: + calls["n"] += 1 + raise httpx.ReadTimeout("timed out", request=request) + + src = _source_with_handler(handler) + try: + with pytest.raises(SourceQueryError) as excinfo: + await src.range_query(promql="up", end=100.0, window_seconds=60.0, step_seconds=15.0) + finally: + await src.close() + + assert calls["n"] == 1, "timeouts must not be retried — a retry only compounds TSDB load" + assert excinfo.value.reason == "timeout" diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index 123b7d3..4e58029 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -3,8 +3,17 @@ # Prometheus :9090 # node-exporter :9100 # detector :9092 (/metrics, /healthz, /-/reload, /ready, /debug/*) +# http-demo :8000 (/metrics; "load" profile only) # Redis :6379 (query-cache backend; HA isn't wired in compose # since leader election needs a kube API server) +# Grafana :3000 (admin / admin) — "load" profile only +# +# Two profiles: +# default (`make dev-up`) — VM, Prometheus, Redis, node-exporter, +# detector. Minimum for detector +# development. +# load (`make dev-up-load`) — adds http-demo, http-demo-loadgen, +# Grafana. Full demo experience. name: promanomaly-dev @@ -27,6 +36,39 @@ services: retries: 20 restart: unless-stopped + http-demo: + # Tiny FastAPI app with a handful of endpoints + /metrics. Driven by + # the loadgen below so the detector has http_* time series to score on. + # Behind the "load" profile so `make dev-up` (minimum dev stack) + # doesn't start it; `make dev-up-load` does. + profiles: ["load"] + build: + context: ./docker/http-demo + dockerfile: Dockerfile.app + image: promanomaly-http-demo:dev + container_name: pa-http-demo + ports: + - "8000:8000" + restart: unless-stopped + + http-demo-loadgen: + # ~20-30 req/min with a slow sinusoidal modulation so the rate has shape. + # Same profile as http-demo since one without the other is useless. + profiles: ["load"] + build: + context: ./docker/http-demo + dockerfile: Dockerfile.loadgen + image: promanomaly-http-demo-loadgen:dev + container_name: pa-http-demo-loadgen + environment: + TARGET: http://http-demo:8000 + BASE_RPM: "25" + AMPLITUDE_RPM: "15" + PERIOD_SECONDS: "600" + depends_on: + - http-demo + restart: unless-stopped + node-exporter: image: prom/node-exporter:v1.11.1 container_name: pa-node-exporter @@ -101,6 +143,31 @@ services: condition: service_healthy restart: unless-stopped + grafana: + # Grafana is in the "load" profile because the bundled dashboards are + # tied to the http-demo data; a minimum `make dev-up` for detector + # development doesn't need it. `make dev-up-load` brings it up. + profiles: ["load"] + image: grafana/grafana:13.0.1 + container_name: pa-grafana + environment: + GF_SECURITY_ADMIN_USER: admin + GF_SECURITY_ADMIN_PASSWORD: admin + GF_AUTH_ANONYMOUS_ENABLED: "true" + GF_AUTH_ANONYMOUS_ORG_ROLE: Viewer + GF_USERS_DEFAULT_THEME: dark + ports: + - "3000:3000" + volumes: + - ./docker/grafana/provisioning:/etc/grafana/provisioning:ro + - ./dashboards/grafana:/etc/grafana/dashboards:ro + - grafana-data:/var/lib/grafana + depends_on: + - victoriametrics + - prometheus + restart: unless-stopped + volumes: vm-data: prom-data: + grafana-data: diff --git a/docker/dev-config.yaml b/docker/dev-config.yaml index dde2ed4..7ec339a 100644 --- a/docker/dev-config.yaml +++ b/docker/dev-config.yaml @@ -24,8 +24,17 @@ safety: query_timeout: 10s on_source_failure: serve_stale # v0.4: drive the query cache through the dev Redis so the - # anomaly_query_cache_* metrics reflect the real backend rather than - # only the in-process LRU. + # anomaly_query_cache_* metrics reflect the real backend (entries, + # misses, expirations, SCAN-based size) rather than only the in-process + # LRU. Expect anomaly_query_cache_hits_total to sit at ~0 here: the + # cache only collapses two fetches of the *same* PromQL within one step + # bucket, but the runner already fetches each query once per tick and + # shares it across that query's detectors, and refresh_interval (30s) > + # step (15s) so consecutive ticks land in different buckets. Hits would + # appear if two query ids shared an identical PromQL. The 0% hit rate is + # the workload, not a fault — the Redis backend itself is covered by + # tests/test_redis_cache.py. Redis I/O is offloaded off the event loop + # (see source.PromQLSource._cache_get), so it stays cheap even at 0%. query_cache: enabled: true backend: redis @@ -151,3 +160,122 @@ groups: label: cluster detectors: - name: MAD + + - name: http_demo + priority: 3 + # Only populated when the docker-compose ``load`` profile is up + # (``make dev-up-load``) — the http-demo app and its loadgen are + # behind that profile. With the minimum dev stack these queries + # return empty and the detector emits ``anomaly_signal_missing`` + # once ``expect_grace_runs`` is exceeded, which is the intended + # behaviour: it demonstrates the absence-detection path. + # + # That absence demo only fires for queries that set ``expect: true`` + # — ``anomaly_signal_missing`` is gated on it (see runner.py). The + # reliably-populated queries below opt in; ``http_errors_rate`` does + # NOT, because it is a deliberately sparse signal (errors are ~7% of + # ``/api/flaky`` traffic) and would false-fire ``signal_missing`` + # during quiet error windows even while the load profile is up. + # + # The loadgen modulates request rate on a 10-minute sine + # (PERIOD_SECONDS=600 in docker/http-demo/loadgen.py). MAD and + # Hampel rolling windows are sized at 15m to span more than one + # full period so the baseline tracks the cycle instead of being + # offset by it. + ensemble: + method: voting + min_detectors: 2 + queries: + # Per-endpoint request rate. The bursty, sinusoidal shape from + # the loadgen produces clear "score moves with the cycle" signal + # in /metrics, useful for eyeballing detector calibration. + - id: http_requests_rate + promql: | + sum by (endpoint) ( + rate(http_requests_total{endpoint=~"/api/.*|/healthz"}[2m]) + ) + expect: true + expect_grace_runs: 6 + detectors: + - name: MAD + params: + window: 15m + - name: Hampel + params: + window: 15m + t0: 3 + - name: ZScoreEWMA + params: + alpha: 0.1 + + # Error rate. ``/api/flaky`` returns 500 ~7% of the time so this + # is a sparse intermittent signal — exactly the shape MAD handles + # well and Gaussian assumptions break on. + - id: http_errors_rate + promql: | + sum by (endpoint) ( + rate(http_requests_total{status=~"5..",endpoint=~"/api/.*|/healthz"}[2m]) + ) + expect_grace_runs: 6 + detectors: + - name: MAD + params: + window: 15m + - name: IQR + params: + k: 1.5 + + # p95 latency derived from the histogram. Latency is noisy and + # tied to load, so a change-point detector earns its keep here. + # + # min_abs_delta is the practical-significance floor that makes this + # query usable. p95 from histogram_quantile snaps to bucket edges + # (0.005, 0.01, 0.025, 0.05, …), so a fast, stable endpoint like + # /api/users or /healthz sits on a near-constant ~25ms baseline with + # a tiny MAD/Hampel spread — a one-bucket quantization wobble then + # scores 10-100 "MADs" and fires outside_threshold on a + # sub-millisecond move (it flapped ~40% of runs without this floor). + # The 50ms floor suppresses those practically-irrelevant wobbles + # while still letting a real regression (25ms -> 100ms, or the slow + # endpoint's larger swings) fire. The raw anomaly_score is + # unaffected — only the alerting bit is gated — so dashboards still + # see the statistical signal. + - id: http_request_p95_seconds + promql: | + histogram_quantile( + 0.95, + sum by (endpoint, le) ( + rate(http_request_duration_seconds_bucket{endpoint=~"/api/.*|/healthz"}[2m]) + ) + ) + expect: true + expect_grace_runs: 6 + min_abs_delta: 0.05 + detectors: + - name: MAD + params: + window: 15m + - name: Hampel + params: + window: 15m + t0: 3 + - name: CUSUM + + # In-flight concurrency. Gauge, not derived from a counter, so + # a different shape of series than the rate-based ones above. + # Two detectors so the group's voting ensemble (min_detectors: 2) + # can actually fire for this query — a single-detector query under + # voting yields a composite that can never light up. ZScoreEWMA + # tracks the level of a smooth concurrency gauge, complementing + # MAD's robust spread estimate. + - id: http_in_flight + promql: max by (job) (http_requests_in_flight) + expect: true + expect_grace_runs: 6 + detectors: + - name: MAD + params: + window: 15m + - name: ZScoreEWMA + params: + alpha: 0.1 diff --git a/docker/grafana/provisioning/dashboards/dashboards.yml b/docker/grafana/provisioning/dashboards/dashboards.yml new file mode 100644 index 0000000..1e7c48d --- /dev/null +++ b/docker/grafana/provisioning/dashboards/dashboards.yml @@ -0,0 +1,13 @@ +apiVersion: 1 + +providers: + - name: promanomaly + orgId: 1 + folder: promanomaly + type: file + disableDeletion: false + editable: true + updateIntervalSeconds: 30 + options: + path: /etc/grafana/dashboards + foldersFromFilesStructure: false diff --git a/docker/grafana/provisioning/datasources/datasources.yml b/docker/grafana/provisioning/datasources/datasources.yml new file mode 100644 index 0000000..f10b869 --- /dev/null +++ b/docker/grafana/provisioning/datasources/datasources.yml @@ -0,0 +1,27 @@ +apiVersion: 1 + +datasources: + # VictoriaMetrics is the default: it holds everything Prometheus has + # (via remote_write, 30d retention) plus the detector's anomaly scores + # (scraped from /metrics). Querying Prometheus directly is only useful + # for debugging the scrape pipeline. + - name: VictoriaMetrics + type: prometheus + access: proxy + uid: victoriametrics + url: http://victoriametrics:8428 + isDefault: true + editable: true + jsonData: + httpMethod: POST + timeInterval: 15s + + - name: Prometheus + type: prometheus + access: proxy + uid: prometheus + url: http://prometheus:9090 + editable: true + jsonData: + httpMethod: POST + timeInterval: 15s diff --git a/docker/http-demo/Dockerfile.app b/docker/http-demo/Dockerfile.app new file mode 100644 index 0000000..db49130 --- /dev/null +++ b/docker/http-demo/Dockerfile.app @@ -0,0 +1,17 @@ +FROM python:3.12-slim + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + PIP_NO_CACHE_DIR=1 \ + PIP_DISABLE_PIP_VERSION_CHECK=1 + +WORKDIR /app +RUN pip install --no-cache-dir \ + "fastapi==0.115.0" \ + "uvicorn[standard]==0.30.6" \ + "prometheus-client==0.21.0" + +COPY app.py /app/app.py + +EXPOSE 8000 +CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000", "--no-access-log"] diff --git a/docker/http-demo/Dockerfile.loadgen b/docker/http-demo/Dockerfile.loadgen new file mode 100644 index 0000000..b34db7e --- /dev/null +++ b/docker/http-demo/Dockerfile.loadgen @@ -0,0 +1,13 @@ +FROM python:3.12-slim + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + PIP_NO_CACHE_DIR=1 \ + PIP_DISABLE_PIP_VERSION_CHECK=1 + +WORKDIR /app +RUN pip install --no-cache-dir "httpx==0.27.2" + +COPY loadgen.py /app/loadgen.py + +CMD ["python", "/app/loadgen.py"] diff --git a/docker/http-demo/app.py b/docker/http-demo/app.py new file mode 100644 index 0000000..e35d534 --- /dev/null +++ b/docker/http-demo/app.py @@ -0,0 +1,157 @@ +# Tiny FastAPI app for the dev stack: a handful of HTTP endpoints with varied +# latency / error profiles and a /metrics endpoint that exposes prometheus_client +# counters/histograms/gauges. Driven by docker/http-demo/loadgen.py so the +# detector has http_* time series to score on. +from __future__ import annotations + +import asyncio +import os +import random +import time + +from fastapi import FastAPI, Request, Response +from fastapi.responses import JSONResponse, PlainTextResponse +from prometheus_client import ( + CONTENT_TYPE_LATEST, + CollectorRegistry, + Counter, + Gauge, + Histogram, + generate_latest, +) + +registry = CollectorRegistry() + +http_requests_total = Counter( + "http_requests_total", + "Total HTTP requests handled.", + ["method", "endpoint", "status"], + registry=registry, +) +http_request_duration_seconds = Histogram( + "http_request_duration_seconds", + "HTTP request latency in seconds.", + ["method", "endpoint"], + buckets=(0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0), + registry=registry, +) +http_requests_in_flight = Gauge( + "http_requests_in_flight", + "In-flight HTTP requests.", + registry=registry, +) +http_payload_bytes = Histogram( + "http_payload_bytes", + "Response payload size in bytes.", + ["endpoint"], + buckets=(64, 256, 1024, 4096, 16384, 65536), + registry=registry, +) + +APP_NAME = os.getenv("APP_NAME", "http-demo") +app = FastAPI(title=APP_NAME) + + +@app.middleware("http") +async def observe(request: Request, call_next): + endpoint = ( + request.scope.get("route").path + if request.scope.get("route") + else request.url.path + ) + method = request.method + http_requests_in_flight.inc() + start = time.perf_counter() + try: + response: Response = await call_next(request) + status = response.status_code + except Exception: + status = 500 + raise + finally: + elapsed = time.perf_counter() - start + http_requests_in_flight.dec() + http_request_duration_seconds.labels(method=method, endpoint=endpoint).observe( + elapsed + ) + http_requests_total.labels( + method=method, endpoint=endpoint, status=str(status) + ).inc() + return response + + +def _sleep_lognormal(mu: float, sigma: float, cap: float) -> float: + delay = min(random.lognormvariate(mu, sigma), cap) + return delay + + +@app.get("/") +async def root() -> dict: + body = { + "app": APP_NAME, + "endpoints": [ + "/", + "/api/users", + "/api/products", + "/api/slow", + "/api/flaky", + "/healthz", + ], + } + http_payload_bytes.labels(endpoint="/").observe(len(str(body))) + return body + + +@app.get("/api/users") +async def users() -> dict: + # Fast endpoint, ~5-20ms. + await asyncio.sleep(_sleep_lognormal(mu=-4.5, sigma=0.4, cap=0.2)) + payload = { + "users": [{"id": i, "name": f"user-{i}"} for i in range(random.randint(5, 25))] + } + http_payload_bytes.labels(endpoint="/api/users").observe(len(str(payload))) + return payload + + +@app.get("/api/products") +async def products() -> dict: + # Medium endpoint, ~30-150ms with occasional spikes. + await asyncio.sleep(_sleep_lognormal(mu=-3.0, sigma=0.6, cap=1.5)) + payload = { + "products": [ + {"sku": f"sku-{i}", "price": round(random.uniform(1, 99), 2)} + for i in range(20) + ] + } + http_payload_bytes.labels(endpoint="/api/products").observe(len(str(payload))) + return payload + + +@app.get("/api/slow") +async def slow() -> dict: + # Slow endpoint, ~0.3-2s. Useful for histogram_quantile to have something + # to show. + await asyncio.sleep(_sleep_lognormal(mu=-1.0, sigma=0.5, cap=3.0)) + return {"status": "ok"} + + +@app.get("/api/flaky") +async def flaky() -> Response: + # ~7% of calls return 500. Gives http_requests_total{status="500"} a non-zero + # rate so the detector has an error series to score. + await asyncio.sleep(_sleep_lognormal(mu=-3.5, sigma=0.5, cap=1.0)) + if random.random() < 0.07: + return JSONResponse({"error": "transient"}, status_code=500) + return JSONResponse({"status": "ok"}) + + +@app.get("/healthz") +async def healthz() -> dict: + return {"status": "ok"} + + +@app.get("/metrics") +async def metrics() -> PlainTextResponse: + return PlainTextResponse( + generate_latest(registry).decode("utf-8"), media_type=CONTENT_TYPE_LATEST + ) diff --git a/docker/http-demo/loadgen.py b/docker/http-demo/loadgen.py new file mode 100644 index 0000000..997777b --- /dev/null +++ b/docker/http-demo/loadgen.py @@ -0,0 +1,55 @@ +# Tiny load generator: fires ~20-30 req/min by default at the http-demo app, +# with a slow sinusoidal modulation on top so the rate has shape (and the +# detector has a non-trivial signal to score). Endpoints are picked from a +# weighted mix so http_requests_total{endpoint=...} differs per route. +from __future__ import annotations + +import asyncio +import math +import os +import random +import time +from contextlib import suppress + +import httpx + +TARGET = os.getenv("TARGET", "http://http-demo:8000") +BASE_RPM = float(os.getenv("BASE_RPM", "25")) +AMPLITUDE_RPM = float(os.getenv("AMPLITUDE_RPM", "15")) +PERIOD_SECONDS = float(os.getenv("PERIOD_SECONDS", "600")) # 10-minute cycle +JITTER = float(os.getenv("JITTER", "0.25")) + +ENDPOINTS = [ + ("/api/users", 5), + ("/api/products", 3), + ("/api/slow", 1), + ("/api/flaky", 2), + ("/healthz", 1), +] + + +def pick_endpoint() -> str: + paths, weights = zip(*ENDPOINTS, strict=True) + return random.choices(paths, weights=weights, k=1)[0] + + +def current_rpm(t0: float) -> float: + phase = 2 * math.pi * ((time.monotonic() - t0) % PERIOD_SECONDS) / PERIOD_SECONDS + return max(1.0, BASE_RPM + AMPLITUDE_RPM * math.sin(phase)) + + +async def hammer() -> None: + t0 = time.monotonic() + timeout = httpx.Timeout(5.0) + async with httpx.AsyncClient(timeout=timeout, base_url=TARGET) as client: + while True: + rpm = current_rpm(t0) + interval = 60.0 / rpm + interval *= 1.0 + random.uniform(-JITTER, JITTER) + with suppress(Exception): + await client.get(pick_endpoint()) + await asyncio.sleep(max(0.05, interval)) + + +if __name__ == "__main__": + asyncio.run(hammer()) diff --git a/docker/prometheus.yml b/docker/prometheus.yml index 11784fc..bb99b98 100644 --- a/docker/prometheus.yml +++ b/docker/prometheus.yml @@ -27,6 +27,13 @@ scrape_configs: static_configs: - targets: ["node-exporter:9100"] + - job_name: http-demo + # Started only under the docker-compose ``load`` profile (see + # docker-compose.dev.yml). Targets resolve to nothing on the minimum + # dev stack, which Prometheus tolerates by leaving the scrape pool empty. + static_configs: + - targets: ["http-demo:8000"] + - job_name: promanomaly # honor_labels: anomaly metrics carry the source ``instance`` label # (e.g. ``dev-host-a:9100``) from the underlying time-series. Without diff --git a/docs/auto-select.md b/docs/auto-select.md index cdd707f..a6ae3b2 100644 --- a/docs/auto-select.md +++ b/docs/auto-select.md @@ -33,8 +33,8 @@ groups: | Option | Behaviour | |-------------------------------|-----------| | `false` (default) | Every detector emits its own score | -| `true` | Only the winner emits; carries `best_detector=` | -| `"explicit_plus_best"` | All detectors emit + winner is tagged (great for tuning) | +| `true` | Only the winner emits; the winner is named by `anomaly_best_detector` | +| `"explicit_plus_best"` | All detectors emit + winner named by `anomaly_best_detector` (great for tuning) | ## How It Works @@ -48,13 +48,29 @@ Highest F1 wins. Ties broken by precision, then by order in the config. ## Output -When `auto_select: true`, only the winner’s metrics appear: +When `auto_select: true`, only the winner’s metrics appear, and a separate +`anomaly_best_detector` info gauge names the winner: ``` -anomaly_score{..., detector="MAD", best_detector="MAD", ...} 4.7 +anomaly_score{..., detector="MAD", ...} 4.7 +anomaly_best_detector{..., detector="MAD", ...} 1 ``` -The `best_detector` label is only present when auto-select is enabled. +`anomaly_best_detector=1` is emitted only for the winner, only when +auto-select is enabled. It is a standalone series rather than a label on +`anomaly_score`/`anomaly_outside_threshold`/`anomaly_change_point_total` +**on purpose**: the winner can change between calibration cycles, and a +label that appears and disappears would churn those series' identity — +breaking `rate()`/`increase()` on the change-point counter and resetting +`for:` timers on `outside_threshold` alerts. Keeping the marker on its own +series leaves every per-detector metric with a stable label set. + +To filter the per-detector metrics down to the winner, join on the shared +labels, e.g.: + +```promql +anomaly_score * on (id, group, detector, instance) group_left anomaly_best_detector +``` ## Inspect Before Switching diff --git a/docs/detectors.md b/docs/detectors.md index fad5340..a9d4537 100644 --- a/docs/detectors.md +++ b/docs/detectors.md @@ -639,7 +639,7 @@ groups: - name: IQR ``` -When `auto_select: true`, the runner periodically injects three canonical synthetic anomalies (point spike, step change, level drift) into each series's rolling window and measures every configured detector's F1. The winner's score samples are emitted with an extra `best_detector=` label; the losers are suppressed. `auto_select: "explicit_plus_best"` keeps every detector's output and only tags the winner — useful while tuning. +When `auto_select: true`, the runner periodically injects three canonical synthetic anomalies (point spike, step change, level drift) into each series's rolling window and measures every configured detector's F1. The winner's score samples are emitted and a standalone `anomaly_best_detector{...,detector=} = 1` info gauge names the winner; the losers are suppressed. `auto_select: "explicit_plus_best"` keeps every detector's output and still names the winner via `anomaly_best_detector` — useful while tuning. The marker is a separate series rather than a label on the per-detector metrics, so the winner can flip between cycles without churning the identity of `anomaly_score`/`anomaly_outside_threshold`/`anomaly_change_point_total`. The full algorithm and tuning knobs live in [`auto-select.md`](auto-select.md). diff --git a/docs/operations.md b/docs/operations.md index 16050d0..acd3f9f 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -386,7 +386,7 @@ exporter: Semantics: - The canonical detector labels — `id`, `group`, `detector`, `detector_instance` — are added **after** the filter and cannot be stripped. Listing one of them in `drop` is rejected at config-load time so a configuration mistake fails at validate-time rather than silently leaving the label in production. -- `allow` and `drop` operate on **source** labels (everything copied from the PromQL response). The canonical labels and any opt-in feature labels (`best_detector`, `warming_up`) bypass the filter. +- `allow` and `drop` operate on **source** labels (everything copied from the PromQL response). The canonical labels and any opt-in feature labels (`cohort_label`, `warming_up`) bypass the filter. (The auto-select winner is no longer a label — it is the standalone `anomaly_best_detector` info gauge.) - `drop` takes precedence over `allow` — if a label appears in both, it is stripped. Useful when the allow-list is an inherited template and a downstream config layer needs to subtract. - The filter applies uniformly to `anomaly_score`, `anomaly_baseline`, `anomaly_duration_seconds`, `anomaly_change_point_total`, `anomaly_confidence_score`, `anomaly_baseline_stability`, and the composite `anomaly_composite_*` metrics — anywhere source labels could otherwise reach the wire.