peter
45c0ee7d58
What a job last did lived in the API process's memory and was gone at every restart; the audit log kept a run only when it was manual, failed or changed something. Now each run is a job_runs row written at its start and its end in its own session: who started it, its outcome as data, what it spent of the YouTube quota (quota_events.job_run_id) and the lines it logged (log_events.run_id, filterable in the log console). - a run a stopped process left open is marked interrupted at startup - "run now" is a one-shot job on the scheduler, which now always starts (interval jobs only with SCHEDULER_ENABLED); the manual actor is reset after the run, since pool threads are reused - the dashboard's last run comes from job_runs, so it survives a deploy - runs leave the audit log: migration 0109 moves the old job_run rows over (and back on downgrade); the audit page's run folding goes - job_run_retention_days (default 30) prunes old runs in audit_gc
369 lines
15 KiB
Python
369 lines
15 KiB
Python
"""The `log_events` sink: the stored copy of the log the diagnostics console reads.
|
||
|
||
stdout stays the complete log (docker keeps 3×50 MB of it); this table keeps the part worth querying
|
||
for `log_retention_days`:
|
||
|
||
* every WARNING and above,
|
||
* the `http.request` line of every request at INFO or above (not the DEBUG-level health checks and
|
||
static files),
|
||
* every line of a request whose user raised their diagnostics level (`logs.LevelGate`
|
||
marks them), which is the detail they consented to have kept,
|
||
* nothing a caller marked `extra={logs.PERSIST: False}` — how the request log caps anonymous
|
||
traffic per client.
|
||
|
||
What the table must never do is slow down or break the code that logs. So a line is redacted and
|
||
turned into a row on the CALLER's thread (the same `logs.record_fields` the stdout line is built
|
||
from), put on a bounded queue without blocking, and written in batches by one background thread
|
||
on its own connection — never inside a request's transaction. A full queue drops the row and
|
||
counts it; a failed write drops the batch and says so on stderr. Both are reported rather than
|
||
retried: a log that retries into a sick database is how a small outage becomes a big one.
|
||
|
||
Flood control is the second half. On 2026-09-16 a YouTube RSS outage made prod log 11k near-identical
|
||
warnings in four hours. stdout keeps them all; the table keeps `REPEATS_PER_MINUTE` of each message
|
||
template per minute and one `log.repeats_suppressed` row saying how many more there were.
|
||
|
||
Core layer: imports `logs`, `db`, `models`, `config` — nothing above them.
|
||
"""
|
||
|
||
import atexit
|
||
import json
|
||
import logging
|
||
import queue
|
||
import re
|
||
import sys
|
||
import threading
|
||
import time
|
||
from datetime import datetime, timedelta, timezone
|
||
from typing import Any
|
||
|
||
from sqlalchemy import delete, insert
|
||
from sqlalchemy.engine import Engine
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app import logs
|
||
from app.config import settings
|
||
from app.db import engine
|
||
from app.models import LogEvent
|
||
from app.utils import iso
|
||
|
||
QUEUE_MAX = 10_000
|
||
BATCH_MAX = 500
|
||
FLUSH_SECONDS = 1.0
|
||
REPEATS_PER_MINUTE = 30
|
||
REPEAT_WINDOW_SECONDS = 60.0
|
||
# Distinct templates tracked per window. Past this many, a new template is stored without counting
|
||
# — the limiter must not become the unbounded thing it exists to prevent.
|
||
_REPEAT_KEYS_MAX = 2_000
|
||
|
||
# Column widths from models.LogEvent. A value past its column would fail the WHOLE batch it rides in.
|
||
_LIMITS = {
|
||
"level": 10,
|
||
"source": 16,
|
||
"logger": 128,
|
||
"event": 64,
|
||
"request_id": 64,
|
||
"debug_session_id": 64,
|
||
"version": 64,
|
||
"msg": 8_000,
|
||
"exc": 32_000,
|
||
}
|
||
|
||
_log = logs.get_logger(__name__)
|
||
|
||
|
||
def should_store(record: logging.LogRecord) -> bool:
|
||
"""Whether this record belongs in `log_events` (see the module docstring)."""
|
||
if getattr(record, logs.PERSIST, True) is False:
|
||
return False
|
||
if record.levelno >= logging.WARNING or getattr(record, "elevated", False):
|
||
return True
|
||
return getattr(record, "event", None) == "http.request" and record.levelno >= logging.INFO
|
||
|
||
|
||
def _without_nul(value: Any) -> Any:
|
||
"""`value` with every NUL character replaced, in keys and nested values alike. Postgres refuses
|
||
NUL in text and `\\u0000` in JSONB, and a refused row fails the WHOLE batch it rides in — every
|
||
user's rows with it. Anything logged can carry one (`GET /assets/%00` logs the decoded path)."""
|
||
if isinstance(value, str):
|
||
return value.replace("\x00", "�")
|
||
if isinstance(value, dict):
|
||
return {_without_nul(k): _without_nul(v) for k, v in value.items()}
|
||
if isinstance(value, list):
|
||
return [_without_nul(v) for v in value]
|
||
return value
|
||
|
||
|
||
def to_row(fields: dict[str, Any]) -> dict[str, Any]:
|
||
"""`logs.record_fields` output → a `log_events` row: known columns only, text clipped to its
|
||
column and free of NUL, attrs made JSON-safe (anything exotic becomes its `str`, as on stdout)."""
|
||
row = {
|
||
column: fields.get(column) for column in LogEvent.__table__.columns.keys() if column != "id"
|
||
}
|
||
for column, limit in _LIMITS.items():
|
||
value = row.get(column)
|
||
if isinstance(value, str):
|
||
value = _without_nul(value)
|
||
row[column] = value[: limit - 1] + "…" if len(value) > limit else value
|
||
if row["attrs"] is not None:
|
||
row["attrs"] = _without_nul(
|
||
json.loads(json.dumps(row["attrs"], ensure_ascii=False, default=str))
|
||
)
|
||
return row
|
||
|
||
|
||
def without_erased(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
"""The rows, with the account taken off any whose account was erased meanwhile
|
||
(`logs.forget_user`). Checked right before the INSERT, under `logs.erasure_guard()` held until
|
||
the commit: a row built before the erasure can still be waiting for its batch."""
|
||
for row in rows:
|
||
if logs.is_erased(row.get("user_id")):
|
||
row["user_id"] = None
|
||
return rows
|
||
|
||
|
||
def _notice(code: str, msg: str, **attrs: Any) -> dict[str, Any]:
|
||
"""A row the sink writes about ITSELF. Built directly, not logged: logging it would route it back
|
||
through this sink (and the writer thread's own records are deliberately ignored)."""
|
||
return to_row(
|
||
{
|
||
"ts": datetime.now(timezone.utc),
|
||
"level": "WARNING",
|
||
"sev": logs.SEVERITY["WARNING"],
|
||
"source": logs.current_source(),
|
||
"logger": _log.name,
|
||
"event": code,
|
||
"msg": msg,
|
||
"attrs": attrs,
|
||
"version": settings.app_version,
|
||
}
|
||
)
|
||
|
||
|
||
_DIGITS = re.compile(r"\d+")
|
||
|
||
|
||
def _template(record: logging.LogRecord) -> str:
|
||
"""The flood limiter's grouping key: the message template. Except when the template is a bare
|
||
`%s` — a pass-through for someone else's text (yt-dlp's lines, `formats._YdlLog`), where the
|
||
template says nothing and every distinct line would share one bucket. Then the text itself
|
||
groups, with its numbers folded, so `Retrying fragment 5 (1/10)` and `(2/10)` are still one
|
||
repeat and a different warning is not crowded out by them."""
|
||
msg = str(record.msg)
|
||
if msg != "%s" or not record.args:
|
||
return msg
|
||
return _DIGITS.sub("#", record.getMessage())[:200]
|
||
|
||
|
||
class _Repeats:
|
||
"""Per-template admission for one minute at a time. Used by the writer thread only (unlocked)."""
|
||
|
||
def __init__(self, per_window: int, window_seconds: float):
|
||
self.per_window = per_window
|
||
self.window = window_seconds
|
||
self.started = time.monotonic()
|
||
self.seen: dict[tuple[str, str, int], int] = {}
|
||
|
||
def admit(self, key: tuple[str, str, int] | None) -> bool:
|
||
if key is None:
|
||
return True
|
||
count = self.seen.get(key)
|
||
if count is None and len(self.seen) >= _REPEAT_KEYS_MAX:
|
||
return True
|
||
self.seen[key] = (count or 0) + 1
|
||
return self.seen[key] <= self.per_window
|
||
|
||
def roll(self, now: float, *, final: bool = False) -> list[dict[str, Any]]:
|
||
"""Close the window if it is over — or, `final`, because the process is: one notice per
|
||
template that went past the limit. Without `final` a process that lives under a minute
|
||
(every download child) would drop its repeats and never say so."""
|
||
if not final and now - self.started < self.window:
|
||
return []
|
||
notices = [
|
||
_notice(
|
||
"log.repeats_suppressed",
|
||
f"{count - self.per_window} more like this were not stored",
|
||
logger=logger,
|
||
template=logs.redact_text(template),
|
||
suppressed=count - self.per_window,
|
||
window_seconds=int(self.window),
|
||
)
|
||
for (logger, template, _level), count in self.seen.items()
|
||
if count > self.per_window
|
||
]
|
||
self.seen.clear()
|
||
self.started = now
|
||
return notices
|
||
|
||
|
||
class LogSink(logging.Handler):
|
||
"""The handler plus its writer. `start()` runs the writer thread; a test drives `write_pending()`
|
||
directly instead, on its own thread, with no timing involved."""
|
||
|
||
def __init__(self, bind: Engine = engine):
|
||
super().__init__(level=logging.NOTSET)
|
||
# Correlation fields are contextvars: capture them now, on the thread that logged.
|
||
self.addFilter(logs.ContextFilter())
|
||
self._bind = bind
|
||
self._queue: queue.Queue[tuple[tuple[str, str, int] | None, dict[str, Any]]] = queue.Queue(
|
||
QUEUE_MAX
|
||
)
|
||
self._dropped = 0
|
||
self._dropped_lock = threading.Lock()
|
||
self._repeats = _Repeats(REPEATS_PER_MINUTE, REPEAT_WINDOW_SECONDS)
|
||
self._stopping = threading.Event()
|
||
self._writer: threading.Thread | None = None
|
||
|
||
# --- caller side ---------------------------------------------------------------------------
|
||
def emit(self, record: logging.LogRecord) -> None:
|
||
if threading.current_thread() is self._writer:
|
||
return # the writer's own trouble (a driver warning mid-insert) must not feed itself
|
||
if not should_store(record):
|
||
return
|
||
try:
|
||
row = to_row(logs.record_fields(record))
|
||
# silent-ok: `handleError` is the report — logging's own out-of-band channel to stderr.
|
||
# A log call here would be handed to this handler, which is the one that just failed.
|
||
except Exception: # noqa: BLE001 — a record we cannot turn into a row is reported, not raised
|
||
self.handleError(record)
|
||
return
|
||
# Request lines are one per request by nature; only the repeated SAME message is a flood.
|
||
key = (
|
||
None
|
||
if row["event"] == "http.request"
|
||
else (record.name, _template(record), record.levelno)
|
||
)
|
||
try:
|
||
self._queue.put_nowait((key, row))
|
||
# silent-ok: counted, not logged — the count goes out as the `log.sink.dropped` notice row
|
||
# the writer emits, which is the one report that does not feed this queue again.
|
||
except queue.Full:
|
||
with self._dropped_lock:
|
||
self._dropped += 1
|
||
|
||
# --- writer side ---------------------------------------------------------------------------
|
||
def write_pending(self, wait: float = 0.0, *, final: bool = False) -> int:
|
||
"""Write one batch: whatever is queued (up to BATCH_MAX), waiting up to `wait` seconds for the
|
||
first item. Returns the number of rows written, notices included. `final` closes the
|
||
repeat window early — the last batch before the process exits."""
|
||
rows: list[dict[str, Any]] = []
|
||
try:
|
||
item = self._queue.get(timeout=wait) if wait > 0 else self._queue.get_nowait()
|
||
# silent-ok: an idle sink. Nothing failed; there was simply nothing waiting.
|
||
except queue.Empty:
|
||
item = None
|
||
while item is not None:
|
||
key, row = item
|
||
if self._repeats.admit(key):
|
||
rows.append(row)
|
||
if len(rows) >= BATCH_MAX:
|
||
break
|
||
try:
|
||
item = self._queue.get_nowait()
|
||
# silent-ok: the batch ends here because the queue ran dry, which is the normal end.
|
||
except queue.Empty:
|
||
item = None
|
||
rows.extend(self._repeats.roll(time.monotonic(), final=final))
|
||
with self._dropped_lock:
|
||
dropped, self._dropped = self._dropped, 0
|
||
if dropped:
|
||
rows.append(
|
||
_notice("log.sink.dropped", f"{dropped} log events were dropped", dropped=dropped)
|
||
)
|
||
if not rows:
|
||
return 0
|
||
try:
|
||
# The connection is checked out BEFORE the guard, and every request path takes them in
|
||
# that order too (its session already holds a connection when it reaches the guard). The
|
||
# other way round, this thread would hold a process-global lock while waiting up to
|
||
# `pool_timeout` for a connection, and the requests that could return one are the ones
|
||
# blocked on the lock. The guard still spans the check, the insert and the commit.
|
||
with self._bind.connect() as conn, logs.erasure_guard(), conn.begin():
|
||
conn.execute(insert(LogEvent), without_erased(rows))
|
||
except Exception as exc: # noqa: BLE001 — the database is down or refused the batch
|
||
# silent-ok / noqa T201: the sink's OWN failure. A log call here would be handed to
|
||
# this sink, which is the thing that just failed — straight into a loop. stderr is
|
||
# the one way out that does not re-enter.
|
||
print( # noqa: T201
|
||
f"log sink: could not store {len(rows)} log events: "
|
||
f"{logs.redact_text(str(exc))[:300]}",
|
||
file=sys.stderr,
|
||
)
|
||
return 0
|
||
return len(rows)
|
||
|
||
def _run(self) -> None:
|
||
while not self._stopping.is_set():
|
||
self.write_pending(wait=FLUSH_SECONDS)
|
||
while self.write_pending(): # drain what arrived before the stop
|
||
pass
|
||
self.write_pending(final=True) # and the repeat notices of a window cut short
|
||
|
||
def start(self) -> None:
|
||
self._writer = threading.Thread(target=self._run, name="log-sink", daemon=True)
|
||
self._writer.start()
|
||
|
||
def stop(self, timeout: float = 5.0) -> None:
|
||
self._stopping.set()
|
||
if self._writer is not None:
|
||
self._writer.join(timeout)
|
||
|
||
|
||
_sink: LogSink | None = None
|
||
|
||
|
||
def start() -> None:
|
||
"""Attach the sink to the root logger and start its writer. Call after `logs.configure_logging`
|
||
(which replaces the root handlers) and once the schema exists. A no-op when `LOG_DB_SINK` is off
|
||
or it is already running."""
|
||
global _sink
|
||
if _sink is not None or not settings.log_db_sink:
|
||
return
|
||
_sink = LogSink()
|
||
# The same level gate as stdout: the root may be lowered for one user's elevated requests, and
|
||
# everyone else's detail must not reach the table through that gap either.
|
||
_sink.addFilter(logs.LevelGate())
|
||
logging.getLogger().addHandler(_sink)
|
||
_sink.start()
|
||
# A short-lived process (the download child) exits without a lifespan: flush on the way out.
|
||
atexit.register(stop)
|
||
|
||
|
||
def stop() -> None:
|
||
"""Detach the sink and write what is still queued."""
|
||
global _sink
|
||
if _sink is None:
|
||
return
|
||
logging.getLogger().removeHandler(_sink)
|
||
_sink.stop()
|
||
_sink = None
|
||
|
||
|
||
def gc(db: Session, retention_days: int) -> int:
|
||
"""Stage the deletion of rows older than `retention_days` (at least one day — stored logs always
|
||
expire). The caller commits. Returns the number of rows removed."""
|
||
cutoff = datetime.now(timezone.utc) - timedelta(days=max(1, retention_days))
|
||
result = db.execute(delete(LogEvent).where(LogEvent.ts < cutoff))
|
||
return result.rowcount or 0
|
||
|
||
|
||
def serialize(row: LogEvent) -> dict:
|
||
"""A stored row as the API sends it — one shape for every reader: the admin console and
|
||
a debug session's evidence list."""
|
||
return {
|
||
"id": row.id,
|
||
"ts": iso(row.ts),
|
||
"level": row.level,
|
||
"source": row.source,
|
||
"logger": row.logger,
|
||
"event": row.event,
|
||
"msg": row.msg,
|
||
"attrs": row.attrs,
|
||
"exc": row.exc,
|
||
"request_id": row.request_id,
|
||
"user_id": row.user_id,
|
||
"job_id": row.job_id,
|
||
"run_id": row.run_id,
|
||
"debug_session_id": row.debug_session_id,
|
||
"version": row.version,
|
||
}
|