haproxy-openmanager/backend/utils/request_log_settings.py

"""v1.11.0 — operator-tunable settings for the request/response log.

The middleware runs on EVERY request, so the hot path must not touch the
database. `get_config()` returns a module-global immutable snapshot with no
`await`; `refresh_config()` reloads it from `system_settings` and is called

  * once at startup, right after migrations,
  * every `_TTL_SECONDS` from the sink's writer loop (off the request path),
  * synchronously at the end of `PUT /api/request-logs/settings`, so an
    operator's change takes effect immediately instead of up to 30s later.

asyncpg has no JSONB codec registered on this pool (see
database/connection.py), so every value comes back as a raw JSON *string* and
needs the `isinstance(v, str)` + `json.loads` guard used elsewhere in this
codebase (services/acme_service.py, utils/activity_log.py).
"""
import json
import logging
import time
from dataclasses import dataclass, replace
from typing import Any, Dict, Optional, Tuple

from database.connection import get_database_connection, close_database_connection

logger = logging.getLogger("haproxy_openmanager.request_log")

SETTINGS_CATEGORY = "requestlog"

# Kept in sync with the seed in database/migrations.ensure_request_log_settings().
# backend/tests/test_request_log_settings.py asserts the two agree, so a change
# here without a change there fails the suite rather than drifting silently.
DEFAULT_EXCLUDE_PATHS = (
    "/api/request-logs",
    "/api/health",
    "/api/docs",
    "/api/redoc",
    "/api/openapi.json",
    "/.well-known/acme-challenge",
    "/api/agents/heartbeat",
    "/static",
    "/favicon.ico",
)


@dataclass(frozen=True)
class RequestLogConfig:
    enabled: bool = True
    capture_inbound: bool = True
    capture_outbound: bool = True
    capture_bodies: bool = True
    capture_get: bool = True
    # SUCCESSFUL agent polls only. Off by default because the row rate of this
    # table is otherwise a linear function of fleet size, not of operator
    # activity: each agent runs a 30s cycle that issues three logged calls
    # (config, pending-requests, upgrade-status; the heartbeat is already
    # excluded) plus two more every fifth cycle. Measured, that is ~9 800 rows
    # per day PER AGENT, so a 200-node fleet writes ~2M rows/day and reaches the
    # 500 000 max_rows cap in about six hours - at which point the shipped
    # "7 days of successes, 30 days of failures" is not 7 and 30, it is 0.25.
    # FAILED agent calls are always kept regardless of this flag: they are the
    # half an operator actually needs, and they are rare.
    capture_agent_success: bool = False
    max_body_bytes: int = 8192
    sample_rate: float = 1.0
    exclude_paths: Tuple[str, ...] = DEFAULT_EXCLUDE_PATHS
    success_retention_days: int = 7
    error_retention_days: int = 30
    max_rows: int = 500000
    prune_interval_minutes: int = 60

    def as_dict(self) -> Dict[str, Any]:
        return {
            "enabled": self.enabled,
            "capture_inbound": self.capture_inbound,
            "capture_outbound": self.capture_outbound,
            "capture_bodies": self.capture_bodies,
            "capture_get": self.capture_get,
            "capture_agent_success": self.capture_agent_success,
            "max_body_bytes": self.max_body_bytes,
            "sample_rate": self.sample_rate,
            "exclude_paths": list(self.exclude_paths),
            "success_retention_days": self.success_retention_days,
            "error_retention_days": self.error_retention_days,
            "max_rows": self.max_rows,
            "prune_interval_minutes": self.prune_interval_minutes,
        }


DEFAULT_CONFIG = RequestLogConfig()

_CACHE: RequestLogConfig = DEFAULT_CONFIG
_CACHE_AT: float = 0.0
_TTL_SECONDS: float = 30.0

# Bounds, mirrored by the Pydantic model in routers/request_logs.py. Kept here
# too because refresh_config() reads whatever is in the table, which may have
# been written by an older build or by hand.
_BOUNDS = {
    "max_body_bytes": (0, 262144),
    "success_retention_days": (1, 365),
    "error_retention_days": (1, 365),
    "max_rows": (1000, 50_000_000),
    "prune_interval_minutes": (5, 1440),
}

MAX_EXCLUDE_PATHS = 64
MAX_EXCLUDE_PATH_LENGTH = 200


def get_config() -> RequestLogConfig:
    """Hot-path read: no await, no DB, no lock. Returns the last snapshot."""
    return _CACHE


def set_config(config: RequestLogConfig) -> None:
    """Replace the snapshot directly. Used by the settings PUT handler (which
    already has the validated values) and by tests."""
    global _CACHE, _CACHE_AT
    _CACHE = config
    _CACHE_AT = time.monotonic()


def _clamp_int(raw: Any, field: str, fallback: int) -> int:
    try:
        value = int(raw)
    except (TypeError, ValueError):
        return fallback
    low, high = _BOUNDS[field]
    return max(low, min(high, value))


def _clamp_float(raw: Any, fallback: float, low: float, high: float) -> float:
    try:
        value = float(raw)
    except (TypeError, ValueError):
        return fallback
    return max(low, min(high, value))


def _as_bool(raw: Any, fallback: bool) -> bool:
    if isinstance(raw, bool):
        return raw
    if isinstance(raw, (int, float)):
        return bool(raw)
    if isinstance(raw, str):
        lowered = raw.strip().lower()
        if lowered in ("true", "1", "yes", "on"):
            return True
        if lowered in ("false", "0", "no", "off"):
            return False
    return fallback


def normalize_exclude_paths(raw: Any, fallback: Tuple[str, ...]) -> Tuple[str, ...]:
    """Coerce whatever is stored into a bounded tuple of path prefixes."""
    if not isinstance(raw, (list, tuple)):
        return fallback
    out = []
    for item in raw:
        if not isinstance(item, str):
            continue
        candidate = item.strip()
        if not candidate.startswith("/") or len(candidate) > MAX_EXCLUDE_PATH_LENGTH:
            continue
        out.append(candidate)
        if len(out) >= MAX_EXCLUDE_PATHS:
            break
    return tuple(out) if out else fallback


def config_from_mapping(values: Dict[str, Any], base: Optional[RequestLogConfig] = None) -> RequestLogConfig:
    """Build a config from a plain suffix→value mapping, clamping every field.

    Unknown keys are ignored and missing keys keep the value from `base`
    (default: the shipped defaults), so a partially-seeded table still yields a
    complete, usable config.
    """
    base = base or DEFAULT_CONFIG
    return replace(
        base,
        enabled=_as_bool(values.get("enabled", base.enabled), base.enabled),
        capture_inbound=_as_bool(values.get("capture_inbound", base.capture_inbound), base.capture_inbound),
        capture_outbound=_as_bool(values.get("capture_outbound", base.capture_outbound), base.capture_outbound),
        capture_bodies=_as_bool(values.get("capture_bodies", base.capture_bodies), base.capture_bodies),
        capture_get=_as_bool(values.get("capture_get", base.capture_get), base.capture_get),
        capture_agent_success=_as_bool(
            values.get("capture_agent_success", base.capture_agent_success),
            base.capture_agent_success,
        ),
        max_body_bytes=_clamp_int(values.get("max_body_bytes", base.max_body_bytes), "max_body_bytes", base.max_body_bytes),
        sample_rate=_clamp_float(values.get("sample_rate", base.sample_rate), base.sample_rate, 0.0, 1.0),
        exclude_paths=normalize_exclude_paths(values.get("exclude_paths"), base.exclude_paths),
        success_retention_days=_clamp_int(
            values.get("success_retention_days", base.success_retention_days),
            "success_retention_days", base.success_retention_days,
        ),
        error_retention_days=_clamp_int(
            values.get("error_retention_days", base.error_retention_days),
            "error_retention_days", base.error_retention_days,
        ),
        max_rows=_clamp_int(values.get("max_rows", base.max_rows), "max_rows", base.max_rows),
        prune_interval_minutes=_clamp_int(
            values.get("prune_interval_minutes", base.prune_interval_minutes),
            "prune_interval_minutes", base.prune_interval_minutes,
        ),
    )


def _decode_setting_value(raw: Any) -> Any:
    """JSONB comes back as a raw string on this pool — parse it, but keep the
    original text if it is not valid JSON (an operator may have hand-written
    `7` or `seven`)."""
    if isinstance(raw, str):
        try:
            return json.loads(raw)
        except (json.JSONDecodeError, ValueError):
            return raw
    return raw


async def load_settings_rows(conn) -> Dict[str, Any]:
    """Read the `requestlog.*` rows into a suffix→value mapping."""
    rows = await conn.fetch(
        "SELECT key, value FROM system_settings WHERE category = $1",
        SETTINGS_CATEGORY,
    )
    values: Dict[str, Any] = {}
    for row in rows:
        key = row["key"]
        suffix = key.split(".", 1)[1] if "." in key else key
        values[suffix] = _decode_setting_value(row["value"])
    return values


async def refresh_config(force: bool = True) -> RequestLogConfig:
    """Reload the snapshot from the database.

    Never raises and never leaves a half-built config behind: on any failure
    the previous snapshot is kept, so a transient DB blip cannot silently turn
    logging off (or on).
    """
    global _CACHE_AT
    if not force and (time.monotonic() - _CACHE_AT) < _TTL_SECONDS:
        return _CACHE

    conn = None
    try:
        conn = await get_database_connection()
        values = await load_settings_rows(conn)
        if values:
            set_config(config_from_mapping(values))
        else:
            # Table not seeded yet (fresh install mid-migration) — keep the
            # in-code defaults but stamp the timestamp so we don't re-query
            # every tick.
            _CACHE_AT = time.monotonic()
        return _CACHE
    except Exception as exc:
        logger.debug(f"refresh_config: keeping previous snapshot ({exc})")
        _CACHE_AT = time.monotonic()
        return _CACHE
    finally:
        if conn is not None:
            try:
                await close_database_connection(conn)
            except Exception:
                pass


async def maybe_refresh_config() -> RequestLogConfig:
    """TTL-gated refresh, called from the sink's writer loop."""
    return await refresh_config(force=False)