diff --git a/backend/config.py b/backend/config.py index 2fb00ae..d46aa20 100644 --- a/backend/config.py +++ b/backend/config.py @@ -78,7 +78,24 @@ REQUEST_LOG_ENABLED = _bool_env("REQUEST_LOG_ENABLED", True) # reported through GET /api/request-logs/stats) — the request path never blocks # on the database. REQUEST_LOG_QUEUE_MAX = _int_env("REQUEST_LOG_QUEUE_MAX", 2000, 100, 100000) +# HARD memory ceiling for the same queue, per worker. The row count above does +# NOT bound memory on its own, because how much a row weighs is an operator +# setting: `requestlog.max_body_bytes` is editable from Settings and its stated +# ceiling is 256 KB, which a row can carry twice (request + response). Measured +# on the real dataclass, the two limits multiply out to: +# +# defaults (2 000 rows x 8 KB) 33.9 MiB 3.3% of the 1 GiB pod limit +# max_body_bytes at its 256 KB ceiling 1003 MiB at the pod limit +# REQUEST_LOG_QUEUE_MAX at its ceiling 1695 MiB over the pod limit +# +# Both of those are reachable from documented, in-range values, and the drop +# warning used to advise raising the queue - so following the tool's own advice +# could OOM the worker. Whichever limit is hit FIRST now stops the queue, so +# memory stays bounded no matter what the other is set to. +REQUEST_LOG_QUEUE_MAX_BYTES = _int_env( + "REQUEST_LOG_QUEUE_MAX_BYTES", 64 * 1024 * 1024, 1024 * 1024, 1024 * 1024 * 1024 +) # Rows per batched INSERT: one pool acquire per batch, not per request. REQUEST_LOG_BATCH_SIZE = _int_env("REQUEST_LOG_BATCH_SIZE", 100, 1, 1000) # Max wait before a partial batch is flushed (milliseconds). -REQUEST_LOG_FLUSH_MS = _int_env("REQUEST_LOG_FLUSH_MS", 500, 50, 10000) \ No newline at end of file +REQUEST_LOG_FLUSH_MS = _int_env("REQUEST_LOG_FLUSH_MS", 500, 50, 10000) diff --git a/backend/routers/request_logs.py b/backend/routers/request_logs.py index 6f23c6c..8b527b4 100644 --- a/backend/routers/request_logs.py +++ b/backend/routers/request_logs.py @@ -292,7 +292,13 @@ async def get_request_log_stats( "total_rows": (totals or {}).get("total_rows", 0), "oldest_at": totals["oldest_at"].isoformat() if totals and totals["oldest_at"] else None, "newest_at": totals["newest_at"].isoformat() if totals and totals["newest_at"] else None, - "sink": request_log_sink.stats, + # THIS WORKER only. The sink is a module global, so with + # UVICORN_WORKERS > 1 each process keeps its own queue and its own + # counters, and whichever worker happens to serve this request is + # the one being reported. Labelled rather than aggregated: there is + # no cross-process channel here, and a number that looks fleet-wide + # but is not would understate drops by exactly the worker count. + "sink": {**request_log_sink.stats, "scope": "this worker only"}, "retention": { "success_retention_days": get_config().success_retention_days, "error_retention_days": get_config().error_retention_days, diff --git a/backend/utils/request_log_sink.py b/backend/utils/request_log_sink.py index 64e8bc1..52b7f8e 100644 --- a/backend/utils/request_log_sink.py +++ b/backend/utils/request_log_sink.py @@ -25,7 +25,12 @@ 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 +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 @@ -113,6 +118,22 @@ class RequestLogRow: 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 @@ -165,8 +186,11 @@ class RequestLogRow: class RequestLogSink: """Bounded queue + single batching writer task (one per uvicorn worker).""" - def __init__(self, maxsize: int, batch_size: int, flush_ms: int): + 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 @@ -189,6 +213,8 @@ class RequestLogSink: 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, @@ -240,14 +266,27 @@ class RequestLogSink: 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"(capacity {self._maxsize}; raise REQUEST_LOG_QUEUE_MAX or lower " - f"requestlog.sample_rate)" + 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. @@ -259,6 +298,7 @@ class RequestLogSink: """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 @@ -267,7 +307,9 @@ class RequestLogSink: if remaining <= 0: break try: - batch.append(await asyncio.wait_for(queue.get(), timeout=remaining)) + 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 @@ -333,7 +375,9 @@ class RequestLogSink: while not queue.empty() and loop.time() < deadline: batch: List[RequestLogRow] = [] while not queue.empty() and len(batch) < self._batch_size: - batch.append(queue.get_nowait()) + 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 diff --git a/frontend/src/components/Settings.js b/frontend/src/components/Settings.js index e5e6b1b..5750a26 100644 --- a/frontend/src/components/Settings.js +++ b/frontend/src/components/Settings.js @@ -104,13 +104,29 @@ const Settings = () => { const onRequestLogSave = async (values) => { setRlSaving(true); try { - await axios.put('/api/request-logs/settings', { + const res = await axios.put('/api/request-logs/settings', { ...values, // Values come back from the InputNumber controls as numbers already; // the endpoint is properly typed, so no per-value JSON.stringify here // (unlike the ACME form above, which talks to the legacy endpoint). exclude_paths: values.exclude_paths || [], }); + // Show what the server ACTUALLY applied, not what was typed. Values are + // clamped server-side, and clearing the exclude list does not mean "log + // everything": normalize_exclude_paths() falls back to the shipped + // defaults so the log viewer and the raw-body heartbeat stay excluded. + // Without this the form would keep displaying an empty list that is not + // in effect. + const applied = res?.data?.settings; + if (applied) { + rlForm.setFieldsValue(applied); + const typed = values.exclude_paths || []; + if (typed.length === 0 && (applied.exclude_paths || []).length > 0) { + message.warning( + 'An empty exclude list is not applied as "log everything" — the shipped defaults were restored.' + ); + } + } message.success('Request log settings saved'); } catch (err) { message.error(err?.response?.data?.detail || 'Failed to save request log settings');