From 5e208353e31388aced7ffd0a2b036aa61d122edb Mon Sep 17 00:00:00 2001 From: npeter83 Date: Sat, 19 Sep 2026 01:40:46 +0200 Subject: [PATCH] feat(diagnostics): admin query API over the stored log GET /api/admin/logs pages `log_events` newest first on a (ts, id) keyset: a browser row's ts is clamped into its consent window, so ts alone is not unique. Filters AND together: level, source, account (or `none`), event code (or `none`), request id, job id, a free-text search over the message, code, request id, stack and attributes (typed text is a literal), and a time window. Each page names the accounts it mentions. GET /api/admin/logs/facets counts what the filter chips offer over the window alone, so a chip row does not shrink as filters are picked. Admin-only and read-only; reading is not audited (the console's live tail polls). There is deliberately no "ids after N" tail: several processes write this table and a batch takes its ids at insert, not at commit, so a tail keyed on ids could pass a row that commits late. --- backend/app/main.py | 2 + backend/app/routes/admin_logs.py | 199 +++++++++++++++++++++++++++++++ backend/tests/test_admin_logs.py | 154 ++++++++++++++++++++++++ 3 files changed, 355 insertions(+) create mode 100644 backend/app/routes/admin_logs.py create mode 100644 backend/tests/test_admin_logs.py diff --git a/backend/app/main.py b/backend/app/main.py index 5896a885..bf0e4708 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -32,6 +32,7 @@ from app.plex import stream as plex_stream from app.plex import subtitles as plex_subtitles from app.routes import ( admin, + admin_logs as admin_logs_routes, audit as audit_routes, auth as auth_routes, channels, @@ -285,6 +286,7 @@ app.include_router(admin.router) app.include_router(config_routes.router) app.include_router(scheduler_routes.router) app.include_router(audit_routes.router) +app.include_router(admin_logs_routes.router) app.include_router(quota.router) app.include_router(version.router) app.include_router(setup_routes.router) diff --git a/backend/app/routes/admin_logs.py b/backend/app/routes/admin_logs.py new file mode 100644 index 00000000..4f351b4b --- /dev/null +++ b/backend/app/routes/admin_logs.py @@ -0,0 +1,199 @@ +"""The admin diagnostics console's read side (E22.6): query `log_events`, the server's stored log. + +Admin-only and read-only — rows leave the table through the retention job (`log_retention_days`), +never through here. Reading is not audited: it changes nothing, and the console's live tail polls, +so a row per read would bury the audit log's real entries. + +What a row holds was redacted when it was written (`logs.record_fields`, and the ingest route for a +browser's rows), so this serves the columns as they are; a value the redactor took out reads as +`[redacted]` / `[email]`, which is the point — the console shows that something was there. + +Paging is a keyset over `(ts desc, id desc)`: a browser row's `ts` is clamped into its consent +window, so `ts` alone is not unique, and an offset would skip or repeat rows while the writer adds +new ones at the top. The live tail re-reads the first page rather than asking for "ids after N": +several processes write this table, and a batch's ids are taken when it inserts, not when it +commits, so a slower batch can land below an id the tail already passed — and would never show. +""" + +import base64 +import binascii +import json +from datetime import datetime, timezone +from typing import Literal + +from fastapi import APIRouter, Depends, HTTPException, Query +from sqlalchemy import Text, and_, cast, func, or_, select +from sqlalchemy.orm import Session + +from app.auth import admin_user +from app.db import get_db +from app.models import LogEvent, User +from app.utils import iso, like_escape + +router = APIRouter(prefix="/api/admin/logs", tags=["admin"]) + +PAGE_MAX = 200 +# A facet list is a chip row, not a report: past this many distinct events the rest are reachable +# through the free-text search, and the response stays small whatever a self-host logs. +FACET_EVENTS_MAX = 100 + +Level = Literal["TRACE", "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] +# `none` selects the rows that have no account / no event code (a scheduler line, an anonymous +# request) — a real value to filter on, not the absence of a filter. +NONE = "none" + + +def _utc(value: datetime | None) -> datetime | None: + """A bound without a zone is UTC — the column is `timestamptz`, and a naive value would be read + in the database session's zone instead.""" + if value is None or value.tzinfo is not None: + return value + return value.replace(tzinfo=timezone.utc) + + +def _encode_cursor(row: LogEvent) -> str: + return base64.urlsafe_b64encode(json.dumps([row.ts.isoformat(), row.id]).encode()).decode() + + +def _decode_cursor(cursor: str) -> tuple[datetime, int]: + try: + ts, row_id = json.loads(base64.urlsafe_b64decode(cursor.encode())) + return _utc(datetime.fromisoformat(ts)), int(row_id) + except (binascii.Error, json.JSONDecodeError, ValueError, TypeError) as exc: + raise HTTPException(status_code=400, detail={"code": "bad_cursor"}) from exc + + +def _serialize(row: LogEvent) -> dict: + 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, + "debug_session_id": row.debug_session_id, + "version": row.version, + } + + +def _accounts(db: Session, user_ids: set[int]) -> dict[str, str]: + """The accounts a page names, by id → address. An id with no account left (erased after the + line was written) is simply absent, and the client shows the bare id.""" + if not user_ids: + return {} + rows = db.execute(select(User.id, User.email).where(User.id.in_(user_ids))).all() + return {str(uid): email for uid, email in rows} + + +@router.get("") +def list_logs( + levels: list[Level] = Query(default=[]), + sources: list[str] = Query(default=[], max_length=8), + user: str | None = Query(default=None, pattern=r"^(none|[0-9]{1,10})$"), + event: str | None = Query( + default=None, max_length=64, pattern=r"^(none|[a-z][a-z0-9_]*(\.[a-z0-9_]+)*)$" + ), + request_id: str | None = Query(default=None, pattern=r"^[A-Za-z0-9._-]{8,64}$"), + job_id: int | None = Query(default=None, ge=1), + q: str | None = Query(default=None, max_length=200), + since: datetime | None = None, + until: datetime | None = None, + before: str | None = Query(default=None, max_length=200), + limit: int = Query(default=100, ge=1, le=PAGE_MAX), + _: User = Depends(admin_user), + db: Session = Depends(get_db), +) -> dict: + """One page of rows, newest first, matching every filter given (they AND together); `before` + continues from a previous page's `next_cursor`.""" + query = select(LogEvent) + if levels: + query = query.where(LogEvent.level.in_(levels)) + if sources: + query = query.where(LogEvent.source.in_(sources)) + if user is not None: + query = query.where( + LogEvent.user_id.is_(None) if user == NONE else LogEvent.user_id == int(user) + ) + if event is not None: + query = query.where(LogEvent.event.is_(None) if event == NONE else LogEvent.event == event) + if request_id is not None: + query = query.where(LogEvent.request_id == request_id) + if job_id is not None: + query = query.where(LogEvent.job_id == job_id) + if since is not None: + query = query.where(LogEvent.ts >= _utc(since)) + if until is not None: + query = query.where(LogEvent.ts < _utc(until)) + if q and q.strip(): + like = f"%{like_escape(q.strip())}%" + query = query.where( + or_( + LogEvent.msg.ilike(like), + LogEvent.event.ilike(like), + LogEvent.request_id.ilike(like), + LogEvent.exc.ilike(like), + cast(LogEvent.attrs, Text).ilike(like), + ) + ) + if before is not None: + ts, row_id = _decode_cursor(before) + # The `ts <=` conjunct is implied by the OR; it is there so the planner can bound the + # `ts` index scan instead of filtering every newer row first. + query = query.where( + and_( + LogEvent.ts <= ts, + or_(LogEvent.ts < ts, and_(LogEvent.ts == ts, LogEvent.id < row_id)), + ) + ) + + # One more than asked, to know whether there is a next page without a second count query. + rows = ( + db.execute(query.order_by(LogEvent.ts.desc(), LogEvent.id.desc()).limit(limit + 1)) + .scalars() + .all() + ) + more = len(rows) > limit + rows = rows[:limit] + return { + "items": [_serialize(r) for r in rows], + "accounts": _accounts(db, {r.user_id for r in rows if r.user_id is not None}), + "next_cursor": _encode_cursor(rows[-1]) if more else None, + } + + +@router.get("/facets") +def log_facets( + since: datetime | None = None, + _: User = Depends(admin_user), + db: Session = Depends(get_db), +) -> dict: + """What the filter chips offer, with counts: every level, source and account present in the + window, and its most frequent events. Over the window alone, not the other filters — a chip + row that shrank as you clicked would hide the way back.""" + window = [LogEvent.ts >= _utc(since)] if since is not None else [] + + def counts(col, cap: int | None = None) -> list[dict]: + n = func.count().label("n") + stmt = select(col, n).where(*window).group_by(col).order_by(n.desc(), col) + if cap is not None: + stmt = stmt.limit(cap) + return [ + {"value": NONE if value is None else value, "count": count} + for value, count in db.execute(stmt).all() + ] + + users = counts(LogEvent.user_id) + return { + "levels": counts(LogEvent.level), + "sources": counts(LogEvent.source), + "events": counts(LogEvent.event, FACET_EVENTS_MAX), + "users": [{**u, "value": str(u["value"])} for u in users], + "accounts": _accounts(db, {u["value"] for u in users if u["value"] != NONE}), + "total": db.scalar(select(func.count()).select_from(LogEvent).where(*window)) or 0, + } diff --git a/backend/tests/test_admin_logs.py b/backend/tests/test_admin_logs.py new file mode 100644 index 00000000..3bf24702 --- /dev/null +++ b/backend/tests/test_admin_logs.py @@ -0,0 +1,154 @@ +"""The admin diagnostics console's query API (E22.6 S1), DB-lane.""" + +from datetime import datetime, timedelta, timezone + +import pytest +from sqlalchemy import insert + +from app.models import LogEvent + +T0 = datetime(2026, 9, 19, 12, 0, tzinfo=timezone.utc) + + +def _row(**kw) -> dict: + row = { + "ts": T0, + "level": "INFO", + "sev": 9, + "source": "api", + "logger": "siftlode.http", + "event": "http.request", + "msg": "GET /api/feed 200", + "attrs": None, + "exc": None, + "request_id": None, + "user_id": None, + "job_id": None, + "version": "0.90.0", + } + row.update(kw) + return row + + +def _add(db, *rows: dict) -> None: + db.execute(insert(LogEvent), list(rows)) + db.commit() + + +@pytest.fixture +def admin(db, make_user, as_user): + return as_user(make_user(email="admin@example.com", role="admin")) + + +def _msgs(resp) -> list[str]: + assert resp.status_code == 200, resp.text + return [it["msg"] for it in resp.json()["items"]] + + +def test_only_an_admin_reads_the_log(db, make_user, as_user, client): + _add(db, _row()) + assert as_user(make_user()).get("/api/admin/logs").status_code == 403 + assert as_user(make_user()).get("/api/admin/logs/facets").status_code == 403 + assert client.get("/api/admin/logs").status_code == 401 + + +def test_newest_first_and_every_row_once_across_pages_even_on_equal_timestamps(db, admin): + # A browser row's `ts` is clamped, so several rows share one: the id breaks the tie. + _add(db, *[_row(msg=f"same {i}") for i in range(5)], _row(ts=T0 - timedelta(1), msg="older")) + + seen, cursor = [], None + while True: + resp = admin.get( + "/api/admin/logs", params={"limit": 2, "before": cursor} if cursor else {"limit": 2} + ) + seen += _msgs(resp) + cursor = resp.json()["next_cursor"] + if cursor is None: + break + + assert seen == ["same 4", "same 3", "same 2", "same 1", "same 0", "older"] + + +@pytest.mark.parametrize( + ("params", "expected"), + [ + ({"levels": ["ERROR", "WARNING"]}, {"err", "warn"}), + ({"sources": ["worker"]}, {"err"}), + ({"user": "1"}, {"mine"}), + ({"user": "none"}, {"err", "warn", "plain"}), + ({"event": "download.job.failed"}, {"err"}), + ({"event": "none"}, {"warn"}), + ({"request_id": "req-abcdef12"}, {"mine"}), + ({"job_id": 42}, {"err"}), + ({"q": "FORBIDDEN"}, {"err"}), # in attrs, any case + ({"q": "KeyError"}, {"warn"}), # in the stack + ({"q": "100%"}, {"plain"}), # a literal, not a wildcard + ({"since": (T0 - timedelta(hours=1)).isoformat()}, {"mine", "err", "warn"}), + ({"until": (T0 - timedelta(hours=1)).isoformat()}, {"plain", "decoy"}), + ({"levels": ["INFO"], "user": "none"}, {"plain"}), + ], +) +def test_each_filter_selects_exactly_its_rows(db, admin, make_user, params, expected): + # user 1 is the admin the fixture made + _add( + db, + _row(msg="mine", user_id=1, request_id="req-abcdef12"), + _row( + msg="err", + level="ERROR", + sev=17, + source="worker", + event="download.job.failed", + job_id=42, + attrs={"code": "forbidden"}, + ), + _row(msg="warn", level="WARNING", sev=13, event=None, exc="Traceback …\nKeyError: 'x'"), + _row(msg="plain", ts=T0 - timedelta(hours=2), attrs={"note": "100% done"}), + # Matched by no filter above but `until`: another account, and `100%` read as a pattern. + _row(msg="decoy", ts=T0 - timedelta(hours=3), attrs={"note": "1000 done"}, user_id=99), + ) + assert set(_msgs(admin.get("/api/admin/logs", params=params))) == expected + + +def test_a_page_names_its_accounts_and_leaves_an_erased_one_bare(db, admin): + _add(db, _row(user_id=1), _row(user_id=77)) + + assert admin.get("/api/admin/logs").json()["accounts"] == {"1": "admin@example.com"} + + +@pytest.mark.parametrize( + "params", + [ + {"before": "not-a-cursor"}, + {"before": "WyJ4IiwgMV0="}, # ["x", 1]: decodes, not a timestamp + ], +) +def test_a_bad_cursor_is_a_400_with_a_code(db, admin, params): + resp = admin.get("/api/admin/logs", params=params) + assert resp.status_code == 400 + assert resp.json()["detail"]["code"] == "bad_cursor" + + +def test_facets_count_the_window_and_name_the_rows_without_a_value(db, admin): + _add( + db, + _row(user_id=1), + _row(user_id=1, level="ERROR", sev=17, event="sync.rss.failed"), + _row(event=None), + _row(ts=T0 - timedelta(days=2), source="worker"), + ) + + body = admin.get( + "/api/admin/logs/facets", params={"since": (T0 - timedelta(days=1)).isoformat()} + ).json() + + assert body["total"] == 3 + assert body["levels"] == [{"value": "INFO", "count": 2}, {"value": "ERROR", "count": 1}] + assert body["sources"] == [{"value": "api", "count": 3}] + assert body["events"] == [ + {"value": "http.request", "count": 1}, + {"value": "sync.rss.failed", "count": 1}, + {"value": "none", "count": 1}, + ] + assert body["users"] == [{"value": "1", "count": 2}, {"value": "none", "count": 1}] + assert body["accounts"] == {"1": "admin@example.com"}