diff --git a/.env.example b/.env.example index de179240cc..072d22440d 100644 --- a/.env.example +++ b/.env.example @@ -238,6 +238,16 @@ HINDSIGHT_API_LOG_LEVEL=info # Expose async-operation queue + consolidation-backlog gauges on /metrics. # Runs periodic per-schema COUNT queries on a background task (disabled by default). # HINDSIGHT_API_METRICS_BACKLOG_ENABLED=true +# +# Runtime-stall observability (enabled by default). When a liveness probe fails, +# these tell you WHY: a blocked event loop vs DB connection-pool exhaustion. +# The loop watchdog logs the offending stack when the loop is unresponsive; the +# DB-pool acquire timing logs (and exposes hindsight.db.pool.waiting) when +# callers queue for a connection. Both are cheap; tune or disable if needed. +# HINDSIGHT_API_LOOP_WATCHDOG_ENABLED=false +# HINDSIGHT_API_LOOP_WATCHDOG_STALL_THRESHOLD_MS=1000 +# HINDSIGHT_API_LOOP_WATCHDOG_POLL_INTERVAL_MS=250 +# HINDSIGHT_API_DB_ACQUIRE_WARN_THRESHOLD_MS=1000 # ----------------------------------------------------------------------------- # Control Plane (Optional) diff --git a/hindsight-api-slim/hindsight_api/api/http.py b/hindsight-api-slim/hindsight_api/api/http.py index 7a339fc3e4..8682123ee0 100644 --- a/hindsight-api-slim/hindsight_api/api/http.py +++ b/hindsight-api-slim/hindsight_api/api/http.py @@ -3102,6 +3102,7 @@ async def lifespan(app: FastAPI): config = get_config() poller = None poller_task = None + loop_watchdog = None # Initialize OpenTelemetry metrics try: @@ -3145,6 +3146,12 @@ async def lifespan(app: FastAPI): metrics_collector.set_db_pool(memory._pool) logging.info("DB pool metrics configured") + # Start the event-loop stall watchdog (logs the culprit stack if a task + # blocks the loop, so a failing /health can be told apart from pool exhaustion). + from ..loop_watchdog import start_loop_watchdog + + loop_watchdog = start_loop_watchdog(asyncio.get_running_loop()) + # Start worker poller if the backend supports it. # All current backends (PostgreSQL, Oracle) support async worker/poller. if config.worker_enabled and memory._backend.supports_worker_poller: @@ -3189,6 +3196,10 @@ async def lifespan(app: FastAPI): yield + # Stop the loop watchdog first so it doesn't fire during teardown. + if loop_watchdog is not None: + loop_watchdog.stop() + # Shutdown worker poller if running if poller is not None: await poller.shutdown_graceful(timeout=30.0) diff --git a/hindsight-api-slim/hindsight_api/config.py b/hindsight-api-slim/hindsight_api/config.py index 0e42f8ae2b..f9eb53a773 100644 --- a/hindsight-api-slim/hindsight_api/config.py +++ b/hindsight-api-slim/hindsight_api/config.py @@ -493,6 +493,12 @@ def _resolve_operation_strict_schema(operation_env: str) -> bool: ENV_METRICS_INCLUDE_BANK_ID = "HINDSIGHT_API_METRICS_INCLUDE_BANK_ID" ENV_METRICS_BACKLOG_ENABLED = "HINDSIGHT_API_METRICS_BACKLOG_ENABLED" +# Runtime-stall observability (loop watchdog + DB pool acquire instrumentation) +ENV_LOOP_WATCHDOG_ENABLED = "HINDSIGHT_API_LOOP_WATCHDOG_ENABLED" +ENV_LOOP_WATCHDOG_STALL_THRESHOLD_MS = "HINDSIGHT_API_LOOP_WATCHDOG_STALL_THRESHOLD_MS" +ENV_LOOP_WATCHDOG_POLL_INTERVAL_MS = "HINDSIGHT_API_LOOP_WATCHDOG_POLL_INTERVAL_MS" +ENV_DB_ACQUIRE_WARN_THRESHOLD_MS = "HINDSIGHT_API_DB_ACQUIRE_WARN_THRESHOLD_MS" + # Vertex AI configuration ENV_LLM_VERTEXAI_PROJECT_ID = "HINDSIGHT_API_LLM_VERTEXAI_PROJECT_ID" ENV_LLM_VERTEXAI_REGION = "HINDSIGHT_API_LLM_VERTEXAI_REGION" @@ -1158,6 +1164,16 @@ def _parse_strategy_boosts(raw: str | None) -> dict[str, str]: DEFAULT_METRICS_INCLUDE_BANK_ID = False # Disabled by default to avoid high-cardinality OTel metric growth DEFAULT_METRICS_BACKLOG_ENABLED = False # Disabled by default: runs periodic per-schema COUNT queries +# Runtime-stall observability defaults. Both are cheap and on by default: the +# watchdog is a single background thread pinging the loop; the DB-pool acquire +# timing is a monotonic() delta per acquire. They turn a failing liveness probe +# from "pod restarted, cause unknown" into a logged root cause (blocked loop vs +# pool exhaustion). +DEFAULT_LOOP_WATCHDOG_ENABLED = True +DEFAULT_LOOP_WATCHDOG_STALL_THRESHOLD_MS = 1000 # log a stall once the loop is unresponsive this long +DEFAULT_LOOP_WATCHDOG_POLL_INTERVAL_MS = 250 # how often the watchdog thread pings the loop +DEFAULT_DB_ACQUIRE_WARN_THRESHOLD_MS = 1000 # log a warning when a pool acquire waits this long + # Audit log defaults DEFAULT_AUDIT_LOG_ENABLED = False # Disabled by default DEFAULT_AUDIT_LOG_ACTIONS = "" # Empty = audit all eligible actions @@ -2049,6 +2065,12 @@ class HindsightConfig: metrics_include_bank_id: bool metrics_backlog_enabled: bool + # Runtime-stall observability (static, server-level only) + loop_watchdog_enabled: bool + loop_watchdog_stall_threshold_ms: int + loop_watchdog_poll_interval_ms: int + db_acquire_warn_threshold_ms: int + # Audit log configuration # audit_log_enabled is hierarchical (env -> tenant -> bank): a deployment can # audit some banks and not others. The actions allowlist and retention window @@ -3168,6 +3190,18 @@ def from_env(cls) -> "HindsightConfig": in ("true", "1", "yes"), metrics_backlog_enabled=os.getenv(ENV_METRICS_BACKLOG_ENABLED, str(DEFAULT_METRICS_BACKLOG_ENABLED)).lower() in ("true", "1", "yes"), + # Runtime-stall observability (static, server-level only) + loop_watchdog_enabled=os.getenv(ENV_LOOP_WATCHDOG_ENABLED, str(DEFAULT_LOOP_WATCHDOG_ENABLED)).lower() + in ("true", "1", "yes"), + loop_watchdog_stall_threshold_ms=int( + os.getenv(ENV_LOOP_WATCHDOG_STALL_THRESHOLD_MS, str(DEFAULT_LOOP_WATCHDOG_STALL_THRESHOLD_MS)) + ), + loop_watchdog_poll_interval_ms=int( + os.getenv(ENV_LOOP_WATCHDOG_POLL_INTERVAL_MS, str(DEFAULT_LOOP_WATCHDOG_POLL_INTERVAL_MS)) + ), + db_acquire_warn_threshold_ms=int( + os.getenv(ENV_DB_ACQUIRE_WARN_THRESHOLD_MS, str(DEFAULT_DB_ACQUIRE_WARN_THRESHOLD_MS)) + ), # Audit log configuration (static, server-level only) audit_log_enabled=os.getenv(ENV_AUDIT_LOG_ENABLED, str(DEFAULT_AUDIT_LOG_ENABLED)).lower() == "true", audit_log_actions=[ diff --git a/hindsight-api-slim/hindsight_api/engine/db/oracle.py b/hindsight-api-slim/hindsight_api/engine/db/oracle.py index ed842ecdeb..0619dc28ab 100644 --- a/hindsight-api-slim/hindsight_api/engine/db/oracle.py +++ b/hindsight-api-slim/hindsight_api/engine/db/oracle.py @@ -23,6 +23,8 @@ from contextlib import asynccontextmanager from typing import Any, NamedTuple +from .pool_instrumentation import PoolStats, acquire_conn + class _OracleJSONEncoder(json.JSONEncoder): """JSON encoder that handles datetime and UUID objects.""" @@ -1246,6 +1248,7 @@ def __init__(self) -> None: # SESSION_USER so default-schema acquisitions can explicitly reset a # connection that was previously used for a tenant schema. self._default_schema: str | None = None + self._acquire_warn_threshold_s: float = 1.0 async def initialize( self, @@ -1261,6 +1264,10 @@ async def initialize( oracledb = _import_oracledb() self._oracledb = oracledb + from ...config import get_config + + self._acquire_warn_threshold_s = get_config().db_acquire_warn_threshold_ms / 1000.0 + # Parse URL-format DSN (oracle://user:pass@host:port/service) from urllib.parse import urlparse @@ -1322,10 +1329,23 @@ async def _set_session_schema(self, conn: Any) -> None: # expression" and aborts every acquire(). cursor.close() + def _pool_stats(self) -> PoolStats | None: + """Snapshot for slow-acquire logs, from oracledb pool attributes.""" + pool = self._pool + if pool is None: + return None + try: + busy = pool.busy + return PoolStats(in_use=busy, max=pool.max, idle=pool.opened - busy) + except Exception: + return None + @asynccontextmanager async def acquire(self) -> AsyncIterator[OracleConnection]: pool = self._ensure_pool() - conn = await pool.acquire() + conn = await acquire_conn( + pool.acquire(), pool_stats=self._pool_stats, warn_threshold_s=self._acquire_warn_threshold_s + ) try: await self._set_session_schema(conn) yield OracleConnection(conn) @@ -1341,7 +1361,9 @@ async def acquire(self) -> AsyncIterator[OracleConnection]: @asynccontextmanager async def transaction(self) -> AsyncIterator[OracleConnection]: pool = self._ensure_pool() - conn = await pool.acquire() + conn = await acquire_conn( + pool.acquire(), pool_stats=self._pool_stats, warn_threshold_s=self._acquire_warn_threshold_s + ) try: await self._set_session_schema(conn) yield OracleConnection(conn) diff --git a/hindsight-api-slim/hindsight_api/engine/db/pool_instrumentation.py b/hindsight-api-slim/hindsight_api/engine/db/pool_instrumentation.py new file mode 100644 index 0000000000..8a24bf3ee6 --- /dev/null +++ b/hindsight-api-slim/hindsight_api/engine/db/pool_instrumentation.py @@ -0,0 +1,137 @@ +"""Instrumentation for database connection-pool acquisition. + +asyncpg exposes pool *size* and *idle* counts, but not how many callers are +currently **queued waiting** for a connection — and that queue depth is the +signal that actually distinguishes a saturated pool from a healthy one. When the +pool is exhausted, ``/health`` (which itself acquires a connection to run +``SELECT 1``) blocks in ``pool.acquire()`` until a connection frees or the acquire +times out, so a liveness probe can fail **with the event loop completely idle**. + +This module tracks the process-wide count of in-flight acquisitions that have not +yet obtained a connection, and times each acquire so a slow one logs with full +pool stats. It is the DB-side counterpart to ``loop_watchdog`` (which covers loop +stalls); together, a stuck ``/health`` can be attributed to either a blocked loop +or pool exhaustion from the logs alone. + +The counter is a plain int mutated only from the event-loop thread (asyncpg +acquisitions are awaited on the loop), so no lock is needed. +""" + +from __future__ import annotations + +import logging +import time +from collections.abc import AsyncIterator, Callable +from contextlib import asynccontextmanager +from dataclasses import dataclass +from typing import Any + +logger = logging.getLogger("hindsight.db.pool") + +_waiting = 0 # callers currently blocked in pool.acquire(), process-wide + + +@dataclass(frozen=True, slots=True) +class PoolStats: + """Point-in-time connection-pool utilization snapshot.""" + + in_use: int + max: int + idle: int + + +def waiting_count() -> int: + """Number of callers currently blocked waiting to acquire a pooled connection.""" + return _waiting + + +@asynccontextmanager +async def instrument_acquire( + acquire_cm: Any, + *, + pool_stats: Callable[[], PoolStats | None] | None = None, + warn_threshold_s: float, +) -> AsyncIterator[Any]: + """Wrap a pool's ``acquire()`` context manager with wait tracking + slow-acquire logging. + + Args: + acquire_cm: an async context manager yielding a connection (e.g. the object + returned by ``asyncpg.Pool.acquire()``). + pool_stats: optional zero-arg callable returning a ``PoolStats`` snapshot for + the slow-acquire log line. + warn_threshold_s: log a warning when the acquire itself takes at least this long. + + Yields: + The acquired connection. + """ + global _waiting + _waiting += 1 + start = time.monotonic() + acquired = False + try: + async with acquire_cm as conn: + acquired = True + _waiting -= 1 + _record_acquire_wait(time.monotonic() - start, pool_stats, warn_threshold_s) + yield conn + finally: + # If __aenter__ raised (acquire timeout / cancellation), we never + # decremented above — do it here so the waiter count can't leak. + if not acquired: + _waiting -= 1 + + +async def acquire_conn( + acquire_awaitable: Any, + *, + pool_stats: Callable[[], PoolStats | None] | None = None, + warn_threshold_s: float, +) -> Any: + """Await a pool acquire that returns a connection, with wait tracking + slow log. + + For pools whose acquire is ``conn = await pool.acquire()`` (oracledb) rather than + an async context manager (asyncpg — use ``instrument_acquire`` for those). The + caller is responsible for releasing the returned connection. + """ + global _waiting + _waiting += 1 + start = time.monotonic() + try: + conn = await acquire_awaitable + finally: + _waiting -= 1 + _record_acquire_wait(time.monotonic() - start, pool_stats, warn_threshold_s) + return conn + + +def _record_acquire_wait( + wait_s: float, + pool_stats: Callable[[], PoolStats | None] | None, + warn_threshold_s: float, +) -> None: + try: + from ...metrics import get_metrics_collector + + get_metrics_collector().record_db_acquire_wait(wait_s) + except Exception: + pass + + if wait_s < warn_threshold_s: + return + + stats: PoolStats | None = None + if pool_stats is not None: + try: + stats = pool_stats() + except Exception: + stats = None + logger.warning( + "slow DB pool acquire: waited %.3fs for a connection " + "(in_use=%s max=%s idle=%s waiting=%s). The pool is likely saturated; " + "/health can stall on connection acquisition while the event loop is free.", + wait_s, + stats.in_use if stats else None, + stats.max if stats else None, + stats.idle if stats else None, + _waiting, + ) diff --git a/hindsight-api-slim/hindsight_api/engine/db/postgresql.py b/hindsight-api-slim/hindsight_api/engine/db/postgresql.py index a3816aa202..701e86bf4e 100644 --- a/hindsight-api-slim/hindsight_api/engine/db/postgresql.py +++ b/hindsight-api-slim/hindsight_api/engine/db/postgresql.py @@ -15,6 +15,7 @@ import asyncpg # noqa: F401 from .base import DatabaseBackend, DatabaseConnection +from .pool_instrumentation import PoolStats, instrument_acquire logger = logging.getLogger(__name__) @@ -76,6 +77,7 @@ def run_migrations(self, dsn: str, *, schema: str | None = None) -> None: def __init__(self) -> None: self._pool: asyncpg.Pool | None = None + self._acquire_warn_threshold_s: float = 1.0 async def initialize( self, @@ -88,6 +90,9 @@ async def initialize( statement_cache_size: int = 0, init_callback: Any | None = None, ) -> None: + from ...config import get_config + + self._acquire_warn_threshold_s = get_config().db_acquire_warn_threshold_ms / 1000.0 self._pool = await asyncpg.create_pool( dsn, min_size=min_size, @@ -121,16 +126,28 @@ async def shutdown(self) -> None: def is_ready(self) -> bool: return self._pool is not None + def _pool_stats(self) -> PoolStats | None: + """Snapshot for slow-acquire logs. in_use = live connections minus idle ones.""" + pool = self._pool + if pool is None: + return None + idle = pool.get_idle_size() + return PoolStats(in_use=pool.get_size() - idle, max=pool.get_max_size(), idle=idle) + @asynccontextmanager async def acquire(self) -> AsyncIterator[PostgresConnection]: pool = self._ensure_pool() - async with pool.acquire() as conn: + async with instrument_acquire( + pool.acquire(), pool_stats=self._pool_stats, warn_threshold_s=self._acquire_warn_threshold_s + ) as conn: yield PostgresConnection(conn) @asynccontextmanager async def transaction(self) -> AsyncIterator[PostgresConnection]: pool = self._ensure_pool() - async with pool.acquire() as conn: + async with instrument_acquire( + pool.acquire(), pool_stats=self._pool_stats, warn_threshold_s=self._acquire_warn_threshold_s + ) as conn: async with conn.transaction(): yield PostgresConnection(conn) diff --git a/hindsight-api-slim/hindsight_api/engine/memory_engine.py b/hindsight-api-slim/hindsight_api/engine/memory_engine.py index b2d37f8e0e..1e6fdba4b4 100644 --- a/hindsight-api-slim/hindsight_api/engine/memory_engine.py +++ b/hindsight-api-slim/hindsight_api/engine/memory_engine.py @@ -3278,15 +3278,45 @@ async def health_check(self) -> dict: try: backend = await self._get_backend() + # Time the acquire separately from the query. A slow acquire points at + # pool exhaustion (readiness), a slow query at the database itself; both + # are surfaced in the probe response so a failing/slow /health is + # self-diagnosing rather than an opaque restart. + acquire_start = time.monotonic() async with backend.acquire() as conn: + acquire_ms = (time.monotonic() - acquire_start) * 1000.0 result = await conn.fetchval("SELECT 1") - if result == 1: - return {"status": "healthy", "database": "connected"} - else: - return {"status": "unhealthy", "database": "unexpected response"} + health = { + "status": "healthy" if result == 1 else "unhealthy", + "database": "connected" if result == 1 else "unexpected response", + "db_acquire_ms": round(acquire_ms, 1), + } + health.update(self._pool_health_stats(backend)) + return health except Exception as e: return {"status": "unhealthy", "database": "error", "error": str(e)} + @staticmethod + def _pool_health_stats(backend: Any) -> dict: + """Best-effort pool utilization for the health payload (never raises).""" + stats: dict[str, Any] = {} + try: + from .db.pool_instrumentation import waiting_count + + stats["db_pool_waiting"] = waiting_count() + except Exception: + pass + try: + pool_stats = getattr(backend, "_pool_stats", None) + snapshot = pool_stats() if callable(pool_stats) else None + if snapshot is not None: + stats["db_pool_in_use"] = snapshot.in_use + stats["db_pool_max"] = snapshot.max + stats["db_pool_idle"] = snapshot.idle + except Exception: + pass + return stats + async def close(self): """Close the connection pool and shutdown background workers.""" logger.info("close() started") diff --git a/hindsight-api-slim/hindsight_api/loop_watchdog.py b/hindsight-api-slim/hindsight_api/loop_watchdog.py new file mode 100644 index 0000000000..ac65ab6de1 --- /dev/null +++ b/hindsight-api-slim/hindsight_api/loop_watchdog.py @@ -0,0 +1,147 @@ +"""Event-loop stall watchdog. + +Hindsight's worker and API run the ``/health`` handler and all task work on one +asyncio event loop. If something does blocking (synchronous) work on that loop — +CPU-bound parsing, a mis-offloaded SDK call, a third-party library that signs a +request inline — the loop stops servicing coroutines, ``/health`` can't be +scheduled, and a Kubernetes liveness probe fails even though the process is "up". + +This watchdog makes that condition self-diagnosing. It runs in a **separate OS +thread** (deliberately: a coroutine-based monitor would be frozen by the very +stall it's trying to observe), pings the loop, and when the loop fails to service +the ping within a threshold it logs the loop thread's current stack — naming the +exact frame that is blocking. It never raises and never touches the loop's work; +it only observes. Unlike monkeypatch-based blocking detectors it works with +uvloop, because it relies only on ``loop.call_soon_threadsafe`` and +``sys._current_frames()``. + +It is the loop-side counterpart to the DB-pool acquire instrumentation +(``engine/db/pool_instrumentation.py``): together they let a stuck ``/health`` be +attributed to either a blocked loop or connection-pool exhaustion from the logs +alone. +""" + +from __future__ import annotations + +import logging +import sys +import threading +import time +import traceback +from collections.abc import Callable + +logger = logging.getLogger("hindsight.loop_watchdog") + + +def start_loop_watchdog(loop) -> "LoopWatchdog | None": + """Build and start a watchdog for ``loop`` from config, or return None if disabled. + + Call this once, from inside the running loop's process (worker CLI / API lifespan), + and call ``.stop()`` on the returned handle at shutdown. + """ + from .config import get_config + + config = get_config() + if not config.loop_watchdog_enabled: + return None + watchdog = LoopWatchdog( + loop, + stall_threshold_s=config.loop_watchdog_stall_threshold_ms / 1000.0, + poll_interval_s=config.loop_watchdog_poll_interval_ms / 1000.0, + ) + watchdog.start() + return watchdog + + +class LoopWatchdog: + """Detects event-loop stalls from an off-loop thread and logs the culprit stack. + + Args: + loop: the asyncio event loop to monitor. + stall_threshold_s: log when the loop takes at least this long to service a ping. + poll_interval_s: how often to ping the loop. + on_stall: optional callback ``(blocked_for_s, stack_text)`` invoked on each + detected stall instead of the default log+metric path. Used for testing. + """ + + def __init__( + self, + loop, + *, + stall_threshold_s: float = 1.0, + poll_interval_s: float = 0.25, + on_stall: Callable[[float, str], None] | None = None, + ) -> None: + self._loop = loop + self._stall_threshold_s = stall_threshold_s + self._poll_interval_s = poll_interval_s + self._on_stall = on_stall + self._stop = threading.Event() + self._loop_thread_id: int | None = None + self._thread = threading.Thread(target=self._run, name="loop-watchdog", daemon=True) + self._started = False + + def start(self) -> None: + """Start monitoring. Must not block the loop — the id is captured via pings. + + When called from the loop thread itself (the normal case: worker ``run()`` / + API lifespan), ``threading.get_ident()`` is already the loop thread id, so we + seed it here; each ping then re-affirms it authoritatively. We deliberately do + NOT schedule-and-wait for a callback: that would deadlock, because the loop + can't run the callback while ``start()`` is blocking it. + """ + self._loop_thread_id = threading.get_ident() + self._started = True + self._thread.start() + logger.info( + "Loop watchdog started (stall_threshold=%.2fs, poll_interval=%.2fs)", + self._stall_threshold_s, + self._poll_interval_s, + ) + + def stop(self) -> None: + self._stop.set() + if self._started and self._thread.is_alive(): + self._thread.join(timeout=self._poll_interval_s + self._stall_threshold_s + 1.0) + + def _run(self) -> None: + while not self._stop.wait(self._poll_interval_s): + serviced = threading.Event() + sent_at = time.monotonic() + + def _ping() -> None: + # Runs on the loop thread — capture its id authoritatively, then + # signal that the loop serviced this ping. + self._loop_thread_id = threading.get_ident() + serviced.set() + + try: + self._loop.call_soon_threadsafe(_ping) + except RuntimeError: + return # loop closed — nothing left to watch + if not serviced.wait(self._stall_threshold_s): + self._report(sent_at) + # Block until the loop finally services the ping so we emit one + # report per stall, not one per poll while it stays blocked. + serviced.wait() + + def _report(self, sent_at: float) -> None: + frame = sys._current_frames().get(self._loop_thread_id or -1) + stack = "".join(traceback.format_stack(frame)) if frame is not None else "" + blocked_for = time.monotonic() - sent_at + if self._on_stall is not None: + self._on_stall(blocked_for, stack) + return + logger.warning( + "EVENT LOOP BLOCKED for >= %.2fs (%.2fs and counting). The loop is not " + "servicing coroutines — /health cannot be scheduled. Blocking frame:\n%s", + self._stall_threshold_s, + blocked_for, + stack, + ) + try: + from .metrics import get_metrics_collector + + get_metrics_collector().record_loop_stall(blocked_for) + except Exception: + pass diff --git a/hindsight-api-slim/hindsight_api/metrics.py b/hindsight-api-slim/hindsight_api/metrics.py index 5f25c1b951..8b04ccdc75 100644 --- a/hindsight-api-slim/hindsight_api/metrics.py +++ b/hindsight-api-slim/hindsight_api/metrics.py @@ -292,6 +292,14 @@ def record_http_request(self, method: str, endpoint: str, status_code_getter: Ca """Context manager to record HTTP request metrics.""" raise NotImplementedError + def record_db_acquire_wait(self, wait_seconds: float): + """Record how long a caller waited to acquire a pooled DB connection.""" + raise NotImplementedError + + def record_loop_stall(self, stall_seconds: float): + """Record a detected event-loop stall (blocked longer than the watchdog threshold).""" + raise NotImplementedError + def set_db_pool(self, pool: "asyncpg.Pool"): """Set the database pool for metrics collection.""" pass @@ -345,6 +353,14 @@ def record_http_request(self, method: str, endpoint: str, status_code_getter: Ca """No-op HTTP request recording.""" yield + def record_db_acquire_wait(self, wait_seconds: float): + """No-op DB acquire-wait recording.""" + pass + + def record_loop_stall(self, stall_seconds: float): + """No-op loop-stall recording.""" + pass + class MetricsCollector(MetricsCollectorBase): """ @@ -426,6 +442,25 @@ def __init__(self): unit="requests", ) + # Runtime-stall observability: how long callers wait for a pooled DB + # connection (pool-exhaustion signal) and detected event-loop stalls + # (blocked-loop signal). See loop_watchdog.py and db/pool_instrumentation.py. + self.db_acquire_wait = self.meter.create_histogram( + name="hindsight.db.pool.acquire_wait", + description="Time spent waiting to acquire a pooled database connection", + unit="s", + ) + self.event_loop_stalls = self.meter.create_counter( + name="hindsight.event_loop.stalls", + description="Number of detected event-loop stalls (loop blocked past the watchdog threshold)", + unit="stalls", + ) + self.event_loop_stall_duration = self.meter.create_histogram( + name="hindsight.event_loop.stall_duration", + description="Duration of detected event-loop stalls in seconds", + unit="s", + ) + # Process metrics (observable gauges - collected on scrape) self._setup_process_metrics() @@ -646,6 +681,15 @@ def record_http_request(self, method: str, endpoint: str, status_code_getter: Ca # Decrement in-progress self.http_requests_in_progress.add(-1, base_attributes) + def record_db_acquire_wait(self, wait_seconds: float): + """Record how long a caller waited to acquire a pooled DB connection.""" + self.db_acquire_wait.record(wait_seconds) + + def record_loop_stall(self, stall_seconds: float): + """Record a detected event-loop stall. Called from the watchdog thread.""" + self.event_loop_stalls.add(1) + self.event_loop_stall_duration.record(stall_seconds) + def _setup_process_metrics(self): """Set up observable gauges for process metrics.""" if _resource_mod is None: @@ -771,6 +815,20 @@ def get_pool_max_size(_options): except Exception: pass + def get_pool_waiting(_options): + """Number of callers currently blocked waiting to acquire a connection. + + asyncpg does not expose this; it's tracked in db/pool_instrumentation.py. + This is the gauge that actually distinguishes pool exhaustion (a high, + sustained value) from a merely busy-but-healthy pool. + """ + try: + from .engine.db.pool_instrumentation import waiting_count + + yield metrics.Observation(waiting_count()) + except Exception: + pass + # Create observable gauges for pool metrics self.meter.create_observable_gauge( name="hindsight.db.pool.size", @@ -800,6 +858,13 @@ def get_pool_max_size(_options): unit="{connections}", ) + self.meter.create_observable_gauge( + name="hindsight.db.pool.waiting", + callbacks=[get_pool_waiting], + description="Callers currently blocked waiting to acquire a pooled connection", + unit="{connections}", + ) + def _setup_backlog_metrics(self): """Observable gauges for the async-operation queue and the consolidation backlog. diff --git a/hindsight-api-slim/hindsight_api/worker/main.py b/hindsight-api-slim/hindsight_api/worker/main.py index 7799d32ea8..51a9c6d4b5 100644 --- a/hindsight-api-slim/hindsight_api/worker/main.py +++ b/hindsight-api-slim/hindsight_api/worker/main.py @@ -320,6 +320,13 @@ def signal_handler(): ) server = uvicorn.Server(uvicorn_config) + # Start the event-loop stall watchdog: if a task blocks the loop, this + # logs the culprit stack so a failing /health can be attributed to a + # blocked loop (vs DB-pool exhaustion, which the pool instrumentation logs). + from ..loop_watchdog import start_loop_watchdog + + loop_watchdog = start_loop_watchdog(loop) + # Run the poller and HTTP server concurrently poller_task = asyncio.create_task(poller.run()) http_task = asyncio.create_task(server.serve()) @@ -333,6 +340,9 @@ def signal_handler(): print("\nReceived interrupt, initiating graceful shutdown...") # Graceful shutdown + if loop_watchdog is not None: + loop_watchdog.stop() + print("Shutting down HTTP server...") server.should_exit = True diff --git a/hindsight-api-slim/tests/test_loop_watchdog.py b/hindsight-api-slim/tests/test_loop_watchdog.py new file mode 100644 index 0000000000..765349a799 --- /dev/null +++ b/hindsight-api-slim/tests/test_loop_watchdog.py @@ -0,0 +1,73 @@ +"""Unit tests for the event-loop stall watchdog. + +Deterministic (no LLM): we block the loop with a synchronous sleep and assert the +off-loop watchdog thread detects it and captures the culprit stack, and that +genuinely off-loop work does not trip it. +""" + +import asyncio +import time + +from hindsight_api.loop_watchdog import LoopWatchdog + + +async def test_watchdog_detects_on_loop_block(): + stalls: list[tuple[float, str]] = [] + wd = LoopWatchdog( + asyncio.get_running_loop(), + stall_threshold_s=0.1, + poll_interval_s=0.02, + on_stall=lambda dur, stack: stalls.append((dur, stack)), + ) + wd.start() + try: + await asyncio.sleep(0.1) # let the watchdog settle into steady polling + time.sleep(0.6) # BLOCK the event loop synchronously + await asyncio.sleep(0.2) # give the watchdog a chance to have reported + finally: + wd.stop() + + assert stalls, "watchdog did not detect the synchronous loop block" + blocked_for, stack = stalls[0] + assert blocked_for >= 0.1 + # The captured stack must name the frame that was blocking the loop. + assert "test_watchdog_detects_on_loop_block" in stack + + +async def test_watchdog_ignores_offloop_work(): + stalls: list[tuple[float, str]] = [] + wd = LoopWatchdog( + asyncio.get_running_loop(), + stall_threshold_s=0.1, + poll_interval_s=0.02, + on_stall=lambda dur, stack: stalls.append((dur, stack)), + ) + wd.start() + try: + await asyncio.sleep(0.1) + # Sync sleep offloaded to a thread — the loop stays free, exactly the + # pattern litellm uses for boto3 credential resolution. + await asyncio.get_running_loop().run_in_executor(None, time.sleep, 0.5) + await asyncio.sleep(0.1) + finally: + wd.stop() + + assert not stalls, f"watchdog falsely reported a stall for off-loop work: {stalls}" + + +async def test_watchdog_quiet_when_loop_responsive(): + stalls: list[tuple[float, str]] = [] + wd = LoopWatchdog( + asyncio.get_running_loop(), + stall_threshold_s=0.1, + poll_interval_s=0.02, + on_stall=lambda dur, stack: stalls.append((dur, stack)), + ) + wd.start() + try: + for _ in range(10): + await asyncio.sleep(0.03) + finally: + wd.stop() + + assert not stalls, f"watchdog reported a stall on a responsive loop: {stalls}" diff --git a/hindsight-api-slim/tests/test_metrics.py b/hindsight-api-slim/tests/test_metrics.py index cb8d8fe3b9..186b3e5f99 100644 --- a/hindsight-api-slim/tests/test_metrics.py +++ b/hindsight-api-slim/tests/test_metrics.py @@ -66,14 +66,11 @@ class TestMetricsCollector: def mock_meter(self): """Create a mock meter for testing.""" meter = MagicMock() - # Create separate mocks for each histogram (operation_duration, llm_duration, http_request_duration) - histogram_mocks = [MagicMock(), MagicMock(), MagicMock()] - meter.create_histogram.side_effect = histogram_mocks - # Create separate mocks for each counter (operation_total, llm_tokens_input, - # llm_tokens_output, llm_calls_total, llm_tokens_cached_input, - # llm_tokens_thoughts, http_requests_total) - counter_mocks = [MagicMock() for _ in range(7)] - meter.create_counter.side_effect = counter_mocks + # Return a fresh mock per instrument, regardless of how many the collector + # creates — so adding a histogram/counter to MetricsCollector never requires + # bumping a hard-coded count here. + meter.create_histogram.side_effect = lambda *a, **k: MagicMock() + meter.create_counter.side_effect = lambda *a, **k: MagicMock() return meter @pytest.fixture @@ -361,14 +358,11 @@ class TestLLMMetrics: def mock_meter(self): """Create a mock meter for testing.""" meter = MagicMock() - # Create separate mocks for each histogram (operation_duration, llm_duration, http_request_duration) - histogram_mocks = [MagicMock(), MagicMock(), MagicMock()] - meter.create_histogram.side_effect = histogram_mocks - # Create separate mocks for each counter (operation_total, llm_tokens_input, - # llm_tokens_output, llm_calls_total, llm_tokens_cached_input, - # llm_tokens_thoughts, http_requests_total) - counter_mocks = [MagicMock() for _ in range(7)] - meter.create_counter.side_effect = counter_mocks + # Return a fresh mock per instrument, regardless of how many the collector + # creates — so adding a histogram/counter to MetricsCollector never requires + # bumping a hard-coded count here. + meter.create_histogram.side_effect = lambda *a, **k: MagicMock() + meter.create_counter.side_effect = lambda *a, **k: MagicMock() return meter @pytest.fixture diff --git a/hindsight-api-slim/tests/test_pool_instrumentation.py b/hindsight-api-slim/tests/test_pool_instrumentation.py new file mode 100644 index 0000000000..04caded96e --- /dev/null +++ b/hindsight-api-slim/tests/test_pool_instrumentation.py @@ -0,0 +1,115 @@ +"""Unit tests for DB pool acquire instrumentation (waiter counter + slow-acquire log). + +Deterministic (no DB): we drive the instrumentation with fake acquire context +managers / awaitables and assert the process-wide waiter count is accurate through +success, mid-acquire, and failure, and that a slow acquire logs with pool stats. +""" + +import asyncio +import logging + +import pytest + +from hindsight_api.engine.db.pool_instrumentation import ( + PoolStats, + acquire_conn, + instrument_acquire, + waiting_count, +) + + +class _GatedAcquire: + """Async CM whose __aenter__ blocks until released, to observe mid-acquire state.""" + + def __init__(self, entered: asyncio.Event, release: asyncio.Event): + self._entered = entered + self._release = release + + async def __aenter__(self): + self._entered.set() + await self._release.wait() + return "conn" + + async def __aexit__(self, *exc): + return False + + +class _BoomAcquire: + async def __aenter__(self): + raise RuntimeError("acquire failed") + + async def __aexit__(self, *exc): + return False + + +async def test_instrument_acquire_tracks_waiters(): + assert waiting_count() == 0 + entered = asyncio.Event() + release = asyncio.Event() + + async def use(): + async with instrument_acquire(_GatedAcquire(entered, release), warn_threshold_s=999) as conn: + assert conn == "conn" + # Once acquired, the caller is no longer waiting. + assert waiting_count() == 0 + + task = asyncio.create_task(use()) + await entered.wait() + # Blocked inside __aenter__ -> counted as one waiter. + assert waiting_count() == 1 + release.set() + await task + assert waiting_count() == 0 + + +async def test_instrument_acquire_decrements_on_failure(): + assert waiting_count() == 0 + with pytest.raises(RuntimeError): + async with instrument_acquire(_BoomAcquire(), warn_threshold_s=999): + pass + # The waiter count must not leak when the acquire itself raises. + assert waiting_count() == 0 + + +async def test_slow_acquire_logs_with_pool_stats(caplog): + def stats(): + return PoolStats(in_use=10, max=10, idle=0) + + class _Instant: + async def __aenter__(self): + return "conn" + + async def __aexit__(self, *exc): + return False + + # threshold 0 => any wait (>= 0s) is logged. + with caplog.at_level(logging.WARNING, logger="hindsight.db.pool"): + async with instrument_acquire(_Instant(), pool_stats=stats, warn_threshold_s=0.0) as conn: + assert conn == "conn" + + msgs = [r.getMessage() for r in caplog.records] + assert any("slow DB pool acquire" in m for m in msgs) + assert any("in_use=10" in m and "max=10" in m for m in msgs) + + +async def test_acquire_conn_await_style_tracks_and_returns(): + assert waiting_count() == 0 + + async def acquire_awaitable(): + await asyncio.sleep(0) + return "conn" + + conn = await acquire_conn(acquire_awaitable(), warn_threshold_s=999) + assert conn == "conn" + assert waiting_count() == 0 + + +async def test_acquire_conn_decrements_on_failure(): + assert waiting_count() == 0 + + async def boom(): + raise RuntimeError("acquire failed") + + with pytest.raises(RuntimeError): + await acquire_conn(boom(), warn_threshold_s=999) + assert waiting_count() == 0 diff --git a/hindsight-docs/docs/developer/configuration.md b/hindsight-docs/docs/developer/configuration.md index 0fea0c97d1..6adb245622 100644 --- a/hindsight-docs/docs/developer/configuration.md +++ b/hindsight-docs/docs/developer/configuration.md @@ -1816,6 +1816,28 @@ See `scripts/dev/grafana/README.md` for detailed setup instructions. Other options: See `scripts/dev/openlit/README.md` for OpenLIT or `scripts/dev/jaeger/README.md` for standalone Jaeger. +### Runtime-Stall Diagnostics + +The API and worker run the `/health` handler and all task work on a single asyncio +event loop, and `/health` acquires a database connection. So a liveness probe can +fail for two very different reasons: the **event loop is blocked** by synchronous +work (a restart helps), or the **connection pool is exhausted** and `/health` can't +get a connection even though the loop is idle (a restart usually doesn't help). These +diagnostics tell the two apart from the logs and metrics alone, instead of leaving +you with an opaque restart. + +| Variable | Description | Default | +|----------|-------------|---------| +| `HINDSIGHT_API_LOOP_WATCHDOG_ENABLED` | Run a background thread that detects event-loop stalls and logs the blocking stack. Also emits `hindsight_event_loop_stalls` / `hindsight_event_loop_stall_duration`. | `true` | +| `HINDSIGHT_API_LOOP_WATCHDOG_STALL_THRESHOLD_MS` | Log a stall once the loop is unresponsive for at least this long. | `1000` | +| `HINDSIGHT_API_LOOP_WATCHDOG_POLL_INTERVAL_MS` | How often the watchdog thread pings the loop. | `250` | +| `HINDSIGHT_API_DB_ACQUIRE_WARN_THRESHOLD_MS` | Log a warning (with pool stats) when acquiring a pooled connection waits at least this long. | `1000` | + +The DB-pool acquire path also exposes `hindsight_db_pool_waiting` (callers currently +queued for a connection) and the `hindsight_db_pool_acquire_wait` histogram. A slow +or failing `/health` response additionally carries `db_acquire_ms`, `db_pool_waiting`, +`db_pool_in_use`, and `db_pool_max` for triage. + ### Metrics Hindsight exposes Prometheus metrics at the `/metrics` endpoint, including: diff --git a/hindsight-embed/hindsight_embed/env.example b/hindsight-embed/hindsight_embed/env.example index de179240cc..072d22440d 100644 --- a/hindsight-embed/hindsight_embed/env.example +++ b/hindsight-embed/hindsight_embed/env.example @@ -238,6 +238,16 @@ HINDSIGHT_API_LOG_LEVEL=info # Expose async-operation queue + consolidation-backlog gauges on /metrics. # Runs periodic per-schema COUNT queries on a background task (disabled by default). # HINDSIGHT_API_METRICS_BACKLOG_ENABLED=true +# +# Runtime-stall observability (enabled by default). When a liveness probe fails, +# these tell you WHY: a blocked event loop vs DB connection-pool exhaustion. +# The loop watchdog logs the offending stack when the loop is unresponsive; the +# DB-pool acquire timing logs (and exposes hindsight.db.pool.waiting) when +# callers queue for a connection. Both are cheap; tune or disable if needed. +# HINDSIGHT_API_LOOP_WATCHDOG_ENABLED=false +# HINDSIGHT_API_LOOP_WATCHDOG_STALL_THRESHOLD_MS=1000 +# HINDSIGHT_API_LOOP_WATCHDOG_POLL_INTERVAL_MS=250 +# HINDSIGHT_API_DB_ACQUIRE_WARN_THRESHOLD_MS=1000 # ----------------------------------------------------------------------------- # Control Plane (Optional) diff --git a/skills/hindsight-docs/references/developer/configuration.md b/skills/hindsight-docs/references/developer/configuration.md index 6d2f259163..411c589221 100644 --- a/skills/hindsight-docs/references/developer/configuration.md +++ b/skills/hindsight-docs/references/developer/configuration.md @@ -1816,6 +1816,28 @@ See `scripts/dev/grafana/README.md` for detailed setup instructions. Other options: See `scripts/dev/openlit/README.md` for OpenLIT or `scripts/dev/jaeger/README.md` for standalone Jaeger. +### Runtime-Stall Diagnostics + +The API and worker run the `/health` handler and all task work on a single asyncio +event loop, and `/health` acquires a database connection. So a liveness probe can +fail for two very different reasons: the **event loop is blocked** by synchronous +work (a restart helps), or the **connection pool is exhausted** and `/health` can't +get a connection even though the loop is idle (a restart usually doesn't help). These +diagnostics tell the two apart from the logs and metrics alone, instead of leaving +you with an opaque restart. + +| Variable | Description | Default | +|----------|-------------|---------| +| `HINDSIGHT_API_LOOP_WATCHDOG_ENABLED` | Run a background thread that detects event-loop stalls and logs the blocking stack. Also emits `hindsight_event_loop_stalls` / `hindsight_event_loop_stall_duration`. | `true` | +| `HINDSIGHT_API_LOOP_WATCHDOG_STALL_THRESHOLD_MS` | Log a stall once the loop is unresponsive for at least this long. | `1000` | +| `HINDSIGHT_API_LOOP_WATCHDOG_POLL_INTERVAL_MS` | How often the watchdog thread pings the loop. | `250` | +| `HINDSIGHT_API_DB_ACQUIRE_WARN_THRESHOLD_MS` | Log a warning (with pool stats) when acquiring a pooled connection waits at least this long. | `1000` | + +The DB-pool acquire path also exposes `hindsight_db_pool_waiting` (callers currently +queued for a connection) and the `hindsight_db_pool_acquire_wait` histogram. A slow +or failing `/health` response additionally carries `db_acquire_ms`, `db_pool_waiting`, +`db_pool_in_use`, and `db_pool_max` for triage. + ### Metrics Hindsight exposes Prometheus metrics at the `/metrics` endpoint, including: