""" Structured JSON logging for AARS backend. Writes newline-delimited JSON to: /app/logs/app-YYYY-MM-DD.jsonl - INFO+ (general) /app/logs/error-YYYY-MM-DD.jsonl - ERROR only /app/logs/access-YYYY-MM-DD.jsonl - HTTP access log Features: - ISO 8601 UTC timestamps - X-Request-ID context propagation via ContextVar - Automatic secret redaction (passwords, tokens, JWTs, etc.) - Daily rotation via TimedRotatingFileHandler """ import json import logging import re import sys from contextvars import ContextVar from datetime import datetime, timezone from logging.handlers import TimedRotatingFileHandler from typing import Any # ============ Request context ============ request_id_var: ContextVar[str] = ContextVar("request_id", default="-") user_id_var: ContextVar[int] = ContextVar("user_id", default=0) def set_request_id(rid: str) -> None: request_id_var.set(rid) def get_request_id() -> str: return request_id_var.get() def set_user_id(uid: int) -> None: user_id_var.set(uid) # ============ Redactor ============ REDACT_KEYS = { "password", "passwd", "pwd", "pass", "token", "access_token", "refresh_token", "id_token", "jwt", "bearer", "authorization", "api_key", "apikey", "secret", "client_secret", "cookie", "set-cookie", "session", "database_url", "db_url", "connection_string", "x-api-key", "x-auth-token", } REDACT_PATTERNS = [ (re.compile(r"eyJ[A-Za-z0-9_\-]+\.[A-Za-z0-9_\-]+\.[A-Za-z0-9_\-]+"), "[REDACTED:JWT]"), (re.compile(r"(?i)(bearer\s+)[A-Za-z0-9\-_.]+"), r"\1[REDACTED]"), ] REDACTED = "[REDACTED]" def _scrub_str(s: str) -> str: for pat, repl in REDACT_PATTERNS: s = pat.sub(repl, s) return s def _scrub(obj: Any) -> Any: """Recursively redact secrets from dict/list/str.""" if isinstance(obj, dict): out = {} for k, v in obj.items(): if isinstance(k, str) and k.lower() in REDACT_KEYS: out[k] = REDACTED else: out[k] = _scrub(v) return out if isinstance(obj, list): return [_scrub(v) for v in obj] if isinstance(obj, str): return _scrub_str(obj) return obj # ============ JSON Formatter ============ class JSONFormatter(logging.Formatter): SERVICE = "aars-backend" ENV = "production" def format(self, record: logging.LogRecord) -> str: # Standard fields payload = { "ts": datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z"), "level": record.levelname.lower(), "service": self.SERVICE, "env": self.ENV, "request_id": request_id_var.get(), "msg": record.getMessage(), } # Optional fields from extra= for key in ("route", "method", "status", "duration_ms", "user_id", "page_url", "app_version", "browser", "client_ip"): v = getattr(record, key, None) if v is not None: payload[key] = v # Exception info if record.exc_info: exc_type, exc_val, exc_tb = record.exc_info payload["error"] = { "name": exc_type.__name__ if exc_type else "Exception", "message": str(exc_val) if exc_val else "", "stack": self.formatException(record.exc_info), } # Custom extras (e.g. extra={"context": {...}}) if hasattr(record, "context") and record.context: payload["context"] = _scrub(record.context) # Scrub msg + all string fields payload = _scrub(payload) try: return json.dumps(payload, ensure_ascii=False, default=str) except (TypeError, ValueError): payload["msg"] = str(payload.get("msg", ""))[:500] return json.dumps(payload, ensure_ascii=False, default=str) # ============ Setup ============ LOG_DIR = "/app/logs" APP_RETENTION = 7 # days ERROR_RETENTION = 30 class DailyJSONLHandler(logging.Handler): """Writes one line per record to ``-YYYY-MM-DD.jsonl``. Switches file automatically when the local date changes. Honors a maximum retention by deleting old files on rollover. """ def __init__(self, base_name: str, retention_days: int, level: int = logging.INFO): super().__init__(level=level) self.base_name = base_name # e.g. "app" self.retention_days = retention_days self._date_str = None self._stream = None self.setFormatter(JSONFormatter()) def _current_path(self) -> str: import os return os.path.join(LOG_DIR, f"{self.base_name}-{self._date_str}.jsonl") def _open(self) -> None: import os os.makedirs(LOG_DIR, exist_ok=True) path = self._current_path() self._stream = open(path, "a", encoding="utf-8") self._cleanup_old() def _cleanup_old(self) -> None: import os import time cutoff = time.time() - (self.retention_days * 86400) prefix = f"{self.base_name}-" try: for fname in os.listdir(LOG_DIR): if not (fname.startswith(prefix) and fname.endswith(".jsonl")): continue fpath = os.path.join(LOG_DIR, fname) if os.path.getmtime(fpath) < cutoff: os.remove(fpath) except OSError: pass def _maybe_rollover(self) -> None: import os today = datetime.now().strftime("%Y-%m-%d") if self._date_str != today: if self._stream: self._stream.close() self._stream = None self._date_str = today self._open() def emit(self, record: logging.LogRecord) -> None: try: self._maybe_rollover() if self._stream: msg = self.format(record) self._stream.write(msg + "\n") self._stream.flush() except Exception: self.handleError(record) def close(self) -> None: if self._stream: self._stream.close() self._stream = None super().close() def setup_logging() -> None: """Configure root logger. Call once at app startup.""" root = logging.getLogger() root.setLevel(logging.INFO) # Clear handlers (uvicorn re-imports) for h in list(root.handlers): root.removeHandler(h) # App logger — INFO+ app_handler = DailyJSONLHandler("app", APP_RETENTION, level=logging.INFO) root.addHandler(app_handler) # Error logger — ERROR only error_handler = DailyJSONLHandler("error", ERROR_RETENTION, level=logging.ERROR) root.addHandler(error_handler) # Access logger (separate logger, not root) access_logger = logging.getLogger("access") access_logger.setLevel(logging.INFO) access_logger.propagate = False access_handler = DailyJSONLHandler("access", APP_RETENTION, level=logging.INFO) access_logger.addHandler(access_handler) # Console handler — pretty for dev/visible logs # Use a custom filter to inject request_id from ContextVar class _ContextFilter(logging.Filter): def filter(self, record): record.request_id = request_id_var.get() record.user_id = user_id_var.get() return True console = logging.StreamHandler(sys.stdout) console.setLevel(logging.INFO) console.setFormatter(logging.Formatter( "%(asctime)s %(levelname)-5s [%(request_id)s] %(name)s: %(message)s" )) console.addFilter(_ContextFilter()) root.addHandler(console) # Silence noisy libs logging.getLogger("uvicorn.access").setLevel(logging.WARNING) logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING) logging.getLogger(__name__).info("logging initialized", extra={"context": {"log_dir": LOG_DIR}}) def get_access_logger() -> logging.Logger: return logging.getLogger("access")