haproxy-openmanager/backend/utils/request_log_sink.py

"""v1.11.0 — batching writer for the unified request/response log.

One row per API call is the highest write volume in this system, and the
asyncpg pool (min=10/max=50, see database/connection.py) is shared with every
request handler and four background loops. Acquiring a connection per logged
request would exhaust it under any real load, so instead:

    hot path  ──offer(row)──▶  bounded asyncio.Queue  ──▶  single writer task
                                (drops when full)          (executemany batches)

The hot path never awaits I/O and never raises. When the queue is full rows are
counted as dropped and reported through `GET /api/request-logs/stats`, so a
saturated logger is visible rather than silent.

Redaction deliberately happens HERE, on the writer task, not in the middleware:
the recursive walk is the most expensive part of building a row and it has no
business running inside the request coroutine.
"""
import asyncio
import json
import logging
import random
from contextvars import ContextVar
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional

from config import (
    REQUEST_LOG_BATCH_SIZE,
    REQUEST_LOG_FLUSH_MS,
    REQUEST_LOG_QUEUE_MAX,
    REQUEST_LOG_QUEUE_MAX_BYTES,
)
from database.connection import get_database_connection, close_database_connection
from utils.request_log_redaction import decode_body, redact_headers
from utils.request_log_settings import get_config, maybe_refresh_config

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

# Set by the inbound middleware; read by outbound_span so an outbound call
# inherits the id of the inbound request that caused it. That is what turns
# "operator clicked Issue Certificate" and "we POSTed to Let's Encrypt" into
# one readable trace.
request_id_context: ContextVar[Optional[str]] = ContextVar(
    "request_log_request_id", default=None
)

# `target` on an INBOUND row means the same thing it means on an outbound one:
# who was on the other end. Set by the middleware when the caller authenticated
# with an agent API key rather than a user JWT, which is how agent traffic is
# told apart from operator traffic without a database lookup on the hot path.
# Deliberately the same literal as http_instrumentation.TARGET_AGENT, so
# `target = 'agent'` selects the whole conversation with the fleet in both
# directions; `direction` separates them when that matters.
TARGET_INBOUND_AGENT = "agent"

_INSERT_SQL = """
INSERT INTO request_logs (
    request_id, direction, target, method, url, path, query_params,
    status_code, status_class, duration_ms, user_id, username, client_ip,
    user_agent, request_headers, request_body, request_body_bytes,
    response_headers, response_body, response_body_bytes, error, truncated,
    created_at
) VALUES (
    $1, $2, $3, $4, $5, $6, $7::jsonb,
    $8, $9, $10, $11, $12, $13::inet,
    $14, $15::jsonb, $16::jsonb, $17,
    $18::jsonb, $19::jsonb, $20, $21, $22,
    $23
)
"""


def _jsonb(value: Any) -> Optional[str]:
    """asyncpg has no JSONB codec on this pool, so JSONB params travel as text
    and are cast in SQL — the house idiom (utils/activity_log.py)."""
    if value is None:
        return None
    try:
        return json.dumps(value, default=str)
    except Exception:
        return json.dumps({"_serialize_error": True})


@dataclass
class RequestLogRow:
    """One captured exchange, still holding RAW body bytes.

    Decoding and redaction run in `to_params()` on the writer task.
    """

    request_id: str
    direction: str
    method: str
    url: str
    target: Optional[str] = None
    path: Optional[str] = None
    query_string: Optional[str] = None
    query_params: Optional[Dict[str, Any]] = None
    status_code: Optional[int] = None
    duration_ms: int = 0
    user_id: Optional[int] = None
    username: Optional[str] = None
    client_ip: Optional[str] = None
    user_agent: Optional[str] = None
    request_headers: Optional[Dict[str, str]] = None
    response_headers: Optional[Dict[str, str]] = None
    request_body_raw: Optional[bytes] = None
    request_body_bytes: int = 0
    request_content_type: Optional[str] = None
    response_body_raw: Optional[bytes] = None
    response_body_bytes: int = 0
    response_content_type: Optional[str] = None
    # Pre-decoded body override, used by outbound spans that hold a dict/str
    # rather than wire bytes (e.g. the synthetic JWS summary).
    request_body_value: Optional[Any] = None
    response_body_value: Optional[Any] = None
    error: Optional[str] = None
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

    def queue_weight(self) -> int:
        """Approximate bytes this row holds while it waits in the queue.

        The buffered bodies are the only part that varies by orders of
        magnitude (0 to 2 x max_body_bytes); everything else is a handful of
        short strings and two small header dicts, measured at ~1.4 KB per row.
        Used to enforce a byte budget alongside the row count, so memory does
        not become a function of an operator-editable setting.
        """
        weight = 1400
        if self.request_body_raw:
            weight += len(self.request_body_raw)
        if self.response_body_raw:
            weight += len(self.response_body_raw)
        return weight

    @property
    def status_class(self) -> int:
        """`status_code // 100`, or 0 when there was no HTTP response at all
        (transport error / unhandled exception). 0 is what the error-retention
        prune treats as an error alongside >= 4."""
        if not self.status_code:
            return 0
        return int(self.status_code) // 100

    def to_params(self) -> List[Any]:
        req_body, req_truncated = (self.request_body_value, False)
        if req_body is None:
            req_body, req_truncated = decode_body(
                self.request_body_raw, self.request_content_type, self.request_body_bytes
            )

        res_body, res_truncated = (self.response_body_value, False)
        if res_body is None:
            res_body, res_truncated = decode_body(
                self.response_body_raw, self.response_content_type, self.response_body_bytes
            )

        return [
            self.request_id[:64],
            self.direction,
            self.target[:32] if self.target else None,
            (self.method or "")[:10],
            self.url or "",
            self.path[:512] if self.path else None,
            _jsonb(self.query_params),
            self.status_code,
            self.status_class,
            max(0, int(self.duration_ms)),
            self.user_id,
            self.username[:50] if self.username else None,
            self.client_ip,
            self.user_agent[:1024] if self.user_agent else None,
            _jsonb(redact_headers(self.request_headers)),
            _jsonb(req_body),
            max(0, int(self.request_body_bytes)),
            _jsonb(redact_headers(self.response_headers)),
            _jsonb(res_body),
            max(0, int(self.response_body_bytes)),
            self.error[:4000] if self.error else None,
            bool(req_truncated or res_truncated),
            self.created_at,
        ]


class RequestLogSink:
    """Bounded queue + single batching writer task (one per uvicorn worker)."""

    def __init__(self, maxsize: int, batch_size: int, flush_ms: int,
                 max_bytes: int = REQUEST_LOG_QUEUE_MAX_BYTES):
        self._maxsize = maxsize
        self._max_bytes = max_bytes
        self._queued_bytes = 0
        self._batch_size = batch_size
        self._flush_seconds = flush_ms / 1000.0
        self._queue: Optional[asyncio.Queue] = None
        self._dropped = 0
        self._written = 0
        self._failed = 0
        self._running = False

    # -- lifecycle ---------------------------------------------------------

    def _ensure_queue(self) -> asyncio.Queue:
        # Created lazily so importing this module never needs a running loop
        # (matters for the test suite, which imports main.py without one).
        if self._queue is None:
            self._queue = asyncio.Queue(maxsize=self._maxsize)
        return self._queue

    @property
    def stats(self) -> Dict[str, int]:
        return {
            "queued": self._queue.qsize() if self._queue is not None else 0,
            "queue_capacity": self._maxsize,
            "queued_bytes": self._queued_bytes,
            "queue_capacity_bytes": self._max_bytes,
            "written": self._written,
            "dropped": self._dropped,
            "failed_batches": self._failed,
            "running": 1 if self._running else 0,
        }

    # -- producer side (hot path) -----------------------------------------

    def offer(self, row: RequestLogRow) -> None:
        """Enqueue a row. NEVER blocks, NEVER raises.

        Sampling is applied here rather than in the middleware so both
        directions go through one policy: successful *inbound* traffic can be
        sampled down, errors never are.
        """
        try:
            cfg = get_config()
            if not cfg.enabled:
                return
            if row.direction == "inbound" and not cfg.capture_inbound:
                return
            if row.direction == "outbound" and not cfg.capture_outbound:
                return
            # Fleet-scale gate, and the reason this table's size is a function of
            # operator activity rather than of node count. An agent's 30s cycle
            # issues three logged calls, plus two more every fifth cycle: ~9 800
            # rows/day PER AGENT, all of them 200s saying "nothing changed". At a
            # few hundred nodes that is millions of rows a day, and the row cap is
            # then reached in hours, which silently shortens the configured
            # retention for EVERYTHING else in the table - including the failures
            # the log exists for. Successes are dropped, failures never are.
            if (
                row.direction == "inbound"
                and row.target == TARGET_INBOUND_AGENT
                and not cfg.capture_agent_success
                and row.status_class in (1, 2, 3)
            ):
                return
            if (
                row.direction == "inbound"
                and cfg.sample_rate < 1.0
                and row.status_class in (1, 2, 3)
                and random.random() > cfg.sample_rate
            ):
                return
            if not cfg.capture_bodies:
                row.request_body_raw = None
                row.response_body_raw = None
                row.request_body_value = None
                row.response_body_value = None

            # Byte budget, checked BEFORE the row count. The count alone does
            # not bound memory: how much a row weighs is an operator setting,
            # and `max_body_bytes` at its documented 256 KB ceiling puts the
            # default 2 000-row queue at ~1 GiB, which is the whole pod limit.
            # Dropping here is the same visible, counted drop as a full queue.
            weight = row.queue_weight()
            if self._queued_bytes + weight > self._max_bytes:
                raise asyncio.QueueFull

            self._ensure_queue().put_nowait(row)
            self._queued_bytes += weight
        except asyncio.QueueFull:
            self._dropped += 1
            if self._dropped % 500 == 1:
                logger.warning(
                    f"request_log: queue full, {self._dropped} row(s) dropped so far "
                    f"({self._queue.qsize() if self._queue is not None else 0}/{self._maxsize} rows, "
                    f"{self._queued_bytes // 1024} KiB/{self._max_bytes // 1024} KiB). "
                    f"Lower requestlog.max_body_bytes or requestlog.sample_rate. "
                    f"Raising REQUEST_LOG_QUEUE_MAX also raises the memory this "
                    f"worker can hold, so raise REQUEST_LOG_QUEUE_MAX_BYTES with it."
                )
        except Exception as exc:
            # Instrumentation must never break the thing it instruments.
            logger.debug(f"request_log: offer() failed: {exc}")

    # -- consumer side (writer task) --------------------------------------

    async def _collect(self) -> List[RequestLogRow]:
        """Wait for at least one row, then drain up to batch_size or flush_ms."""
        queue = self._ensure_queue()
        first = await queue.get()
        self._queued_bytes = max(0, self._queued_bytes - first.queue_weight())
        batch = [first]
        loop = asyncio.get_running_loop()
        deadline = loop.time() + self._flush_seconds
        while len(batch) < self._batch_size:
            remaining = deadline - loop.time()
            if remaining <= 0:
                break
            try:
                nxt = await asyncio.wait_for(queue.get(), timeout=remaining)
                self._queued_bytes = max(0, self._queued_bytes - nxt.queue_weight())
                batch.append(nxt)
            except asyncio.TimeoutError:
                break
        return batch

    async def _write(self, batch: List[RequestLogRow]) -> None:
        if not batch:
            return
        params = []
        for row in batch:
            try:
                params.append(row.to_params())
            except Exception as exc:
                logger.debug(f"request_log: row serialization failed, skipped: {exc}")
        if not params:
            return

        conn = None
        try:
            conn = await get_database_connection()
            await conn.executemany(_INSERT_SQL, params)
            self._written += len(params)
        except Exception as exc:
            self._failed += 1
            # A missing table (pre-migration) or a transient pool error must not
            # take the writer loop down — drop the batch and carry on.
            logger.warning(f"request_log: batch write failed ({len(params)} rows): {exc}")
        finally:
            if conn is not None:
                try:
                    await close_database_connection(conn)
                except Exception:
                    pass

    async def run(self) -> None:
        """Writer loop. Started once per worker from startup_event()."""
        self._running = True
        logger.info(
            f"request_log sink started (queue={self._maxsize}, batch={self._batch_size}, "
            f"flush={int(self._flush_seconds * 1000)}ms)"
        )
        try:
            while True:
                try:
                    batch = await self._collect()
                    await self._write(batch)
                    await maybe_refresh_config()
                except asyncio.CancelledError:
                    raise
                except Exception as exc:  # pragma: no cover - defensive
                    logger.error(f"request_log sink loop error: {exc}")
                    await asyncio.sleep(1)
        finally:
            self._running = False

    async def flush(self, timeout: float = 3.0) -> int:
        """Drain and persist whatever is queued. Called on shutdown."""
        queue = self._queue
        if queue is None or queue.empty():
            return 0
        written = 0
        loop = asyncio.get_running_loop()
        deadline = loop.time() + timeout
        while not queue.empty() and loop.time() < deadline:
            batch: List[RequestLogRow] = []
            while not queue.empty() and len(batch) < self._batch_size:
                row = queue.get_nowait()
                self._queued_bytes = max(0, self._queued_bytes - row.queue_weight())
                batch.append(row)
            await self._write(batch)
            written += len(batch)
        return written


request_log_sink = RequestLogSink(
    REQUEST_LOG_QUEUE_MAX, REQUEST_LOG_BATCH_SIZE, REQUEST_LOG_FLUSH_MS
)