From aa493d01a793dccbc19e1ff4ef92718fb66fd3a5 Mon Sep 17 00:00:00 2001 From: ali <117142933+muhammad-ali-e@users.noreply.github.com> Date: Tue, 27 Oct 2026 07:18:58 +0000 Subject: [PATCH 1/3] UN-4251 Expose Prometheus platform metrics from the backend at /metrics Adds a platform_metrics app serving GET /metrics in Prometheus text format: queue depth by state and oldest-message age (pg_queue_message), active and recently-terminal execution counts (workflow_execution), in-progress and stuck batches (pg_barrier_state), plus last-refresh and refresh-failure metrics per group. Known queues and statuses are zero-filled. Each group's snapshot is cached in Redis for 15s behind a non-blocking lock, so one backend pod refreshes a group per interval however many are scraped. A failing group serves its last values and counts the failure without affecting the others; with Redis down each pod refreshes at most once per TTL. Mounted at the URL root next to /health, outside every public ingress prefix, with an optional bearer token (PLATFORM_METRICS_TOKEN). Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_016Hov58G5waUqpMHz9P6FmL --- backend/backend/base_urls.py | 1 + backend/backend/settings/base.py | 67 +++++ backend/platform_metrics/__init__.py | 0 backend/platform_metrics/apps.py | 5 + backend/platform_metrics/cache.py | 210 +++++++++++++++ backend/platform_metrics/exporter.py | 127 +++++++++ backend/platform_metrics/groups.py | 254 ++++++++++++++++++ backend/platform_metrics/tests/__init__.py | 0 backend/platform_metrics/tests/fakes.py | 91 +++++++ backend/platform_metrics/tests/test_cache.py | 198 ++++++++++++++ .../platform_metrics/tests/test_exporter.py | 120 +++++++++ backend/platform_metrics/tests/test_groups.py | 94 +++++++ .../platform_metrics/tests/test_groups_db.py | 157 +++++++++++ backend/platform_metrics/tests/test_views.py | 60 +++++ backend/platform_metrics/urls.py | 5 + backend/platform_metrics/views.py | 51 ++++ backend/pyproject.toml | 1 + backend/uv.lock | 11 + 18 files changed, 1452 insertions(+) create mode 100644 backend/platform_metrics/__init__.py create mode 100644 backend/platform_metrics/apps.py create mode 100644 backend/platform_metrics/cache.py create mode 100644 backend/platform_metrics/exporter.py create mode 100644 backend/platform_metrics/groups.py create mode 100644 backend/platform_metrics/tests/__init__.py create mode 100644 backend/platform_metrics/tests/fakes.py create mode 100644 backend/platform_metrics/tests/test_cache.py create mode 100644 backend/platform_metrics/tests/test_exporter.py create mode 100644 backend/platform_metrics/tests/test_groups.py create mode 100644 backend/platform_metrics/tests/test_groups_db.py create mode 100644 backend/platform_metrics/tests/test_views.py create mode 100644 backend/platform_metrics/urls.py create mode 100644 backend/platform_metrics/views.py diff --git a/backend/backend/base_urls.py b/backend/backend/base_urls.py index 30d5d790da..043804392e 100644 --- a/backend/backend/base_urls.py +++ b/backend/backend/base_urls.py @@ -40,6 +40,7 @@ # Agent-KV product API (views own auth; prefix is middleware-whitelisted) path(f"{settings.AGENT_KV_PATH_PREFIX}/", include("agent_kv.execution_urls")), path("", include("health.urls")), + path("", include("platform_metrics.urls")), # Internal API for worker communication path("internal/", include("backend.internal_base_urls")), ] diff --git a/backend/backend/settings/base.py b/backend/backend/settings/base.py index 76f8f4b050..480f8f335a 100644 --- a/backend/backend/settings/base.py +++ b/backend/backend/settings/base.py @@ -572,6 +572,8 @@ def filter(self, record): "commands", # health checks "health", + # Platform-wide Prometheus /metrics + "platform_metrics", "migrating.v2", "connector_auth_v2", "tenant_account_v2", @@ -1144,6 +1146,71 @@ def filter(self, record): # Whitelisting health check API WHITELISTED_PATHS.append("/health") +# Platform-wide Prometheus /metrics (platform_metrics). Mounted at the root +# like /health, outside every public ingress prefix, so it is served on the +# backend service for in-cluster scraping only. Exempt from session auth; set +# PLATFORM_METRICS_TOKEN to additionally require `Authorization: Bearer `. +WHITELISTED_PATHS.append("/metrics") +PLATFORM_METRICS_TOKEN = os.environ.get("PLATFORM_METRICS_TOKEN", "") +# How long one pod's refresh of a metric group is shared via Redis. +PLATFORM_METRICS_CACHE_TTL_SECONDS = int( + os.environ.get("PLATFORM_METRICS_CACHE_TTL_SECONDS", 15) +) +# Postgres statement timeout for each group's refresh. +PLATFORM_METRICS_QUERY_TIMEOUT_MS = int( + os.environ.get("PLATFORM_METRICS_QUERY_TIMEOUT_MS", 5000) +) +# unstract_workflow_executions_recent counts executions that went terminal +# within this window... +PLATFORM_METRICS_RECENT_WINDOW_SECONDS = int( + os.environ.get("PLATFORM_METRICS_RECENT_WINDOW_SECONDS", 3600) +) +# ...among executions created within this lookback, which keeps the query on +# the created_at index instead of scanning the whole history. +PLATFORM_METRICS_RECENT_LOOKBACK_SECONDS = int( + os.environ.get("PLATFORM_METRICS_RECENT_LOOKBACK_SECONDS", 86400) +) +# Same variable and default as the PG reaper, so unstract_batches_stuck counts +# exactly what the reaper treats as stranded. +PLATFORM_METRICS_BATCH_STUCK_TIMEOUT_SECONDS = int( + os.environ.get("WORKER_PG_BATCH_STUCK_TIMEOUT_SECONDS", 9000) +) +# Queues always reported (as 0 when empty), on top of any queue that currently +# holds messages. Defaults to the queues the PG consumers in +# docker/docker-compose.yaml consume; override with a comma-separated list +# where a deployment runs others. +PLATFORM_METRICS_QUEUES = [ + queue.strip() + for queue in os.environ.get( + "PLATFORM_METRICS_QUEUES", + ",".join( + [ + "agent_kv_callback", + "api_file_processing", + "api_file_processing_callback", + "celery", + "celery_api_deployments", + "celery_callback", + "celery_executor_agentic", + "celery_executor_agentic_table", + "celery_executor_legacy", + "celery_executor_lookup_test", + "celery_executor_simple_prompt_studio", + "celery_executor_smart_table", + "celery_executor_table", + "dashboard_metric_events", + "file_processing", + "file_processing_callback", + "file_processing_priority", + "ide_callback", + "notifications", + "scheduler", + ] + ), + ).split(",") + if queue.strip() +] + # These path will work without organization in request. # `whoami` resolves the organisation from the API key itself, so it carries no # organisation segment -- without this the middleware would read `whoami` as one. diff --git a/backend/platform_metrics/__init__.py b/backend/platform_metrics/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/backend/platform_metrics/apps.py b/backend/platform_metrics/apps.py new file mode 100644 index 0000000000..8f2d47aa8e --- /dev/null +++ b/backend/platform_metrics/apps.py @@ -0,0 +1,5 @@ +from django.apps import AppConfig + + +class PlatformMetricsConfig(AppConfig): + name = "platform_metrics" diff --git a/backend/platform_metrics/cache.py b/backend/platform_metrics/cache.py new file mode 100644 index 0000000000..731cf1f66e --- /dev/null +++ b/backend/platform_metrics/cache.py @@ -0,0 +1,210 @@ +"""Shared, lock-guarded snapshot cache for the platform metric groups. + +Every backend pod serves ``/metrics``, and Prometheus scrapes each of them, so +without coordination the DB would run every group's queries once per pod per +scrape. Instead each group's snapshot lives in Redis for +``PLATFORM_METRICS_CACHE_TTL_SECONDS`` (default 15s), and a non-blocking Redis +lock lets exactly one pod refresh a stale group while the others keep serving +the previous snapshot. DB cost is about one refresh per group per TTL however +many replicas there are. + +Failure handling, per group and independent of the others: + +* the refresh query fails: serve the last snapshot (Redis, else this + process's copy) and count the failure; +* Redis is unreachable: fall back to this process's copy, refreshing it from + the DB at most once per TTL, so DB load stays bounded per pod rather than + per scrape. The failure is counted; +* nothing has ever been fetched (first scrape, while another pod holds the + lock): the group is left out of this scrape. +""" + +from __future__ import annotations + +import json +import logging +import threading +import time +from collections.abc import Callable +from dataclasses import dataclass +from typing import Any, Protocol + +from django.conf import settings + +from .groups import MetricGroup, Snapshot + +logger = logging.getLogger(__name__) + +_KEY_PREFIX = "platform_metrics:v1" +#: How long a snapshot stays in Redis. Far beyond the TTL on purpose: it is the +#: last-known-good value served while refreshes fail. +_RETENTION_SECONDS = 3600 +#: Lock expiry. Above the statement timeout, so a pod that dies mid-refresh +#: frees the group within one window instead of wedging it. +_LOCK_TIMEOUT_SECONDS = 30 + + +class _Lock(Protocol): + def acquire(self, blocking: bool = ...) -> bool: ... + def release(self) -> None: ... + + +class RedisLike(Protocol): + def get(self, name: str) -> Any: ... + def set(self, name: str, value: str, ex: int | None = ...) -> Any: ... + def lock(self, name: str, timeout: float | None = ...) -> _Lock: ... + + +@dataclass(frozen=True) +class CachedSnapshot: + data: Snapshot + #: Unix time of the successful refresh that produced ``data``. + refreshed_at: float + + def age(self, now: float) -> float: + return now - self.refreshed_at + + +class RefreshFailed(Exception): + """A group could not be refreshed. ``snapshot`` is what to serve instead.""" + + def __init__(self, snapshot: CachedSnapshot | None) -> None: + super().__init__("platform metrics refresh failed") + self.snapshot = snapshot + + +def _default_client() -> RedisLike: + from django_redis import get_redis_connection + + return get_redis_connection("default") + + +class SnapshotCache: + """Resolves a group to its current snapshot, refreshing when stale.""" + + def __init__( + self, + client_factory: Callable[[], RedisLike] = _default_client, + clock: Callable[[], float] = time.time, + ) -> None: + self._client_factory = client_factory + self._clock = clock + # Per-process last-known-good copies: the fallback when Redis is down. + self._local: dict[str, CachedSnapshot] = {} + self._local_lock = threading.Lock() + + @staticmethod + def _ttl() -> float: + return float(settings.PLATFORM_METRICS_CACHE_TTL_SECONDS) + + def resolve(self, group: MetricGroup) -> CachedSnapshot | None: + """Return the snapshot to serve for ``group``. + + Raises :class:`RefreshFailed` (carrying the snapshot to serve anyway) + when this call tried and failed to refresh the group. + """ + try: + client = self._client_factory() + cached = self._read(client, group) + except Exception: + logger.warning( + "platform metrics: Redis unavailable for group %r; using the " + "process-local snapshot", + group.name, + exc_info=True, + ) + return self._resolve_local(group) + + if cached is not None and cached.age(self._clock()) < self._ttl(): + return cached + + try: + lock = client.lock(self._lock_key(group), timeout=_LOCK_TIMEOUT_SECONDS) + acquired = lock.acquire(blocking=False) + except Exception: + logger.warning( + "platform metrics: could not take the refresh lock for %r", + group.name, + exc_info=True, + ) + return self._resolve_local(group, fallback=cached) + if not acquired: + # Another pod is refreshing; its result lands for the next scrape. + return cached or self._local.get(group.name) + + try: + # Re-read under the lock: another pod may have refreshed between + # our read and our acquire. + latest = self._read(client, group) + if latest is not None and latest.age(self._clock()) < self._ttl(): + return latest + snapshot = self._refresh(group, fallback=latest or cached) + try: + client.set( + self._data_key(group), + json.dumps( + {"data": snapshot.data, "refreshed_at": snapshot.refreshed_at} + ), + ex=_RETENTION_SECONDS, + ) + except Exception: + # The fresh snapshot is still good to serve from this pod. + logger.warning( + "platform metrics: could not cache group %r in Redis", + group.name, + exc_info=True, + ) + return snapshot + finally: + try: + lock.release() + except Exception: + # Expired under us (a refresh slower than the lock timeout): + # nothing to release. + logger.debug("platform metrics: lock for %r already gone", group.name) + + def _resolve_local( + self, group: MetricGroup, fallback: CachedSnapshot | None = None + ) -> CachedSnapshot | None: + """Redis-less path: refresh this process's copy at most once per TTL. + + Always raises :class:`RefreshFailed`: every scrape that finds Redis + unusable counts as a failure for the group, even when the snapshot it + serves is fresh, so a Redis outage shows on the failure counter. + """ + with self._local_lock: + local = self._local.get(group.name) + if local is not None and local.age(self._clock()) < self._ttl(): + raise RefreshFailed(local) + raise RefreshFailed(self._refresh(group, fallback=local or fallback)) + + def _refresh( + self, group: MetricGroup, fallback: CachedSnapshot | None + ) -> CachedSnapshot: + try: + data = group.fetch() + except Exception as e: + logger.exception("platform metrics: refreshing group %r failed", group.name) + raise RefreshFailed(fallback or self._local.get(group.name)) from e + snapshot = CachedSnapshot(data=data, refreshed_at=self._clock()) + self._local[group.name] = snapshot + return snapshot + + def _read(self, client: RedisLike, group: MetricGroup) -> CachedSnapshot | None: + raw = client.get(self._data_key(group)) + if raw is None: + return None + payload = json.loads(raw) + snapshot = CachedSnapshot( + data=payload["data"], refreshed_at=float(payload["refreshed_at"]) + ) + self._local[group.name] = snapshot + return snapshot + + @staticmethod + def _data_key(group: MetricGroup) -> str: + return f"{_KEY_PREFIX}:data:{group.name}" + + @staticmethod + def _lock_key(group: MetricGroup) -> str: + return f"{_KEY_PREFIX}:lock:{group.name}" diff --git a/backend/platform_metrics/exporter.py b/backend/platform_metrics/exporter.py new file mode 100644 index 0000000000..4a6ee23a74 --- /dev/null +++ b/backend/platform_metrics/exporter.py @@ -0,0 +1,127 @@ +"""Assembles the platform metric groups into one Prometheus exposition. + +One process-wide registry holds two things: the refresh-failure counter, which +must survive across scrapes, and a collector that resolves every group through +the shared :class:`~platform_metrics.cache.SnapshotCache` at scrape time. Each +group is resolved inside its own ``try``, so one failing group never blanks the +others and the scrape itself always succeeds. + +The registry is instance-owned rather than ``prometheus_client``'s default +``REGISTRY``, so nothing else in the process can leak series into this +endpoint and building it twice (tests) cannot double-register. +""" + +from __future__ import annotations + +import logging +import threading +from collections.abc import Iterable, Sequence +from typing import TYPE_CHECKING + +from .cache import RefreshFailed, SnapshotCache +from .groups import METRIC_GROUPS, MetricGroup + +if TYPE_CHECKING: + from prometheus_client import CollectorRegistry, Counter + from prometheus_client.core import Metric + +logger = logging.getLogger(__name__) + + +class _GroupsCollector: + def __init__( + self, + groups: Sequence[MetricGroup], + cache: SnapshotCache, + failures: Counter, + ) -> None: + self._groups = groups + self._cache = cache + self._failures = failures + + def describe(self) -> Iterable[Metric]: + # Present so register() does not run collect() (and the DB) on import. + return [] + + def collect(self) -> Iterable[Metric]: + from prometheus_client.core import GaugeMetricFamily + + refreshed = GaugeMetricFamily( + "unstract_metrics_last_successful_refresh_timestamp", + "Unix time of the last successful refresh of each metric group " + "(alert on time() minus this)", + labels=["group"], + ) + families: list[Metric] = [] + for group in self._groups: + try: + snapshot = self._cache.resolve(group) + except RefreshFailed as failed: + self._failures.labels(group=group.name).inc() + snapshot = failed.snapshot + except Exception: + logger.exception("platform metrics: group %r failed", group.name) + self._failures.labels(group=group.name).inc() + snapshot = None + if snapshot is None: + continue + try: + families.extend(group.families(snapshot.data)) + except Exception: + # A cached snapshot that no longer fits the group (e.g. written + # by an older pod mid-deploy) is dropped, not served half-built. + logger.exception( + "platform metrics: rendering group %r failed", group.name + ) + self._failures.labels(group=group.name).inc() + continue + refreshed.add_metric([group.name], snapshot.refreshed_at) + families.append(refreshed) + return families + + +class PlatformMetricsExporter: + def __init__( + self, + groups: Sequence[MetricGroup] = METRIC_GROUPS, + cache: SnapshotCache | None = None, + ) -> None: + from prometheus_client import CollectorRegistry, Counter + + self.registry: CollectorRegistry = CollectorRegistry() + self.failures = Counter( + "unstract_metrics_refresh_failures", + "Failed refreshes per metric group; the group serves its last " + "cached values meanwhile", + ["group"], + registry=None, + ) + for group in groups: + # Pre-create every series so a group that has never failed reads 0. + self.failures.labels(group=group.name) + # Order matters: a registry collects in registration order, and the + # groups collector is what increments the counter, so it goes first + # for a failure to show in the same scrape that hit it. + self.registry.register( + _GroupsCollector(groups, cache or SnapshotCache(), self.failures) + ) + self.registry.register(self.failures) + + def render(self) -> bytes: + from prometheus_client import generate_latest + + return generate_latest(self.registry) + + +_exporter: PlatformMetricsExporter | None = None +_exporter_lock = threading.Lock() + + +def get_exporter() -> PlatformMetricsExporter: + """The process-wide exporter, built on first scrape.""" + global _exporter + if _exporter is None: + with _exporter_lock: + if _exporter is None: + _exporter = PlatformMetricsExporter() + return _exporter diff --git a/backend/platform_metrics/groups.py b/backend/platform_metrics/groups.py new file mode 100644 index 0000000000..ccd3481407 --- /dev/null +++ b/backend/platform_metrics/groups.py @@ -0,0 +1,254 @@ +"""Metric groups for the platform ``/metrics`` endpoint. + +A group is one unit of refresh: ``fetch`` reads the database and returns a +JSON-safe snapshot, ``families`` turns a snapshot into Prometheus metric +families. The split is what lets the exporter cache a group's snapshot in Redis +(see ``cache.py``) and render it on any backend pod without touching the DB. + +Every group zero-fills its known label values, so a drained queue or an idle +status reads 0 instead of disappearing from the scrape. + +Adding a group: subclass :class:`MetricGroup` and append an instance to +:data:`METRIC_GROUPS`. The endpoint, caching and failure accounting pick it up +without further changes. +""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from collections.abc import Iterable +from datetime import timedelta +from typing import TYPE_CHECKING, Any + +from django.conf import settings +from django.db import connection, transaction +from django.db.models import Count, DurationField, ExpressionWrapper, Min, Q +from django.db.models.functions import Now +from django.utils import timezone + +from unstract.core.data_models import ExecutionStatus, QueueMessageState + +if TYPE_CHECKING: + from prometheus_client.core import Metric + +Snapshot = dict[str, Any] + +_ACTIVE_STATUSES = (ExecutionStatus.PENDING.value, ExecutionStatus.EXECUTING.value) +# Sorted so the series order in the scrape is stable. +_TERMINAL_STATUSES = tuple(sorted(ExecutionStatus.terminal_values())) +_QUEUE_STATES = tuple(state.value for state in QueueMessageState) + + +class MetricGroup(ABC): + """One independently cached and refreshed set of metrics.""" + + #: Label value for this group on the exporter-health metrics, and the + #: suffix of its Redis cache / lock keys. + name: str + + def fetch(self) -> Snapshot: + """Read a fresh snapshot, bounded by the configured statement timeout. + + ``SET LOCAL`` scopes the timeout to this transaction, so a slow group + is cancelled by Postgres instead of holding a scrape (and the group's + refresh lock) open, and nothing leaks onto the pooled connection. + """ + with transaction.atomic(), connection.cursor() as cursor: + cursor.execute( + "SET LOCAL statement_timeout = %s", + [int(settings.PLATFORM_METRICS_QUERY_TIMEOUT_MS)], + ) + return self.query() + + @abstractmethod + def query(self) -> Snapshot: + """Run the group's queries. Must return JSON-serialisable data.""" + + @abstractmethod + def families(self, snapshot: Snapshot) -> Iterable[Metric]: + """Build the group's metric families from a snapshot.""" + + +class QueueMetricGroup(MetricGroup): + """``pg_queue_message`` depth by state and oldest-message age, per queue. + + Same population as the reaper's ``pg_queue_depth`` / + ``pg_queue_oldest_message_age_seconds`` (``refresh_queue_gauges`` in + ``workers/queue_backend/pg_queue/reaper.py``): every row counts, and the age + is ``now() - min(enqueued_at)`` across all states. The per-state counts sum + to the reaper's depth; this group only adds the split. + + One aggregate over the table, like the reaper's. The table holds only + undelivered and in-flight messages (acked rows are deleted), so it stays + small whatever the platform's history. + """ + + name = "queues" + + def query(self) -> Snapshot: + from pg_queue.models import PgQueueMessage + + rows = ( + PgQueueMessage.objects.values("queue_name", "state") + .annotate( + messages=Count("msg_id"), + oldest_age=ExpressionWrapper( + Now() - Min("enqueued_at"), output_field=DurationField() + ), + ) + .order_by() + ) + queues: dict[str, dict[str, Any]] = {} + for row in rows: + entry = queues.setdefault(row["queue_name"], {"states": {}, "oldest": 0.0}) + entry["states"][row["state"]] = row["messages"] + age = row["oldest_age"].total_seconds() if row["oldest_age"] else 0.0 + entry["oldest"] = max(entry["oldest"], max(age, 0.0)) + return {"queues": queues} + + def families(self, snapshot: Snapshot) -> Iterable[Metric]: + from prometheus_client.core import GaugeMetricFamily + + observed: dict[str, dict[str, Any]] = snapshot["queues"] + names = sorted(set(settings.PLATFORM_METRICS_QUEUES) | set(observed)) + messages = GaugeMetricFamily( + "unstract_queue_messages", + "Messages in pg_queue_message by queue and claim state (ready = " + "waiting, claimed = held by a consumer, scheduled = delayed until " + "available_at)", + labels=["queue", "state"], + ) + oldest = GaugeMetricFamily( + "unstract_queue_oldest_message_age_seconds", + "Age of the oldest message in the queue, any state (0 when empty)", + labels=["queue"], + ) + for queue in names: + entry = observed.get(queue, {"states": {}, "oldest": 0.0}) + for state in _QUEUE_STATES: + messages.add_metric([queue, state], entry["states"].get(state, 0)) + oldest.add_metric([queue], entry["oldest"]) + return [messages, oldest] + + +class ExecutionMetricGroup(MetricGroup): + """``workflow_execution`` counts: active now, and terminal within a window. + + Both queries are index-bounded so they never scan the whole history: + + * active (``PENDING`` / ``EXECUTING``) is implied by the predicate of the + partial ``we_active_by_workflow_idx``, so it costs O(active executions); + * recent terminal is bounded by ``created_at`` (``we_created_at_idx``) to + ``PLATFORM_METRICS_RECENT_LOOKBACK_SECONDS``, then filtered on + ``modified_at`` — the time the row last changed, i.e. when it went + terminal. ``modified_at`` is deliberately not indexed (it is rewritten on + every status update), so an execution that ran longer than the lookback + before finishing is not counted. The default lookback (24h) is far above + any normal run. + """ + + name = "executions" + + def query(self) -> Snapshot: + from workflow_manager.workflow_v2.models.execution import WorkflowExecution + + now = timezone.now() + window = timedelta(seconds=int(settings.PLATFORM_METRICS_RECENT_WINDOW_SECONDS)) + lookback = timedelta( + seconds=int(settings.PLATFORM_METRICS_RECENT_LOOKBACK_SECONDS) + ) + active = ( + WorkflowExecution.objects.filter(status__in=_ACTIVE_STATUSES) + .values("status") + .annotate(n=Count("id")) + .order_by() + ) + recent = ( + WorkflowExecution.objects.filter( + status__in=_TERMINAL_STATUSES, + created_at__gte=now - max(lookback, window), + modified_at__gte=now - window, + ) + .values("status") + .annotate(n=Count("id")) + .order_by() + ) + return { + "active": {row["status"]: row["n"] for row in active}, + "recent": {row["status"]: row["n"] for row in recent}, + } + + def families(self, snapshot: Snapshot) -> Iterable[Metric]: + from prometheus_client.core import GaugeMetricFamily + + active = GaugeMetricFamily( + "unstract_workflow_executions", + "Workflow executions currently in a non-terminal status", + labels=["status"], + ) + for status in _ACTIVE_STATUSES: + active.add_metric([status.lower()], snapshot["active"].get(status, 0)) + recent = GaugeMetricFamily( + "unstract_workflow_executions_recent", + "Workflow executions that reached a terminal status within the " + "recent window (PLATFORM_METRICS_RECENT_WINDOW_SECONDS, default 1h)", + labels=["status"], + ) + for status in _TERMINAL_STATUSES: + recent.add_metric([status.lower()], snapshot["recent"].get(status, 0)) + return [active, recent] + + +class BatchMetricGroup(MetricGroup): + """``pg_barrier_state`` fan-outs: in progress, and stuck. + + Matches the reaper's ``pg_barrier_live`` / ``pg_barrier_stranded``: in + progress is ``remaining > 0``; stuck is the reaper's stranded predicate + (no progress for the stuck timeout, or past ``expires_at``) — what its next + recovery pass would pick up. The timeout comes from the same + ``WORKER_PG_BATCH_STUCK_TIMEOUT_SECONDS`` the reaper reads. One row per + in-flight execution, so the table is small. + """ + + name = "batches" + + def query(self) -> Snapshot: + from pg_queue.models import PgBarrierState + + stuck_before = Now() - timedelta( + seconds=int(settings.PLATFORM_METRICS_BATCH_STUCK_TIMEOUT_SECONDS) + ) + counts = PgBarrierState.objects.aggregate( + in_progress=Count("execution_id", filter=Q(remaining__gt=0)), + stuck=Count( + "execution_id", + filter=Q(last_progress_at__lt=stuck_before) | Q(expires_at__lt=Now()), + ), + ) + return {"in_progress": counts["in_progress"], "stuck": counts["stuck"]} + + def families(self, snapshot: Snapshot) -> Iterable[Metric]: + from prometheus_client.core import GaugeMetricFamily + + return [ + GaugeMetricFamily( + "unstract_batches_in_progress", + "In-flight file-batch fan-outs (pg_barrier_state rows with " + "remaining > 0)", + value=snapshot["in_progress"], + ), + GaugeMetricFamily( + "unstract_batches_stuck", + "Fan-outs past the stuck timeout or their expiry: what the " + "reaper's next recovery pass picks up", + value=snapshot["stuck"], + ), + ] + + +#: The groups the endpoint exports, in scrape order. +METRIC_GROUPS: tuple[MetricGroup, ...] = ( + QueueMetricGroup(), + ExecutionMetricGroup(), + BatchMetricGroup(), +) diff --git a/backend/platform_metrics/tests/__init__.py b/backend/platform_metrics/tests/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/backend/platform_metrics/tests/fakes.py b/backend/platform_metrics/tests/fakes.py new file mode 100644 index 0000000000..26ff7c69f2 --- /dev/null +++ b/backend/platform_metrics/tests/fakes.py @@ -0,0 +1,91 @@ +"""In-memory stand-ins for Redis and metric groups (DB- and Redis-free tests).""" + +from __future__ import annotations + +from collections.abc import Iterable +from typing import Any + +from platform_metrics.groups import MetricGroup, Snapshot + + +class FakeLock: + def __init__(self, redis: FakeRedis, name: str) -> None: + self._redis = redis + self._name = name + self.owned = False + + def acquire(self, blocking: bool = True) -> bool: + if self._name in self._redis.locks: + return False + self._redis.locks.add(self._name) + self.owned = True + return True + + def release(self) -> None: + if not self.owned: + raise RuntimeError("not owned") + self._redis.locks.discard(self._name) + self.owned = False + + +class FakeRedis: + def __init__(self) -> None: + self.data: dict[str, str] = {} + self.expiry: dict[str, int | None] = {} + self.locks: set[str] = set() + self.down = False + + def _check(self) -> None: + if self.down: + raise ConnectionError("redis down") + + def get(self, name: str) -> Any: + self._check() + value = self.data.get(name) + return value.encode() if value is not None else None + + def set(self, name: str, value: str, ex: int | None = None) -> bool: + self._check() + self.data[name] = value + self.expiry[name] = ex + return True + + def lock(self, name: str, timeout: float | None = None) -> FakeLock: + self._check() + return FakeLock(self, name) + + +class StubGroup(MetricGroup): + """A group whose fetch result (or failure) the test controls.""" + + def __init__(self, name: str, value: float = 1.0) -> None: + self.name = name + self.value = value + self.fetches = 0 + self.error: Exception | None = None + + def fetch(self) -> Snapshot: + self.fetches += 1 + if self.error is not None: + raise self.error + return {"value": self.value} + + def query(self) -> Snapshot: # pragma: no cover - fetch is overridden + raise NotImplementedError + + def families(self, snapshot: Snapshot) -> Iterable[Any]: + from prometheus_client.core import GaugeMetricFamily + + return [ + GaugeMetricFamily( + f"stub_{self.name}", "stub metric", value=snapshot["value"] + ) + ] + + +class Clock: + def __init__(self, now: float = 1_000_000.0) -> None: + self.now = now + + def __call__(self) -> float: + return self.now diff --git a/backend/platform_metrics/tests/test_cache.py b/backend/platform_metrics/tests/test_cache.py new file mode 100644 index 0000000000..6280c64ff3 --- /dev/null +++ b/backend/platform_metrics/tests/test_cache.py @@ -0,0 +1,198 @@ +"""SnapshotCache: Redis sharing, the refresh lock, and failure fallbacks.""" + +import json + +import pytest +from django.test import override_settings + +from platform_metrics.cache import RefreshFailed, SnapshotCache +from platform_metrics.tests.fakes import Clock, FakeRedis, StubGroup + +DATA_KEY = "platform_metrics:v1:data:g" +LOCK_KEY = "platform_metrics:v1:lock:g" + + +@pytest.fixture(autouse=True) +def _ttl(): + with override_settings(PLATFORM_METRICS_CACHE_TTL_SECONDS=15): + yield + + +@pytest.fixture +def redis(): + return FakeRedis() + + +@pytest.fixture +def clock(): + return Clock() + + +@pytest.fixture +def cache(redis, clock): + return SnapshotCache(client_factory=lambda: redis, clock=clock) + + +def _seed(redis, value, refreshed_at): + redis.data[DATA_KEY] = json.dumps( + {"data": {"value": value}, "refreshed_at": refreshed_at} + ) + + +def test_first_scrape_fetches_and_shares_through_redis(cache, redis, clock): + group = StubGroup("g", value=7) + + snapshot = cache.resolve(group) + + assert snapshot.data == {"value": 7} + assert snapshot.refreshed_at == clock.now + assert group.fetches == 1 + assert json.loads(redis.data[DATA_KEY])["data"] == {"value": 7} + # Kept well past the TTL: it is the stale fallback while refreshes fail. + assert redis.expiry[DATA_KEY] > 15 + assert not redis.locks, "the lock must be released after the refresh" + + +def test_fresh_snapshot_is_served_without_touching_the_db(cache, redis, clock): + _seed(redis, 3, clock.now - 5) + group = StubGroup("g") + + assert cache.resolve(group).data == {"value": 3} + assert group.fetches == 0 + + +def test_snapshot_written_by_another_pod_is_reused(redis, clock): + group = StubGroup("g") + pod_a = SnapshotCache(client_factory=lambda: redis, clock=clock) + pod_b = SnapshotCache(client_factory=lambda: redis, clock=clock) + + pod_a.resolve(group) + pod_b.resolve(group) + + assert group.fetches == 1 + + +def test_stale_snapshot_is_refreshed(cache, redis, clock): + _seed(redis, 3, clock.now - 16) + group = StubGroup("g", value=9) + + assert cache.resolve(group).data == {"value": 9} + assert group.fetches == 1 + + +def test_lock_held_elsewhere_serves_stale_without_fetching(cache, redis, clock): + _seed(redis, 3, clock.now - 60) + redis.locks.add(LOCK_KEY) + group = StubGroup("g") + + snapshot = cache.resolve(group) + + assert snapshot.data == {"value": 3} + assert group.fetches == 0 + + +def test_lock_held_and_nothing_cached_leaves_the_group_out(cache, redis): + redis.locks.add(LOCK_KEY) + group = StubGroup("g") + + assert cache.resolve(group) is None + assert group.fetches == 0 + + +def test_refresh_failure_serves_last_snapshot(cache, redis, clock): + _seed(redis, 3, clock.now - 60) + group = StubGroup("g") + group.error = RuntimeError("db down") + + with pytest.raises(RefreshFailed) as failed: + cache.resolve(group) + + assert failed.value.snapshot.data == {"value": 3} + assert not redis.locks + # The stale value stays in Redis for the other pods. + assert json.loads(redis.data[DATA_KEY])["data"] == {"value": 3} + + +def test_refresh_failure_with_nothing_cached_has_no_snapshot(cache): + group = StubGroup("g") + group.error = RuntimeError("db down") + + with pytest.raises(RefreshFailed) as failed: + cache.resolve(group) + + assert failed.value.snapshot is None + + +def test_redis_down_refreshes_locally_at_most_once_per_ttl(cache, redis, clock): + redis.down = True + group = StubGroup("g", value=5) + + with pytest.raises(RefreshFailed) as first: + cache.resolve(group) + clock.now += 10 + with pytest.raises(RefreshFailed) as second: + cache.resolve(group) + + assert first.value.snapshot.data == {"value": 5} + assert second.value.snapshot.data == {"value": 5} + assert group.fetches == 1, "DB load stays bounded per pod while Redis is down" + + clock.now += 10 + with pytest.raises(RefreshFailed): + cache.resolve(group) + assert group.fetches == 2 + + +def test_redis_down_after_a_good_read_serves_the_local_copy(cache, redis, clock): + _seed(redis, 3, clock.now) + group = StubGroup("g") + cache.resolve(group) + redis.down = True + + with pytest.raises(RefreshFailed) as failed: + cache.resolve(group) + + assert failed.value.snapshot.data == {"value": 3} + assert group.fetches == 0 + + +def test_redis_write_failure_still_serves_the_fresh_snapshot(cache, redis): + group = StubGroup("g", value=4) + + def broken_set(*args, **kwargs): + raise ConnectionError("write failed") + + redis.set = broken_set + + assert cache.resolve(group).data == {"value": 4} + assert not redis.locks + + +def test_groups_are_cached_independently(cache, redis, clock): + first, second = StubGroup("a"), StubGroup("b") + + cache.resolve(first) + cache.resolve(second) + cache.resolve(first) + + assert (first.fetches, second.fetches) == (1, 1) + assert set(redis.data) == { + "platform_metrics:v1:data:a", + "platform_metrics:v1:data:b", + } + + +def test_rechecks_redis_after_taking_the_lock(cache, redis, clock): + """A pod that finished refreshing between our read and our acquire wins.""" + _seed(redis, 3, clock.now - 60) + group = StubGroup("g") + make_lock = redis.lock + + def lock_after_other_pod_refreshed(name, timeout=None): + _seed(redis, 8, clock.now) + return make_lock(name, timeout=timeout) + + redis.lock = lock_after_other_pod_refreshed + + assert cache.resolve(group).data == {"value": 8} + assert group.fetches == 0 diff --git a/backend/platform_metrics/tests/test_exporter.py b/backend/platform_metrics/tests/test_exporter.py new file mode 100644 index 0000000000..13b0187324 --- /dev/null +++ b/backend/platform_metrics/tests/test_exporter.py @@ -0,0 +1,120 @@ +"""Exporter: exposition format, exporter-health metrics, failure isolation.""" + +import pytest +from django.test import override_settings +from prometheus_client.parser import text_string_to_metric_families + +from platform_metrics.cache import SnapshotCache +from platform_metrics.exporter import PlatformMetricsExporter +from platform_metrics.tests.fakes import Clock, FakeRedis, StubGroup + + +@pytest.fixture(autouse=True) +def _ttl(): + with override_settings(PLATFORM_METRICS_CACHE_TTL_SECONDS=15): + yield + + +@pytest.fixture +def clock(): + return Clock() + + +@pytest.fixture +def redis(): + return FakeRedis() + + +def _exporter(groups, redis, clock): + return PlatformMetricsExporter( + groups=groups, + cache=SnapshotCache(client_factory=lambda: redis, clock=clock), + ) + + +def _samples(exporter): + out = {} + for family in text_string_to_metric_families(exporter.render().decode()): + for sample in family.samples: + out[(sample.name, tuple(sorted(sample.labels.items())))] = sample.value + return out + + +def test_renders_every_group_with_health_metrics(redis, clock): + exporter = _exporter([StubGroup("a", 1), StubGroup("b", 2)], redis, clock) + + samples = _samples(exporter) + + assert samples[("stub_a", ())] == 1 + assert samples[("stub_b", ())] == 2 + refreshed = "unstract_metrics_last_successful_refresh_timestamp" + assert samples[(refreshed, (("group", "a"),))] == clock.now + assert samples[(refreshed, (("group", "b"),))] == clock.now + failures = "unstract_metrics_refresh_failures_total" + assert samples[(failures, (("group", "a"),))] == 0 + assert samples[(failures, (("group", "b"),))] == 0 + + +def test_failing_group_does_not_affect_the_others(redis, clock): + healthy, broken = StubGroup("a", 1), StubGroup("b", 2) + broken.error = RuntimeError("db down") + exporter = _exporter([healthy, broken], redis, clock) + + samples = _samples(exporter) + + assert samples[("stub_a", ())] == 1 + assert ("stub_b", ()) not in samples, "nothing cached yet, so nothing to serve" + failures = "unstract_metrics_refresh_failures_total" + assert samples[(failures, (("group", "b"),))] == 1 + assert samples[(failures, (("group", "a"),))] == 0 + + +def test_failing_group_serves_its_last_values(redis, clock): + group = StubGroup("a", 1) + exporter = _exporter([group], redis, clock) + _samples(exporter) + first_refresh = clock.now + + clock.now += 60 + group.value = 99 + group.error = RuntimeError("db down") + samples = _samples(exporter) + + assert samples[("stub_a", ())] == 1 + refreshed = "unstract_metrics_last_successful_refresh_timestamp" + assert samples[(refreshed, (("group", "a"),))] == first_refresh + assert samples[("unstract_metrics_refresh_failures_total", (("group", "a"),))] == 1 + + +def test_unexpected_cache_error_is_counted_not_raised(redis, clock): + class BrokenCache: + def resolve(self, group): + raise ValueError("bug") + + exporter = PlatformMetricsExporter(groups=[StubGroup("a")], cache=BrokenCache()) + + samples = _samples(exporter) + + assert samples[("unstract_metrics_refresh_failures_total", (("group", "a"),))] == 1 + + +def test_snapshot_that_no_longer_renders_is_dropped_and_counted(redis, clock): + """E.g. a snapshot an older pod cached in a different shape mid-deploy.""" + redis.data["platform_metrics:v1:data:a"] = ( + '{"data": {"unexpected": 1}, "refreshed_at": %s}' % clock.now + ) + exporter = _exporter([StubGroup("a"), StubGroup("b", 2)], redis, clock) + + samples = _samples(exporter) + + assert ("stub_a", ()) not in samples + assert samples[("stub_b", ())] == 2 + assert samples[("unstract_metrics_refresh_failures_total", (("group", "a"),))] == 1 + + +def test_output_is_prometheus_text_with_help_and_type(redis, clock): + text = _exporter([StubGroup("a")], redis, clock).render().decode() + + assert "# HELP unstract_metrics_refresh_failures_total" in text + assert "# TYPE unstract_metrics_refresh_failures_total counter" in text + assert "# TYPE unstract_metrics_last_successful_refresh_timestamp gauge" in text diff --git a/backend/platform_metrics/tests/test_groups.py b/backend/platform_metrics/tests/test_groups.py new file mode 100644 index 0000000000..25b0428ace --- /dev/null +++ b/backend/platform_metrics/tests/test_groups.py @@ -0,0 +1,94 @@ +"""Metric groups: names, labels and zero-filling (DB-free).""" + +from django.test import override_settings +from prometheus_client import CollectorRegistry, generate_latest +from prometheus_client.parser import text_string_to_metric_families + +from platform_metrics.groups import ( + BatchMetricGroup, + ExecutionMetricGroup, + QueueMetricGroup, +) + + +def _render(group, snapshot): + class _Collector: + def collect(self): + return group.families(snapshot) + + registry = CollectorRegistry() + registry.register(_Collector()) + text = generate_latest(registry).decode() + samples = {} + for family in text_string_to_metric_families(text): + for sample in family.samples: + samples[(sample.name, tuple(sorted(sample.labels.items())))] = sample.value + return samples + + +@override_settings(PLATFORM_METRICS_QUEUES=["celery", "file_processing"]) +def test_queue_series_are_zero_filled_for_known_queues_and_states(): + samples = _render( + QueueMetricGroup(), + {"queues": {"celery": {"states": {"ready": 4, "claimed": 2}, "oldest": 12.5}}}, + ) + + def messages(queue, state): + return samples[ + ("unstract_queue_messages", (("queue", queue), ("state", state))) + ] + + assert messages("celery", "ready") == 4 + assert messages("celery", "claimed") == 2 + assert messages("celery", "scheduled") == 0 + for state in ("ready", "claimed", "scheduled"): + assert messages("file_processing", state) == 0 + age = "unstract_queue_oldest_message_age_seconds" + assert samples[(age, (("queue", "celery"),))] == 12.5 + assert samples[(age, (("queue", "file_processing"),))] == 0 + + +@override_settings(PLATFORM_METRICS_QUEUES=["celery"]) +def test_queue_not_in_the_known_list_is_still_reported(): + samples = _render( + QueueMetricGroup(), + {"queues": {"custom_queue": {"states": {"scheduled": 1}, "oldest": 3.0}}}, + ) + + queues = { + dict(labels)["queue"] + for name, labels in samples + if name == "unstract_queue_messages" + } + assert queues == {"celery", "custom_queue"} + + +def test_execution_series_are_zero_filled_with_lowercase_status(): + samples = _render( + ExecutionMetricGroup(), + {"active": {"PENDING": 3}, "recent": {"COMPLETED": 10, "ERROR": 1}}, + ) + + assert samples == { + ("unstract_workflow_executions", (("status", "pending"),)): 3, + ("unstract_workflow_executions", (("status", "executing"),)): 0, + ("unstract_workflow_executions_recent", (("status", "completed"),)): 10, + ("unstract_workflow_executions_recent", (("status", "error"),)): 1, + ("unstract_workflow_executions_recent", (("status", "stopped"),)): 0, + } + + +def test_batch_series(): + samples = _render(BatchMetricGroup(), {"in_progress": 5, "stuck": 1}) + + assert samples == { + ("unstract_batches_in_progress", ()): 5, + ("unstract_batches_stuck", ()): 1, + } + + +def test_group_names_are_unique(): + from platform_metrics.groups import METRIC_GROUPS + + names = [group.name for group in METRIC_GROUPS] + assert sorted(names) == ["batches", "executions", "queues"] diff --git a/backend/platform_metrics/tests/test_groups_db.py b/backend/platform_metrics/tests/test_groups_db.py new file mode 100644 index 0000000000..e4c11b3623 --- /dev/null +++ b/backend/platform_metrics/tests/test_groups_db.py @@ -0,0 +1,157 @@ +"""Metric group queries against a real database (integration tier). + +These pin that each group counts what its docstring says it counts, including +the edges that matter for consistency with the PG reaper: the stuck predicate, +the recent-window bounds, and that queue state counts add up to the depth. +""" + +from __future__ import annotations + +import uuid +from datetime import timedelta + +import pytest +from django.test import override_settings +from django.utils import timezone + +from unstract.core.data_models import ExecutionStatus + +pytestmark = pytest.mark.django_db + + +def _message(queue, state, enqueued_ago=timedelta(0)): + from pg_queue.models import PgQueueMessage + + return PgQueueMessage.objects.create( + queue_name=queue, + message={}, + state=state, + enqueued_at=timezone.now() - enqueued_ago, + ) + + +def _execution(status, created_ago=timedelta(minutes=5), modified_ago=None): + from workflow_manager.workflow_v2.models import Workflow, WorkflowExecution + + ex = WorkflowExecution.objects.create( + id=uuid.uuid4(), + workflow=Workflow.objects.create(workflow_name=f"wf-{uuid.uuid4().hex[:8]}"), + status=status, + ) + now = timezone.now() + WorkflowExecution.objects.filter(id=ex.id).update( + created_at=now - created_ago, + modified_at=now - (modified_ago if modified_ago is not None else created_ago), + ) + return ex + + +def _barrier(remaining, progress_ago=timedelta(0), expires_in=timedelta(hours=6)): + from pg_queue.models import PgBarrierState + + now = timezone.now() + return PgBarrierState.objects.create( + execution_id=str(uuid.uuid4()), + remaining=remaining, + # Before both, so an already-expired row still passes the + # expires-after-created constraint. + created_at=now - timedelta(days=1), + expires_at=now + expires_in, + last_progress_at=now - progress_ago, + ) + + +class TestQueueGroup: + def test_counts_by_queue_and_state_with_oldest_age_across_states(self): + from platform_metrics.groups import QueueMetricGroup + + _message("celery", "ready", enqueued_ago=timedelta(seconds=30)) + _message("celery", "ready") + _message("celery", "claimed", enqueued_ago=timedelta(seconds=120)) + _message("scheduler", "scheduled", enqueued_ago=timedelta(seconds=10)) + + queues = QueueMetricGroup().fetch()["queues"] + + assert queues["celery"]["states"] == {"ready": 2, "claimed": 1} + assert queues["celery"]["oldest"] == pytest.approx(120, abs=5) + assert queues["scheduler"]["states"] == {"scheduled": 1} + assert queues["scheduler"]["oldest"] == pytest.approx(10, abs=5) + + def test_empty_table(self): + from platform_metrics.groups import QueueMetricGroup + + assert QueueMetricGroup().fetch() == {"queues": {}} + + +class TestExecutionGroup: + @override_settings( + PLATFORM_METRICS_RECENT_WINDOW_SECONDS=3600, + PLATFORM_METRICS_RECENT_LOOKBACK_SECONDS=86400, + ) + def test_active_and_recent_counts(self): + from platform_metrics.groups import ExecutionMetricGroup + + _execution(ExecutionStatus.PENDING.value) + _execution(ExecutionStatus.EXECUTING.value) + _execution(ExecutionStatus.EXECUTING.value) + # Old execution, but went terminal recently: counted. + _execution( + ExecutionStatus.COMPLETED.value, + created_ago=timedelta(hours=5), + modified_ago=timedelta(minutes=10), + ) + _execution(ExecutionStatus.ERROR.value) + # Went terminal before the window: not counted. + _execution( + ExecutionStatus.COMPLETED.value, + created_ago=timedelta(hours=3), + modified_ago=timedelta(hours=2), + ) + # Created before the lookback: outside the scan, by design. + _execution( + ExecutionStatus.STOPPED.value, + created_ago=timedelta(days=2), + modified_ago=timedelta(minutes=1), + ) + + snapshot = ExecutionMetricGroup().fetch() + + assert snapshot["active"] == {"PENDING": 1, "EXECUTING": 2} + assert snapshot["recent"] == {"COMPLETED": 1, "ERROR": 1} + + +class TestBatchGroup: + @override_settings(PLATFORM_METRICS_BATCH_STUCK_TIMEOUT_SECONDS=600) + def test_in_progress_and_stuck_match_the_reaper_predicates(self): + from platform_metrics.groups import BatchMetricGroup + + _barrier(remaining=3) # live, healthy + _barrier(remaining=2, progress_ago=timedelta(minutes=20)) # live, stuck + _barrier(remaining=1, expires_in=timedelta(minutes=-1)) # live, expired + # remaining == 0 lingerer past the timeout: not live, but stranded — + # the reaper's recovery pass picks it up, so it counts as stuck. + _barrier(remaining=0, progress_ago=timedelta(minutes=20)) + + assert BatchMetricGroup().fetch() == {"in_progress": 3, "stuck": 3} + + +def test_fetch_applies_the_statement_timeout(): + """A slow group is cancelled by Postgres rather than holding the scrape.""" + from django.db import OperationalError, connection + + from platform_metrics.groups import MetricGroup + + class Slow(MetricGroup): + name = "slow" + + def query(self): + with connection.cursor() as cursor: + cursor.execute("SELECT pg_sleep(2)") + return {} + + def families(self, snapshot): + return [] + + with override_settings(PLATFORM_METRICS_QUERY_TIMEOUT_MS=100): + with pytest.raises(OperationalError): + Slow().fetch() diff --git a/backend/platform_metrics/tests/test_views.py b/backend/platform_metrics/tests/test_views.py new file mode 100644 index 0000000000..6faf094844 --- /dev/null +++ b/backend/platform_metrics/tests/test_views.py @@ -0,0 +1,60 @@ +"""GET /metrics: routing, auth exemption, bearer token, content type.""" + +from unittest.mock import MagicMock, patch + +import pytest +from django.test import Client, override_settings + +BODY = b"# HELP stub stub\n# TYPE stub gauge\nstub 1.0\n" + + +@pytest.fixture(autouse=True) +def exporter(): + fake = MagicMock() + fake.render.return_value = BODY + with patch("platform_metrics.views.get_exporter", return_value=fake): + yield fake + + +@pytest.fixture +def client(): + return Client() + + +@override_settings(PLATFORM_METRICS_TOKEN="") +def test_served_without_a_session_when_no_token_is_configured(client): + response = client.get("/metrics") + + assert response.status_code == 200 + assert response["Content-Type"] == "text/plain; version=0.0.4; charset=utf-8" + assert response.content == BODY + + +@override_settings(PLATFORM_METRICS_TOKEN="s3cret") +def test_token_configured_rejects_a_missing_header(client, exporter): + response = client.get("/metrics") + + assert response.status_code == 401 + assert response["WWW-Authenticate"].startswith("Bearer") + exporter.render.assert_not_called() + + +@override_settings(PLATFORM_METRICS_TOKEN="s3cret") +@pytest.mark.parametrize("header", ["Bearer wrong", "Basic s3cret", "s3cret", "Bearer "]) +def test_token_configured_rejects_a_wrong_header(client, exporter, header): + response = client.get("/metrics", HTTP_AUTHORIZATION=header) + + assert response.status_code == 401 + exporter.render.assert_not_called() + + +@override_settings(PLATFORM_METRICS_TOKEN="s3cret") +def test_token_configured_accepts_the_right_bearer(client): + response = client.get("/metrics", HTTP_AUTHORIZATION="Bearer s3cret") + + assert response.status_code == 200 + assert response.content == BODY + + +def test_only_get_is_allowed(client): + assert client.post("/metrics").status_code == 405 diff --git a/backend/platform_metrics/urls.py b/backend/platform_metrics/urls.py new file mode 100644 index 0000000000..2928b1a008 --- /dev/null +++ b/backend/platform_metrics/urls.py @@ -0,0 +1,5 @@ +from django.urls import path + +from .views import metrics + +urlpatterns = [path("metrics", metrics, name="platform-metrics")] diff --git a/backend/platform_metrics/views.py b/backend/platform_metrics/views.py new file mode 100644 index 0000000000..178adeb536 --- /dev/null +++ b/backend/platform_metrics/views.py @@ -0,0 +1,51 @@ +"""``GET /metrics``: platform metrics in Prometheus text format (UN-4251). + +Mounted at the URL root, beside ``/health``, which keeps it off the public +ingress: that sends the backend only prefixed paths (``/api/v1``, +``/deployment``, ``/public``, ...) and everything else to the frontend, so +``/metrics`` is reachable only in-cluster, on the backend service. +The data is platform-wide, so it is also exempt from session auth +(``WHITELISTED_PATHS``) and guarded instead by an optional bearer token, +``PLATFORM_METRICS_TOKEN``, for environments that want one. + +A plain Django view rather than DRF: the response is the exposition bytes as +they are, with no content negotiation or renderer in between. +""" + +import hmac + +from django.conf import settings +from django.db import transaction +from django.http import HttpRequest, HttpResponse +from django.views.decorators.http import require_GET + +from .exporter import get_exporter + +_BEARER = "Bearer " +# The text format generate_latest() writes, spelled out rather than taken from +# prometheus_client's CONTENT_TYPE_LATEST, which newer releases point at the +# 1.0.0 format that older scrapers do not recognise. Matches the PG workers' +# /metrics (workers/queue_backend/pg_queue/metrics.py). +METRICS_CONTENT_TYPE = "text/plain; version=0.0.4; charset=utf-8" + + +def _authorised(request: HttpRequest) -> bool: + token = settings.PLATFORM_METRICS_TOKEN + if not token: + return True + header = request.headers.get("Authorization", "") + if not header.startswith(_BEARER): + return False + return hmac.compare_digest(header[len(_BEARER) :].encode(), token.encode()) + + +# Non-atomic so each group's refresh is its own outermost transaction even with +# ATOMIC_REQUESTS on: its SET LOCAL statement_timeout then ends with it. +@transaction.non_atomic_requests +@require_GET +def metrics(request: HttpRequest) -> HttpResponse: + if not _authorised(request): + response = HttpResponse("Unauthorized\n", status=401, content_type="text/plain") + response["WWW-Authenticate"] = 'Bearer realm="metrics"' + return response + return HttpResponse(get_exporter().render(), content_type=METRICS_CONTENT_TYPE) diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 60ca4a70e5..76cd453884 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -31,6 +31,7 @@ dependencies = [ # contract published clients are generated from. "drf-spectacular==0.30.0", "psycopg2-binary==2.9.9", + "prometheus-client>=0.17.0,<1.0.0", # Platform /metrics exporter "python-dotenv==1.2.2", "python-magic==0.4.27", # For file upload/download "python-socketio==5.17.0", # For log_events diff --git a/backend/uv.lock b/backend/uv.lock index 5e1be44c61..6013652669 100644 --- a/backend/uv.lock +++ b/backend/uv.lock @@ -2603,6 +2603,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/4b/a6/38c8e2f318bf67d338f4d629e93b0b4b9af331f455f0390ea8ce4a099b26/portalocker-3.2.0-py3-none-any.whl", hash = "sha256:3cdc5f565312224bc570c49337bd21428bba0ef363bbcf58b9ef4a9f11779968", size = 22424, upload-time = "2025-06-14T13:20:38.083Z" }, ] +[[package]] +name = "prometheus-client" +version = "0.26.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/52/73/f1334c29c2af4cd9dba6c7817e61b611bd0215e2eb5565c6064a4de18802/prometheus_client-0.26.0.tar.gz", hash = "sha256:04a91bcf94e2cf74a44a1a874d651a2e853ed354b6e822f3b7487751465d5c2b", size = 92910, upload-time = "2026-07-24T19:36:41.893Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/eb/a3/b69efbf4143b5b9859b977770bbbabcc2796b702fa69dc40271e45cd5a56/prometheus_client-0.26.0-py3-none-any.whl", hash = "sha256:fa93d06737aa02bacd05794768508bb97d2fbee28cb3bca04eaae92f0ca953d6", size = 64494, upload-time = "2026-07-24T19:36:40.854Z" }, +] + [[package]] name = "prompt-toolkit" version = "3.0.52" @@ -3729,6 +3738,7 @@ dependencies = [ { name = "google-cloud-recaptcha-enterprise" }, { name = "gunicorn" }, { name = "httpx" }, + { name = "prometheus-client" }, { name = "psycopg2-binary" }, { name = "python-dotenv" }, { name = "python-magic" }, @@ -3797,6 +3807,7 @@ requires-dist = [ { name = "google-cloud-recaptcha-enterprise", specifier = ">=1.28.2" }, { name = "gunicorn", specifier = ">=23.0.0" }, { name = "httpx", specifier = ">=0.27.0" }, + { name = "prometheus-client", specifier = ">=0.17.0,<1.0.0" }, { name = "psycopg2-binary", specifier = "==2.9.9" }, { name = "python-dotenv", specifier = "==1.2.2" }, { name = "python-magic", specifier = "==0.4.27" }, From 72f07ca536087c8375f0ae94268b59aaa14d970a Mon Sep 17 00:00:00 2001 From: ali <117142933+muhammad-ali-e@users.noreply.github.com> Date: Tue, 27 Oct 2026 07:18:59 +0000 Subject: [PATCH 2/3] UN-4251 Count recent completions by finish time; harden metric cache fallbacks - Recent terminal executions are now filtered on created_at + execution_time (stamped on the terminal transition) instead of modified_at, which later writes such as result acknowledgement bump. - A Redis failure on the re-read under the lock now serves the cached snapshot through the failure path instead of dropping the group. - A failed Redis write is counted as a refresh failure, and the newer local copy is preferred over the older Redis value, so the pod does not refresh the DB on every scrape while writes fail. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_016Hov58G5waUqpMHz9P6FmL --- backend/platform_metrics/cache.py | 44 +++++++++++++++---- backend/platform_metrics/groups.py | 32 ++++++++++---- backend/platform_metrics/tests/test_cache.py | 33 +++++++++++++- .../platform_metrics/tests/test_groups_db.py | 22 +++++++--- 4 files changed, 106 insertions(+), 25 deletions(-) diff --git a/backend/platform_metrics/cache.py b/backend/platform_metrics/cache.py index 731cf1f66e..f7b5a7a4a6 100644 --- a/backend/platform_metrics/cache.py +++ b/backend/platform_metrics/cache.py @@ -105,7 +105,7 @@ def resolve(self, group: MetricGroup) -> CachedSnapshot | None: """ try: client = self._client_factory() - cached = self._read(client, group) + cached = self._newest(group, self._read(client, group)) except Exception: logger.warning( "platform metrics: Redis unavailable for group %r; using the " @@ -130,15 +130,23 @@ def resolve(self, group: MetricGroup) -> CachedSnapshot | None: return self._resolve_local(group, fallback=cached) if not acquired: # Another pod is refreshing; its result lands for the next scrape. - return cached or self._local.get(group.name) + return cached try: # Re-read under the lock: another pod may have refreshed between # our read and our acquire. - latest = self._read(client, group) + try: + latest = self._newest(group, self._read(client, group)) + except Exception: + logger.warning( + "platform metrics: Redis failed re-reading group %r under the lock", + group.name, + exc_info=True, + ) + return self._resolve_local(group, fallback=cached) if latest is not None and latest.age(self._clock()) < self._ttl(): return latest - snapshot = self._refresh(group, fallback=latest or cached) + snapshot = self._refresh(group, fallback=latest) try: client.set( self._data_key(group), @@ -147,13 +155,17 @@ def resolve(self, group: MetricGroup) -> CachedSnapshot | None: ), ex=_RETENTION_SECONDS, ) - except Exception: - # The fresh snapshot is still good to serve from this pod. + except Exception as e: + # Still good to serve, and kept as this process's copy, which + # _newest() prefers over the older Redis value until it goes + # stale, so a write-failing Redis does not turn every scrape + # into a DB refresh. Counted, since the other pods miss it. logger.warning( "platform metrics: could not cache group %r in Redis", group.name, exc_info=True, ) + raise RefreshFailed(snapshot) from e return snapshot finally: try: @@ -195,11 +207,25 @@ def _read(self, client: RedisLike, group: MetricGroup) -> CachedSnapshot | None: if raw is None: return None payload = json.loads(raw) - snapshot = CachedSnapshot( + return CachedSnapshot( data=payload["data"], refreshed_at=float(payload["refreshed_at"]) ) - self._local[group.name] = snapshot - return snapshot + + def _newest( + self, group: MetricGroup, shared: CachedSnapshot | None + ) -> CachedSnapshot | None: + """The newer of the Redis snapshot and this process's copy, kept locally. + + The local copy can be the newer one when this pod refreshed but could + not write the result back to Redis. + """ + local = self._local.get(group.name) + if shared is None or ( + local is not None and local.refreshed_at >= shared.refreshed_at + ): + return local + self._local[group.name] = shared + return shared @staticmethod def _data_key(group: MetricGroup) -> str: diff --git a/backend/platform_metrics/groups.py b/backend/platform_metrics/groups.py index ccd3481407..8ffafcbc96 100644 --- a/backend/platform_metrics/groups.py +++ b/backend/platform_metrics/groups.py @@ -22,7 +22,16 @@ from django.conf import settings from django.db import connection, transaction -from django.db.models import Count, DurationField, ExpressionWrapper, Min, Q +from django.db.models import ( + Count, + DateTimeField, + DurationField, + ExpressionWrapper, + F, + Min, + Q, + Value, +) from django.db.models.functions import Now from django.utils import timezone @@ -139,12 +148,14 @@ class ExecutionMetricGroup(MetricGroup): * active (``PENDING`` / ``EXECUTING``) is implied by the predicate of the partial ``we_active_by_workflow_idx``, so it costs O(active executions); * recent terminal is bounded by ``created_at`` (``we_created_at_idx``) to - ``PLATFORM_METRICS_RECENT_LOOKBACK_SECONDS``, then filtered on - ``modified_at`` — the time the row last changed, i.e. when it went - terminal. ``modified_at`` is deliberately not indexed (it is rewritten on - every status update), so an execution that ran longer than the lookback - before finishing is not counted. The default lookback (24h) is far above - any normal run. + ``PLATFORM_METRICS_RECENT_LOOKBACK_SECONDS``, then filtered on when the + execution finished: ``created_at + execution_time``. Both status writers + stamp ``execution_time`` on the terminal transition, so this is the + completion time. ``modified_at`` is not, because later writes bump it + (acknowledging an API result, for one, would make an old completion + count as new). An execution that ran longer than the lookback before + finishing is not counted; the default lookback (24h) is far above any + normal run. """ name = "executions" @@ -163,12 +174,17 @@ def query(self) -> Snapshot: .annotate(n=Count("id")) .order_by() ) + finished_at = ExpressionWrapper( + F("created_at") + F("execution_time") * Value(timedelta(seconds=1)), + output_field=DateTimeField(), + ) recent = ( WorkflowExecution.objects.filter( status__in=_TERMINAL_STATUSES, created_at__gte=now - max(lookback, window), - modified_at__gte=now - window, ) + .annotate(finished_at=finished_at) + .filter(finished_at__gte=now - window) .values("status") .annotate(n=Count("id")) .order_by() diff --git a/backend/platform_metrics/tests/test_cache.py b/backend/platform_metrics/tests/test_cache.py index 6280c64ff3..1518962552 100644 --- a/backend/platform_metrics/tests/test_cache.py +++ b/backend/platform_metrics/tests/test_cache.py @@ -156,7 +156,8 @@ def test_redis_down_after_a_good_read_serves_the_local_copy(cache, redis, clock) assert group.fetches == 0 -def test_redis_write_failure_still_serves_the_fresh_snapshot(cache, redis): +def test_redis_write_failure_serves_fresh_snapshot_and_counts(cache, redis, clock): + _seed(redis, 3, clock.now - 60) group = StubGroup("g", value=4) def broken_set(*args, **kwargs): @@ -164,7 +165,37 @@ def broken_set(*args, **kwargs): redis.set = broken_set + with pytest.raises(RefreshFailed) as failed: + cache.resolve(group) + assert failed.value.snapshot.data == {"value": 4} + assert not redis.locks + + # The older Redis value must not displace the newer local copy, and the + # next scrape inside the TTL must not hit the DB again. + clock.now += 5 assert cache.resolve(group).data == {"value": 4} + assert group.fetches == 1 + + +def test_redis_failing_on_the_reread_under_the_lock_serves_cached(cache, redis, clock): + _seed(redis, 3, clock.now - 60) + group = StubGroup("g") + group.error = RuntimeError("db down") + real_get = redis.get + calls = {"n": 0} + + def get_failing_second_time(name): + calls["n"] += 1 + if calls["n"] > 1: + raise ConnectionError("redis dropped") + return real_get(name) + + redis.get = get_failing_second_time + + with pytest.raises(RefreshFailed) as failed: + cache.resolve(group) + + assert failed.value.snapshot.data == {"value": 3} assert not redis.locks diff --git a/backend/platform_metrics/tests/test_groups_db.py b/backend/platform_metrics/tests/test_groups_db.py index e4c11b3623..588bf40862 100644 --- a/backend/platform_metrics/tests/test_groups_db.py +++ b/backend/platform_metrics/tests/test_groups_db.py @@ -30,7 +30,13 @@ def _message(queue, state, enqueued_ago=timedelta(0)): ) -def _execution(status, created_ago=timedelta(minutes=5), modified_ago=None): +def _execution(status, created_ago=timedelta(minutes=5), ran_for=None): + """An execution created ``created_ago``, terminal after ``ran_for``. + + ``modified_at`` is set to now on purpose: later writes (an API result being + acknowledged, for one) bump it, so the group must not read it as the + completion time. + """ from workflow_manager.workflow_v2.models import Workflow, WorkflowExecution ex = WorkflowExecution.objects.create( @@ -41,7 +47,8 @@ def _execution(status, created_ago=timedelta(minutes=5), modified_ago=None): now = timezone.now() WorkflowExecution.objects.filter(id=ex.id).update( created_at=now - created_ago, - modified_at=now - (modified_ago if modified_ago is not None else created_ago), + execution_time=(ran_for or timedelta(0)).total_seconds(), + modified_at=now, ) return ex @@ -94,24 +101,25 @@ def test_active_and_recent_counts(self): _execution(ExecutionStatus.PENDING.value) _execution(ExecutionStatus.EXECUTING.value) _execution(ExecutionStatus.EXECUTING.value) - # Old execution, but went terminal recently: counted. + # Old execution, but finished recently: counted. _execution( ExecutionStatus.COMPLETED.value, created_ago=timedelta(hours=5), - modified_ago=timedelta(minutes=10), + ran_for=timedelta(hours=4, minutes=50), ) _execution(ExecutionStatus.ERROR.value) - # Went terminal before the window: not counted. + # Finished before the window, touched since (e.g. its result was + # acknowledged just now): not counted. _execution( ExecutionStatus.COMPLETED.value, created_ago=timedelta(hours=3), - modified_ago=timedelta(hours=2), + ran_for=timedelta(hours=1), ) # Created before the lookback: outside the scan, by design. _execution( ExecutionStatus.STOPPED.value, created_ago=timedelta(days=2), - modified_ago=timedelta(minutes=1), + ran_for=timedelta(days=2) - timedelta(minutes=1), ) snapshot = ExecutionMetricGroup().fetch() From 2764b9176dea8d17aba60193f084d53ac3085c0a Mon Sep 17 00:00:00 2001 From: ali <117142933+muhammad-ali-e@users.noreply.github.com> Date: Tue, 27 Oct 2026 07:18:59 +0000 Subject: [PATCH 3/3] UN-4251 Share refresh-failure totals across pods and harden the metrics cache - Count refresh failures in Redis (HINCRBY + HGETALL in one transaction), so every pod reports the same monotonic total; omit it while Redis is down - Add unstract_metrics_cache_up to make a Redis outage visible - Fall back to modified_at as the finish time for swept executions - Size the refresh lock from each group's statement count and query timeout - Treat an unreadable cache entry as missing; per-group local locks - Note in HELP text that queue metrics cover the PG transport only Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_016Hov58G5waUqpMHz9P6FmL --- backend/platform_metrics/cache.py | 128 ++++++++++++++++-- backend/platform_metrics/exporter.py | 70 ++++++---- backend/platform_metrics/groups.py | 28 +++- backend/platform_metrics/tests/fakes.py | 35 +++++ backend/platform_metrics/tests/test_cache.py | 18 +++ .../platform_metrics/tests/test_exporter.py | 56 +++++++- .../platform_metrics/tests/test_groups_db.py | 19 ++- 7 files changed, 301 insertions(+), 53 deletions(-) diff --git a/backend/platform_metrics/cache.py b/backend/platform_metrics/cache.py index f7b5a7a4a6..0b206bb88c 100644 --- a/backend/platform_metrics/cache.py +++ b/backend/platform_metrics/cache.py @@ -13,10 +13,17 @@ * the refresh query fails: serve the last snapshot (Redis, else this process's copy) and count the failure; * Redis is unreachable: fall back to this process's copy, refreshing it from - the DB at most once per TTL, so DB load stays bounded per pod rather than - per scrape. The failure is counted; + the DB at most once per TTL, so DB load stays bounded per gunicorn worker + process rather than per scrape. The failure is counted; +* the Redis entry is unreadable (e.g. written by another version): treated as + missing, so the next refresh overwrites it; * nothing has ever been fetched (first scrape, while another pod holds the lock): the group is left out of this scrape. + +Refresh failures are counted in Redis too (:class:`FailureCounter`), so every +pod and worker process reports the same monotonic total. A per-process counter +would jump between processes from scrape to scrape, which Prometheus reads as +counter resets. """ from __future__ import annotations @@ -39,9 +46,20 @@ #: How long a snapshot stays in Redis. Far beyond the TTL on purpose: it is the #: last-known-good value served while refreshes fail. _RETENTION_SECONDS = 3600 -#: Lock expiry. Above the statement timeout, so a pod that dies mid-refresh -#: frees the group within one window instead of wedging it. -_LOCK_TIMEOUT_SECONDS = 30 +#: Floor for the lock expiry. A pod that dies mid-refresh frees the group +#: within this long instead of wedging it. +_MIN_LOCK_TIMEOUT_SECONDS = 30 + + +def _lock_timeout_seconds(group: MetricGroup) -> float: + """Lock expiry, kept above the longest refresh the statement timeout allows. + + The timeout bounds each statement, so a group's refresh can take its + statement count times that. A shorter lock could expire mid-refresh and + let a second pod start the same queries. + """ + query_timeout = int(settings.PLATFORM_METRICS_QUERY_TIMEOUT_MS) / 1000 + return max(_MIN_LOCK_TIMEOUT_SECONDS, group.statements * query_timeout + 10) class _Lock(Protocol): @@ -53,6 +71,7 @@ class RedisLike(Protocol): def get(self, name: str) -> Any: ... def set(self, name: str, value: str, ex: int | None = ...) -> Any: ... def lock(self, name: str, timeout: float | None = ...) -> _Lock: ... + def pipeline(self, transaction: bool = ...) -> Any: ... @dataclass(frozen=True) @@ -91,7 +110,10 @@ def __init__( self._clock = clock # Per-process last-known-good copies: the fallback when Redis is down. self._local: dict[str, CachedSnapshot] = {} - self._local_lock = threading.Lock() + # One lock per group, so a slow refresh of one group never holds up + # scrapes waiting on another. + self._local_locks: dict[str, threading.Lock] = {} + self._local_locks_guard = threading.Lock() @staticmethod def _ttl() -> float: @@ -119,7 +141,9 @@ def resolve(self, group: MetricGroup) -> CachedSnapshot | None: return cached try: - lock = client.lock(self._lock_key(group), timeout=_LOCK_TIMEOUT_SECONDS) + lock = client.lock( + self._lock_key(group), timeout=_lock_timeout_seconds(group) + ) acquired = lock.acquire(blocking=False) except Exception: logger.warning( @@ -184,12 +208,16 @@ def _resolve_local( unusable counts as a failure for the group, even when the snapshot it serves is fresh, so a Redis outage shows on the failure counter. """ - with self._local_lock: + with self._local_lock(group): local = self._local.get(group.name) if local is not None and local.age(self._clock()) < self._ttl(): raise RefreshFailed(local) raise RefreshFailed(self._refresh(group, fallback=local or fallback)) + def _local_lock(self, group: MetricGroup) -> threading.Lock: + with self._local_locks_guard: + return self._local_locks.setdefault(group.name, threading.Lock()) + def _refresh( self, group: MetricGroup, fallback: CachedSnapshot | None ) -> CachedSnapshot: @@ -203,13 +231,22 @@ def _refresh( return snapshot def _read(self, client: RedisLike, group: MetricGroup) -> CachedSnapshot | None: - raw = client.get(self._data_key(group)) + raw = client.get(self._data_key(group)) # a Redis error propagates if raw is None: return None - payload = json.loads(raw) - return CachedSnapshot( - data=payload["data"], refreshed_at=float(payload["refreshed_at"]) - ) + try: + payload = json.loads(raw) + return CachedSnapshot( + data=payload["data"], refreshed_at=float(payload["refreshed_at"]) + ) + except (ValueError, TypeError, KeyError): + # Not a Redis outage: treat as missing so a refresh overwrites it, + # rather than falling back to per-process refreshes until it expires. + logger.warning( + "platform metrics: ignoring unreadable cache entry for group %r", + group.name, + ) + return None def _newest( self, group: MetricGroup, shared: CachedSnapshot | None @@ -234,3 +271,68 @@ def _data_key(group: MetricGroup) -> str: @staticmethod def _lock_key(group: MetricGroup) -> str: return f"{_KEY_PREFIX}:lock:{group.name}" + + +class FailureCounter: + """Refresh-failure totals per group, shared through Redis. + + Increments are buffered in-process and flushed with ``HINCRBY`` together + with the read, in one transactional pipeline. A failure recorded while + Redis is down is added on this process's next successful scrape; it is + lost if the process exits first, so the total is best-effort during a + Redis outage. While Redis is unreachable, :meth:`totals` returns ``None`` + and the counter is left out of the scrape rather than dropping to this + process's partial count, which Prometheus would read as a reset; the + exporter's ``unstract_metrics_cache_up`` gauge reports the outage instead. + """ + + def __init__(self, client_factory: Callable[[], RedisLike] = _default_client) -> None: + self._client_factory = client_factory + self._pending: dict[str, int] = {} + self._lock = threading.Lock() + + def record(self, group: str) -> None: + with self._lock: + self._pending[group] = self._pending.get(group, 0) + 1 + + def totals(self) -> dict[str, int] | None: + """Flush pending increments, then read the shared totals.""" + # Take the pending counts and do the Redis I/O outside the lock, so a + # slow Redis never blocks record() on another thread. + with self._lock: + pending, self._pending = self._pending, {} + try: + pipe = self._client_factory().pipeline(transaction=True) + for group, amount in pending.items(): + pipe.hincrby(_FAILURES_KEY, group, amount) + pipe.hgetall(_FAILURES_KEY) + raw = pipe.execute()[-1] + except Exception: + self._restore(pending) + logger.warning( + "platform metrics: could not read refresh-failure totals", exc_info=True + ) + return None + totals: dict[str, int] = {} + for key, value in raw.items(): + try: + totals[_text(key)] = int(value) + except (ValueError, TypeError): + logger.warning( + "platform metrics: ignoring unreadable failure total %r", key + ) + return totals + + def _restore(self, pending: dict[str, int]) -> None: + with self._lock: + for group, amount in pending.items(): + self._pending[group] = self._pending.get(group, 0) + amount + + +#: No expiry: the totals must only grow. A Redis flush resets them to 0, which +#: Prometheus handles as an ordinary counter reset. +_FAILURES_KEY = f"{_KEY_PREFIX}:refresh_failures" + + +def _text(value: Any) -> str: + return value.decode() if isinstance(value, bytes) else str(value) diff --git a/backend/platform_metrics/exporter.py b/backend/platform_metrics/exporter.py index 4a6ee23a74..3178995c53 100644 --- a/backend/platform_metrics/exporter.py +++ b/backend/platform_metrics/exporter.py @@ -1,10 +1,11 @@ """Assembles the platform metric groups into one Prometheus exposition. -One process-wide registry holds two things: the refresh-failure counter, which -must survive across scrapes, and a collector that resolves every group through -the shared :class:`~platform_metrics.cache.SnapshotCache` at scrape time. Each -group is resolved inside its own ``try``, so one failing group never blanks the -others and the scrape itself always succeeds. +One process-wide registry holds a collector that resolves every group through +the shared :class:`~platform_metrics.cache.SnapshotCache` at scrape time and +reports refresh failures from the Redis-shared +:class:`~platform_metrics.cache.FailureCounter`. Each group is resolved inside +its own ``try``, so one failing group never blanks the others and the scrape +itself always succeeds. The registry is instance-owned rather than ``prometheus_client``'s default ``REGISTRY``, so nothing else in the process can leak series into this @@ -18,11 +19,11 @@ from collections.abc import Iterable, Sequence from typing import TYPE_CHECKING -from .cache import RefreshFailed, SnapshotCache +from .cache import FailureCounter, RefreshFailed, SnapshotCache from .groups import METRIC_GROUPS, MetricGroup if TYPE_CHECKING: - from prometheus_client import CollectorRegistry, Counter + from prometheus_client import CollectorRegistry from prometheus_client.core import Metric logger = logging.getLogger(__name__) @@ -33,7 +34,7 @@ def __init__( self, groups: Sequence[MetricGroup], cache: SnapshotCache, - failures: Counter, + failures: FailureCounter, ) -> None: self._groups = groups self._cache = cache @@ -44,7 +45,7 @@ def describe(self) -> Iterable[Metric]: return [] def collect(self) -> Iterable[Metric]: - from prometheus_client.core import GaugeMetricFamily + from prometheus_client.core import CounterMetricFamily, GaugeMetricFamily refreshed = GaugeMetricFamily( "unstract_metrics_last_successful_refresh_timestamp", @@ -57,11 +58,11 @@ def collect(self) -> Iterable[Metric]: try: snapshot = self._cache.resolve(group) except RefreshFailed as failed: - self._failures.labels(group=group.name).inc() + self._failures.record(group.name) snapshot = failed.snapshot except Exception: logger.exception("platform metrics: group %r failed", group.name) - self._failures.labels(group=group.name).inc() + self._failures.record(group.name) snapshot = None if snapshot is None: continue @@ -73,10 +74,34 @@ def collect(self) -> Iterable[Metric]: logger.exception( "platform metrics: rendering group %r failed", group.name ) - self._failures.labels(group=group.name).inc() + self._failures.record(group.name) continue refreshed.add_metric([group.name], snapshot.refreshed_at) families.append(refreshed) + # Read after the groups, so failures from this scrape are included. + totals = self._failures.totals() + families.append( + GaugeMetricFamily( + "unstract_metrics_cache_up", + "1 when this process could reach the Redis metrics cache on this " + "scrape, else 0 (groups then serve per-process copies and the " + "failure counter is left out)", + value=0 if totals is None else 1, + ) + ) + if totals is not None: + failures = CounterMetricFamily( + "unstract_metrics_refresh_failures", + "Failed refreshes per metric group. One cluster-wide total kept " + "in Redis, so every backend pod reports the same value: " + "aggregate across instances with max(), not sum(). The group " + "serves its last cached values meanwhile", + labels=["group"], + ) + for group in self._groups: + # Every group's series exists, so one that never failed reads 0. + failures.add_metric([group.name], totals.get(group.name, 0)) + families.append(failures) return families @@ -85,27 +110,16 @@ def __init__( self, groups: Sequence[MetricGroup] = METRIC_GROUPS, cache: SnapshotCache | None = None, + failures: FailureCounter | None = None, ) -> None: - from prometheus_client import CollectorRegistry, Counter + from prometheus_client import CollectorRegistry self.registry: CollectorRegistry = CollectorRegistry() - self.failures = Counter( - "unstract_metrics_refresh_failures", - "Failed refreshes per metric group; the group serves its last " - "cached values meanwhile", - ["group"], - registry=None, - ) - for group in groups: - # Pre-create every series so a group that has never failed reads 0. - self.failures.labels(group=group.name) - # Order matters: a registry collects in registration order, and the - # groups collector is what increments the counter, so it goes first - # for a failure to show in the same scrape that hit it. self.registry.register( - _GroupsCollector(groups, cache or SnapshotCache(), self.failures) + _GroupsCollector( + groups, cache or SnapshotCache(), failures or FailureCounter() + ) ) - self.registry.register(self.failures) def render(self) -> bytes: from prometheus_client import generate_latest diff --git a/backend/platform_metrics/groups.py b/backend/platform_metrics/groups.py index 8ffafcbc96..61102c44a3 100644 --- a/backend/platform_metrics/groups.py +++ b/backend/platform_metrics/groups.py @@ -23,6 +23,7 @@ from django.conf import settings from django.db import connection, transaction from django.db.models import ( + Case, Count, DateTimeField, DurationField, @@ -31,6 +32,7 @@ Min, Q, Value, + When, ) from django.db.models.functions import Now from django.utils import timezone @@ -54,6 +56,9 @@ class MetricGroup(ABC): #: Label value for this group on the exporter-health metrics, and the #: suffix of its Redis cache / lock keys. name: str + #: SQL statements ``query`` runs. The statement timeout bounds each one, so + #: the refresh lock is sized from this. + statements: int = 1 def fetch(self) -> Snapshot: """Read a fresh snapshot, bounded by the configured statement timeout. @@ -124,7 +129,8 @@ def families(self, snapshot: Snapshot) -> Iterable[Metric]: "unstract_queue_messages", "Messages in pg_queue_message by queue and claim state (ready = " "waiting, claimed = held by a consumer, scheduled = delayed until " - "available_at)", + "available_at). PG queue transport only: work dispatched through " + "Celery sits in RabbitMQ and is not counted here", labels=["queue", "state"], ) oldest = GaugeMetricFamily( @@ -153,12 +159,15 @@ class ExecutionMetricGroup(MetricGroup): stamp ``execution_time`` on the terminal transition, so this is the completion time. ``modified_at`` is not, because later writes bump it (acknowledging an API result, for one, would make an old completion - count as new). An execution that ran longer than the lookback before - finishing is not counted; the default lookback (24h) is far above any - normal run. + count as new). The exception is a row terminalised without + ``execution_time`` (still 0), such as the undispatched-execution sweep's + ERROR: its ``modified_at`` is the sweep time, so it is used instead. An + execution that ran longer than the lookback before finishing is not + counted; the default lookback (24h) is far above any normal run. """ name = "executions" + statements = 2 def query(self) -> Snapshot: from workflow_manager.workflow_v2.models.execution import WorkflowExecution @@ -174,8 +183,15 @@ def query(self) -> Snapshot: .annotate(n=Count("id")) .order_by() ) - finished_at = ExpressionWrapper( - F("created_at") + F("execution_time") * Value(timedelta(seconds=1)), + finished_at = Case( + When( + execution_time__gt=0, + then=ExpressionWrapper( + F("created_at") + F("execution_time") * Value(timedelta(seconds=1)), + output_field=DateTimeField(), + ), + ), + default=F("modified_at"), output_field=DateTimeField(), ) recent = ( diff --git a/backend/platform_metrics/tests/fakes.py b/backend/platform_metrics/tests/fakes.py index 26ff7c69f2..fe44c5604f 100644 --- a/backend/platform_metrics/tests/fakes.py +++ b/backend/platform_metrics/tests/fakes.py @@ -33,6 +33,7 @@ def __init__(self) -> None: self.data: dict[str, str] = {} self.expiry: dict[str, int | None] = {} self.locks: set[str] = set() + self.hashes: dict[str, dict[str, int]] = {} self.down = False def _check(self) -> None: @@ -52,8 +53,42 @@ def set(self, name: str, value: str, ex: int | None = None) -> bool: def lock(self, name: str, timeout: float | None = None) -> FakeLock: self._check() + self.last_lock_timeout = timeout return FakeLock(self, name) + def pipeline(self, transaction: bool = True) -> FakePipeline: + return FakePipeline(self) + + def hincrby(self, name: str, key: str, amount: int = 1) -> int: + self._check() + bucket = self.hashes.setdefault(name, {}) + bucket[key] = bucket.get(key, 0) + amount + return bucket[key] + + def hgetall(self, name: str) -> dict[bytes, bytes]: + self._check() + return { + k.encode(): str(v).encode() for k, v in self.hashes.get(name, {}).items() + } + + +class FakePipeline: + """Queues calls and applies them on execute(), all or nothing.""" + + def __init__(self, redis: FakeRedis) -> None: + self._redis = redis + self._calls: list[tuple[str, tuple]] = [] + + def hincrby(self, *args) -> None: + self._calls.append(("hincrby", args)) + + def hgetall(self, *args) -> None: + self._calls.append(("hgetall", args)) + + def execute(self) -> list: + self._redis._check() + return [getattr(self._redis, name)(*args) for name, args in self._calls] + class StubGroup(MetricGroup): """A group whose fetch result (or failure) the test controls.""" diff --git a/backend/platform_metrics/tests/test_cache.py b/backend/platform_metrics/tests/test_cache.py index 1518962552..089138c1e4 100644 --- a/backend/platform_metrics/tests/test_cache.py +++ b/backend/platform_metrics/tests/test_cache.py @@ -227,3 +227,21 @@ def lock_after_other_pod_refreshed(name, timeout=None): assert cache.resolve(group).data == {"value": 8} assert group.fetches == 0 + + +def test_unreadable_entry_is_treated_as_missing_and_overwritten(cache, redis): + redis.data[DATA_KEY] = '{"data": {"value": 1}}' # no refreshed_at + group = StubGroup("g", value=6) + + assert cache.resolve(group).data == {"value": 6} + assert json.loads(redis.data[DATA_KEY])["data"] == {"value": 6} + + +def test_lock_outlives_the_longest_refresh_the_timeout_allows(cache, redis): + group = StubGroup("g") + group.statements = 3 + with override_settings(PLATFORM_METRICS_QUERY_TIMEOUT_MS=20000): + cache.resolve(group) + + # Three statements of up to 20s each, plus headroom. + assert redis.last_lock_timeout >= 60 diff --git a/backend/platform_metrics/tests/test_exporter.py b/backend/platform_metrics/tests/test_exporter.py index 13b0187324..825e0ee75f 100644 --- a/backend/platform_metrics/tests/test_exporter.py +++ b/backend/platform_metrics/tests/test_exporter.py @@ -4,7 +4,7 @@ from django.test import override_settings from prometheus_client.parser import text_string_to_metric_families -from platform_metrics.cache import SnapshotCache +from platform_metrics.cache import FailureCounter, SnapshotCache from platform_metrics.exporter import PlatformMetricsExporter from platform_metrics.tests.fakes import Clock, FakeRedis, StubGroup @@ -29,6 +29,7 @@ def _exporter(groups, redis, clock): return PlatformMetricsExporter( groups=groups, cache=SnapshotCache(client_factory=lambda: redis, clock=clock), + failures=FailureCounter(client_factory=lambda: redis), ) @@ -91,7 +92,11 @@ class BrokenCache: def resolve(self, group): raise ValueError("bug") - exporter = PlatformMetricsExporter(groups=[StubGroup("a")], cache=BrokenCache()) + exporter = PlatformMetricsExporter( + groups=[StubGroup("a")], + cache=BrokenCache(), + failures=FailureCounter(client_factory=lambda: redis), + ) samples = _samples(exporter) @@ -112,6 +117,53 @@ def test_snapshot_that_no_longer_renders_is_dropped_and_counted(redis, clock): assert samples[("unstract_metrics_refresh_failures_total", (("group", "a"),))] == 1 +FAILURES = "unstract_metrics_refresh_failures_total" + + +def test_failure_total_is_shared_across_processes(redis, clock): + """Every gunicorn worker reports the same total, so it never jumps back.""" + broken = StubGroup("a") + broken.error = RuntimeError("db down") + worker_1 = _exporter([broken], redis, clock) + worker_2 = _exporter([broken], redis, clock) + + _samples(worker_1) + clock.now += 60 + samples_2 = _samples(worker_2) + samples_1 = _samples(worker_1) + + assert samples_2[(FAILURES, (("group", "a"),))] == 2 + assert samples_1[(FAILURES, (("group", "a"),))] == 3 + + +def test_failures_while_redis_is_down_are_kept_and_flushed_later(redis, clock): + group = StubGroup("a") + exporter = _exporter([group], redis, clock) + redis.down = True + + samples = _samples(exporter) + + assert (FAILURES, (("group", "a"),)) not in samples, ( + "absent, not this process's partial count" + ) + assert samples[("unstract_metrics_cache_up", ())] == 0 + redis.down = False + clock.now += 60 + samples = _samples(exporter) + assert samples[(FAILURES, (("group", "a"),))] == 1 + assert samples[("unstract_metrics_cache_up", ())] == 1 + + +def test_unreadable_failure_total_does_not_break_the_scrape(redis, clock): + redis.hashes["platform_metrics:v1:refresh_failures"] = {"a": "1.5", "b": 2} + exporter = _exporter([StubGroup("a"), StubGroup("b")], redis, clock) + + samples = _samples(exporter) + + assert samples[(FAILURES, (("group", "a"),))] == 0 + assert samples[(FAILURES, (("group", "b"),))] == 2 + + def test_output_is_prometheus_text_with_help_and_type(redis, clock): text = _exporter([StubGroup("a")], redis, clock).render().decode() diff --git a/backend/platform_metrics/tests/test_groups_db.py b/backend/platform_metrics/tests/test_groups_db.py index 588bf40862..2c22b5a62d 100644 --- a/backend/platform_metrics/tests/test_groups_db.py +++ b/backend/platform_metrics/tests/test_groups_db.py @@ -30,7 +30,9 @@ def _message(queue, state, enqueued_ago=timedelta(0)): ) -def _execution(status, created_ago=timedelta(minutes=5), ran_for=None): +def _execution( + status, created_ago=timedelta(minutes=5), ran_for=None, modified_ago=timedelta(0) +): """An execution created ``created_ago``, terminal after ``ran_for``. ``modified_at`` is set to now on purpose: later writes (an API result being @@ -48,7 +50,7 @@ def _execution(status, created_ago=timedelta(minutes=5), ran_for=None): WorkflowExecution.objects.filter(id=ex.id).update( created_at=now - created_ago, execution_time=(ran_for or timedelta(0)).total_seconds(), - modified_at=now, + modified_at=now - modified_ago, ) return ex @@ -107,7 +109,16 @@ def test_active_and_recent_counts(self): created_ago=timedelta(hours=5), ran_for=timedelta(hours=4, minutes=50), ) - _execution(ExecutionStatus.ERROR.value) + _execution(ExecutionStatus.ERROR.value, ran_for=timedelta(minutes=4)) + # Terminalised by the undispatched sweep just now: execution_time is + # never stamped, so modified_at stands in as the finish time. + _execution(ExecutionStatus.ERROR.value, created_ago=timedelta(hours=2)) + # Swept, but before the window: not counted. + _execution( + ExecutionStatus.ERROR.value, + created_ago=timedelta(hours=5), + modified_ago=timedelta(hours=3), + ) # Finished before the window, touched since (e.g. its result was # acknowledged just now): not counted. _execution( @@ -125,7 +136,7 @@ def test_active_and_recent_counts(self): snapshot = ExecutionMetricGroup().fetch() assert snapshot["active"] == {"PENDING": 1, "EXECUTING": 2} - assert snapshot["recent"] == {"COMPLETED": 1, "ERROR": 1} + assert snapshot["recent"] == {"COMPLETED": 1, "ERROR": 2} class TestBatchGroup: