gsadmin/haproxy-openmanager
1
0
Fork 0
mirror of https://github.com/taylanbakircioglu/haproxy-openmanager.git synced 2026-09-12 05:48:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(requestlog): bound queue memory, and stop the UI reporting things it cannot know

Browse Source 
Three hardening fixes with the same shape: a number that was true under the
defaults and untrue at the edges.

1. QUEUE MEMORY WAS AN OPERATOR SETTING, NOT A LIMIT.

The queue was bounded by ROW COUNT only, and how much a row weighs is
`requestlog.max_body_bytes` - editable from Settings, documented ceiling 256 KB,
and a row can hold that twice (request + response). Measured on the real
dataclass with distinct buffers per row:

    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 are reachable from in-range, documented values, and the drop warning
advised "raise REQUEST_LOG_QUEUE_MAX" - so following the tool's own advice on a
busy install could OOM the worker. REQUEST_LOG_QUEUE_MAX_BYTES (default 64 MiB)
now caps the queue in bytes as well as in rows, whichever binds first, released
as rows drain. Verified: with max_body_bytes at 256 KB the queue holds 7.5 MiB
against an 8 MiB budget where it would otherwise have held 1003 MiB, and it
accepts rows again as soon as the writer drains it. The warning text now names
the setting that actually helps.

2. SINK COUNTERS ARE PER WORKER AND DID NOT SAY SO.

The sink is a module global, so with UVICORN_WORKERS > 1 each process has its
own queue and its own counters, and `GET /api/request-logs/stats` reports
whichever worker happened to serve the request. The feature is sold on "a
saturated logger drops rows visibly"; at 4 workers the visible number was a
quarter of the truth. Labelled `"scope": "this worker only"` rather than
aggregated - there is no cross-process channel here, and a number that looks
fleet-wide but is not is worse than one that admits its scope.

3. AN EMPTY EXCLUDE LIST IS NOT APPLIED AS "LOG EVERYTHING".

normalize_exclude_paths() falls back to the shipped defaults when the list comes
out empty, which is the right call - it keeps the log viewer and the raw-body
heartbeat endpoint excluded - but the UI kept displaying the empty list the
operator typed, so the form showed a policy that was not in effect. The save
handler now re-applies whatever the server actually stored (which also surfaces
server-side clamping of every numeric field) and says plainly that the defaults
were restored.
This commit is contained in:
taylanbakircioglu
2026-08-15 11:04:31 +03:00
parent bec0613ae5
commit bd4a50943f
4 changed files with 92 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/config.py
+18 -1
View File
@@ -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)
REQUEST_LOG_FLUSH_MS = _int_env("REQUEST_LOG_FLUSH_MS", 500, 50, 10000)
backend/routers/request_logs.py
+7 -1
View File
@@ -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,
backend/utils/request_log_sink.py
+50 -6
View File
@@ -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
frontend/src/components/Settings.js
+17 -1
View File
@@ -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');