peter
8f6f89832f
- Live stops polling on a failed read and marks the list as possibly stale (with Retry), instead of re-raising the error dialog every tick. - While live, a refetch reads the newest page only, whatever the cache holds (a late "Load older", a revisited filter set). - Copy is disabled and the list dimmed while a filter change loads. - Dates follow the UI language; the pivot chip uses the shared .chip. - Tests: query-string wire test, Copy, every facet's skip key.
235 lines
9.4 KiB
Python
235 lines
9.4 KiB
Python
"""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 dataclasses import dataclass
|
|
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"]
|
|
# The filters a facet can leave out of its own count (`LogFilters.conditions`).
|
|
Facet = Literal["levels", "sources", "user", "event"]
|
|
# `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}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LogFilters:
|
|
"""The console's filters, shared by the page and its facet counts."""
|
|
|
|
levels: list[str]
|
|
sources: list[str]
|
|
user: str | None
|
|
event: str | None
|
|
request_id: str | None
|
|
job_id: int | None
|
|
q: str | None
|
|
since: datetime | None
|
|
until: datetime | None
|
|
|
|
def conditions(self, skip: Facet | None = None) -> list:
|
|
"""The WHERE conditions, ANDed by the caller. `skip` leaves one filter out: a facet counts
|
|
its own values under every OTHER filter, so picking a level does not zero the other
|
|
levels' counts (the way back), while a request pivot does narrow them (honest counts)."""
|
|
out = []
|
|
if self.levels and skip != "levels":
|
|
out.append(LogEvent.level.in_(self.levels))
|
|
if self.sources and skip != "sources":
|
|
out.append(LogEvent.source.in_(self.sources))
|
|
if self.user is not None and skip != "user":
|
|
out.append(
|
|
LogEvent.user_id.is_(None)
|
|
if self.user == NONE
|
|
else LogEvent.user_id == int(self.user)
|
|
)
|
|
if self.event is not None and skip != "event":
|
|
out.append(
|
|
LogEvent.event.is_(None) if self.event == NONE else LogEvent.event == self.event
|
|
)
|
|
if self.request_id is not None:
|
|
out.append(LogEvent.request_id == self.request_id)
|
|
if self.job_id is not None:
|
|
out.append(LogEvent.job_id == self.job_id)
|
|
if self.since is not None:
|
|
out.append(LogEvent.ts >= _utc(self.since))
|
|
if self.until is not None:
|
|
out.append(LogEvent.ts < _utc(self.until))
|
|
if self.q and self.q.strip():
|
|
like = f"%{like_escape(self.q.strip())}%"
|
|
out.append(
|
|
or_(
|
|
LogEvent.msg.ilike(like),
|
|
LogEvent.event.ilike(like),
|
|
LogEvent.request_id.ilike(like),
|
|
LogEvent.exc.ilike(like),
|
|
cast(LogEvent.attrs, Text).ilike(like),
|
|
)
|
|
)
|
|
return out
|
|
|
|
|
|
def log_filters(
|
|
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,
|
|
) -> LogFilters:
|
|
return LogFilters(list(levels), list(sources), user, event, request_id, job_id, q, since, until)
|
|
|
|
|
|
@router.get("")
|
|
def list_logs(
|
|
filters: LogFilters = Depends(log_filters),
|
|
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).where(*filters.conditions())
|
|
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(
|
|
filters: LogFilters = Depends(log_filters),
|
|
_: User = Depends(admin_user),
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
"""What the filter chips offer, with counts: the levels, sources and accounts present, and the
|
|
most frequent events — each counted under every filter but its own (`LogFilters.conditions`).
|
|
`total` is the "all levels" count: every filter but the level one."""
|
|
|
|
def counts(col, skip: Facet, cap: int | None = None) -> list[dict]:
|
|
n = func.count().label("n")
|
|
stmt = select(col, n).where(*filters.conditions(skip)).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, "user")
|
|
total = select(func.count()).select_from(LogEvent).where(*filters.conditions("levels"))
|
|
return {
|
|
"levels": counts(LogEvent.level, "levels"),
|
|
"sources": counts(LogEvent.source, "sources"),
|
|
"events": counts(LogEvent.event, "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(total) or 0,
|
|
}
|