feat(observability): slow-query log + Prometheus metrics (#324)

* feat(observability): slow-query log + Prometheus metrics (#293)

- Add SQLAlchemy before/after_cursor_execute event listeners in database.py
- Log slow_query warning (structlog) when query duration exceeds 500 ms
- Expose http_db_query_seconds Histogram (every query) via Prometheus
- Expose db_slow_queries_total Counter (slow queries only) via Prometheus
- Threshold configurable via SLOW_QUERY_THRESHOLD_MS env var (default 500)
- Add test_slow_query.py with 25 tests covering all instrumentation paths
- Update test_db_pool.py to patch _register_slow_query_listener where
  create_engine is mocked, fixing pre-existing test compatibility

Closes #293

* fix(observability): guard Prometheus metrics against duplicate registration on reload

test_alembic_env.py calls importlib.reload(database) to test URL
construction. Each reload tried to re-register DB_QUERY_DURATION and
DB_SLOW_QUERY_COUNTER in the global Prometheus registry, raising:

  ValueError: Duplicated timeseries in CollectorRegistry

Fix: add _get_or_create_histogram / _get_or_create_counter helpers that
look up an existing collector by name in REGISTRY._names_to_collectors
and return it instead of constructing a new one. The metric objects are
functionally identical across reloads so all instrumentation continues
to work correctly.

Author: DammyAji

quantara/web_app/db/database.py

@@ -4,19 +4,147 @@
 Reads PostgreSQL connection parameters from environment variables and
 exposes the SQLAlchemy engine, session factory, and declarative base
 for use throughout the web application.
+
+Observability
+─────────────
+A SQLAlchemy ``before_cursor_execute`` / ``after_cursor_execute`` event pair
+is registered on every engine produced by this module.  Any statement that
+takes longer than SLOW_QUERY_THRESHOLD_MS (default 500 ms) is logged at
+WARNING level with the ``slow_query`` event key and the actual duration.
+
+The Prometheus counter ``http_db_query_seconds`` is incremented for **every**
+finished query, regardless of duration, so scrape targets can compute a
+request-rate baseline.  A separate ``db_slow_queries_total`` counter tracks
+only the slow queries, making it easy to build an alert rule such as::
+
+    rate(db_slow_queries_total[5m]) > 0
 """
 
 import os
+import time
 from typing import Generator
 
 from dotenv import load_dotenv
-from sqlalchemy import create_engine
+from prometheus_client import Counter, Histogram
+from sqlalchemy import create_engine, event
+from sqlalchemy.engine import Connection
 from sqlalchemy.orm import Session, declarative_base, sessionmaker
 
+from web_app.utils.logger import get_logger
+
+logger = get_logger(__name__)
+
+# ── Prometheus metrics ────────────────────────────────────────────────────────
+# Guard against duplicate registration when the module is reloaded (e.g. in
+# tests that call ``importlib.reload``).  prometheus_client raises ValueError
+# on duplicate metric names, so we reuse any collector that is already
+# registered under the same name instead of creating a new one.
+
+from prometheus_client import REGISTRY as _PROM_REGISTRY  # noqa: E402
+
+
+def _get_or_create_histogram(name, documentation, labelnames, buckets):
+    """Return an existing Histogram from the registry or create a new one."""
+    existing = _PROM_REGISTRY._names_to_collectors.get(name)
+    if existing is not None:
+        return existing
+    return Histogram(name, documentation, labelnames, buckets=buckets)
+
+
+def _get_or_create_counter(name, documentation):
+    """Return an existing Counter from the registry or create a new one."""
+    # prometheus_client stores counters under '<name>_total' in newer versions.
+    existing = (
+        _PROM_REGISTRY._names_to_collectors.get(name)
+        or _PROM_REGISTRY._names_to_collectors.get(f"{name}_total")
+    )
+    if existing is not None:
+        return existing
+    return Counter(name, documentation)
+
+
+#: Total seconds spent executing database queries (histogram).
+#: Label ``query_type`` is always "sql" — extend if you need finer granularity.
+DB_QUERY_DURATION = _get_or_create_histogram(
+    "http_db_query_seconds",
+    "Duration of individual database queries in seconds",
+    ["query_type"],
+    buckets=(0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0),
+)
+
+#: Incremented every time a query exceeds SLOW_QUERY_THRESHOLD_MS.
+DB_SLOW_QUERY_COUNTER = _get_or_create_counter(
+    "db_slow_queries_total",
+    "Number of database queries that exceeded the slow-query threshold",
+)
+
+# Threshold in milliseconds.  Override via SLOW_QUERY_THRESHOLD_MS env var.
+SLOW_QUERY_THRESHOLD_MS: float = float(
+    os.environ.get("SLOW_QUERY_THRESHOLD_MS", "500")
+)
+
+# ── Module-level engine / session state ──────────────────────────────────────
+
 engine = None
 SessionLocal = None
 Base = declarative_base()
 
+
+# ── Slow-query instrumentation ────────────────────────────────────────────────
+
+
+def _before_cursor_execute(  # pylint: disable=too-many-arguments,unused-argument
+    conn: Connection,
+    cursor,
+    statement: str,
+    parameters,
+    context,
+    executemany: bool,
+) -> None:
+    """Record the wall-clock start time on the connection's info dict."""
+    conn.info.setdefault("_query_start_time", []).append(time.perf_counter())
+
+
+def _after_cursor_execute(  # pylint: disable=too-many-arguments,unused-argument
+    conn: Connection,
+    cursor,
+    statement: str,
+    parameters,
+    context,
+    executemany: bool,
+) -> None:
+    """Compute elapsed time, update Prometheus metrics, and log slow queries."""
+    start_times: list = conn.info.get("_query_start_time", [])
+    if not start_times:
+        return
+
+    start = start_times.pop()
+    elapsed_seconds: float = time.perf_counter() - start
+    elapsed_ms: float = elapsed_seconds * 1000.0
+
+    # Always record to the histogram.
+    DB_QUERY_DURATION.labels(query_type="sql").observe(elapsed_seconds)
+
+    # Log and count slow queries.
+    if elapsed_ms > SLOW_QUERY_THRESHOLD_MS:
+        DB_SLOW_QUERY_COUNTER.inc()
+        logger.warning(
+            "slow_query",
+            duration_ms=round(elapsed_ms, 3),
+            threshold_ms=SLOW_QUERY_THRESHOLD_MS,
+            statement=statement[:500],  # truncate to avoid huge log lines
+        )
+
+
+def _register_slow_query_listener(eng) -> None:
+    """Attach the before/after cursor-execute listeners to *eng*."""
+    event.listen(eng, "before_cursor_execute", _before_cursor_execute)
+    event.listen(eng, "after_cursor_execute", _after_cursor_execute)
+
+
+# ── URL / pool helpers ────────────────────────────────────────────────────────
+
+
 def get_database_url() -> str:
     """Construct and return the database URL from environment variables."""
     load_dotenv(override=False)
@@ -31,6 +159,7 @@ def get_database_url() -> str:
         f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_SERVER}:{DB_PORT}/{DB_NAME}"
     )
 
+
 def _engine_pool_kwargs() -> dict:
     """Return the standard pool-related kwargs for every SQLAlchemy engine.
 
@@ -50,22 +179,37 @@ def _engine_pool_kwargs() -> dict:
     }
 
 
+# ── Engine factories ──────────────────────────────────────────────────────────
+
+
 def init_engine(db_url: str = None):
-    """Construct a SQLAlchemy engine that uses the project's pool policy."""
+    """Construct a SQLAlchemy engine that uses the project's pool policy.
+
+    The slow-query event listeners are automatically registered on the
+    returned engine.
+    """
     if db_url is None:
         db_url = get_database_url()
-    return create_engine(db_url, **_engine_pool_kwargs())
+    eng = create_engine(db_url, **_engine_pool_kwargs())
+    _register_slow_query_listener(eng)
+    return eng
 
 
 def init_db() -> None:
-    """Initialize the module-level database connection and session factory."""
+    """Initialize the module-level database connection and session factory.
+
+    The slow-query event listeners are automatically registered on the
+    engine produced here.
+    """
     global engine, SessionLocal
     if engine is not None:
         return
 
     engine = create_engine(get_database_url(), **_engine_pool_kwargs())
+    _register_slow_query_listener(engine)
     SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
 
+
 def get_database() -> Generator[Session, None, None]:
     """
     FastAPI dependency that yields a database session and ensures cleanup.

quantara/web_app/tests/test_db_pool.py

@@ -9,6 +9,13 @@
 The actual pool behaviour is exercised in the integration suite against a
 real PostgreSQL container. This file focuses narrowly on configuration by
 patching ``create_engine`` so no database is required.
+
+Note: ``_register_slow_query_listener`` is also patched in every test that
+mocks ``create_engine``.  The listener registration calls ``event.listen``
+on the engine object, which requires a real SQLAlchemy engine; using a
+MagicMock would raise ``InvalidRequestError``.  Since pool-configuration
+tests care only about the kwargs forwarded to ``create_engine``, we simply
+no-op the listener step here.
 """
 
 from unittest.mock import patch
@@ -41,7 +48,8 @@ def test_database_init_db_uses_env_pool_settings(monkeypatch):
 
     from web_app.db import database
 
-    with patch("web_app.db.database.create_engine") as create_engine:
+    with patch("web_app.db.database.create_engine") as create_engine, \
+         patch("web_app.db.database._register_slow_query_listener"):
         database.init_db()
         assert create_engine.call_count == 1
         kwargs = create_engine.call_args.kwargs
@@ -58,7 +66,8 @@ def test_database_init_db_falls_back_to_defaults(monkeypatch):
 
     from web_app.db import database
 
-    with patch("web_app.db.database.create_engine") as create_engine:
+    with patch("web_app.db.database.create_engine") as create_engine, \
+         patch("web_app.db.database._register_slow_query_listener"):
         database.init_db()
         kwargs = create_engine.call_args.kwargs
         assert kwargs["pool_size"] == DEFAULT_POOL_SIZE
@@ -85,7 +94,8 @@ def test_db_connector_routes_through_init_engine(monkeypatch):
 
     from web_app.db.crud.base import DBConnector
 
-    with patch("web_app.db.database.create_engine") as create_engine:
+    with patch("web_app.db.database.create_engine") as create_engine, \
+         patch("web_app.db.database._register_slow_query_listener"):
         DBConnector(db_url="postgresql://user:pwd@host:5432/dbname")
         assert create_engine.call_count == 1
         kwargs = create_engine.call_args.kwargs
@@ -102,7 +112,8 @@ def test_db_connector_falls_back_to_defaults(monkeypatch):
 
     from web_app.db.crud.base import DBConnector
 
-    with patch("web_app.db.database.create_engine") as create_engine:
+    with patch("web_app.db.database.create_engine") as create_engine, \
+         patch("web_app.db.database._register_slow_query_listener"):
         DBConnector(db_url="postgresql://user:pwd@host:5432/dbname")
         kwargs = create_engine.call_args.kwargs
         assert kwargs["pool_size"] == DEFAULT_POOL_SIZE