# (c) cavaliba.com - home - log.py import contextlib import contextvars import csv import getpass import json import logging import sys import time import uuid from datetime import datetime, timedelta from django.conf import settings from django.db.models import Q from django.http import StreamingHttpResponse from django.utils import timezone from app_home.configuration import get_configuration from .models import CavalibaEvent # stdlib level numbers, re-exported for call sites: log2.fail(..., level=WARNING) DEBUG = logging.DEBUG INFO = logging.INFO WARNING = logging.WARNING ERROR = logging.ERROR CRITICAL = logging.CRITICAL # ===================================================================== # log2 - stdlib-based logging with retained ECS events # ===================================================================== # # log2.debug(msg, *args, level=DEBUG) transient: export only, never stored # log2.access(action, **kw) retained, success, DEBUG # log2.ok(action, **kw) retained, success, INFO # log2.fail(action, **kw) retained, failure, ERROR # log2.deny(action, **kw) retained, denied, WARNING # log2.timed(action, **kw) as ev context manager: ok/fail + duration_ms; # ev["count"] = n inside the block # # kw: target=(type, id) | Instance, reason=, count=, message=, extra=, level=, # duration_ms=, domain=, category= # # Retained events are written directly to CavalibaEvent (never level-gated), # then exported on logger "cavaliba.event". Debug lines go to # "cavaliba.debug.". Routing and levels: settings.LOGGING. SUCCESS = "success" FAILURE = "failure" DENIED = "denied" AMBIENT_FIELDS = ( "channel", "origin", "domain", "actor", "actor_effective", "actor_kind", "client_ip", "request_id", ) ACTOR_FIELDS = ("actor", "actor_effective", "actor_kind") _ambient = contextvars.ContextVar("cavaliba_log_ambient", default=None) _event_logger = logging.getLogger("cavaliba.event") _max_lengths = None # serialized size limit of an event's extra: keeps a JSON export line well under Docker's # 16 KB line split, and far under MariaDB's max_allowed_packet EVENT_EXTRA_MAX = 12000 def _column_limits(): """Return {column: max_length} for CavalibaEvent char columns (cached).""" global _max_lengths if _max_lengths is None: _max_lengths = { f.name: f.max_length for f in CavalibaEvent._meta.get_fields() if getattr(f, "max_length", None) } return _max_lengths def _target(target): """Return (target_type, target_id) from a 2-tuple, an Instance, or None.""" if target is None: return "", "" if isinstance(target, (tuple, list)): return str(target[0] or ""), str(target[1] if len(target) > 1 else "") return str(getattr(target, "classname", "") or ""), str(getattr(target, "keyname", "") or "") def _safe_extra(extra): """Return extra as JSON-safe values (non-JSON types become strings), capped at EVENT_EXTRA_MAX serialized bytes: above it, only its scalar entries are kept, plus {"truncated": true, "size": n}.""" if not extra: return {} raw = json.dumps(extra, ensure_ascii=False, default=str) if len(raw.encode("utf-8")) <= EVENT_EXTRA_MAX: return json.loads(raw) data = json.loads(raw) kept = {k: v for k, v in data.items() if isinstance(v, (str, int, float, bool)) or v is None} kept = {k: v for k, v in kept.items() if not (isinstance(v, str) and len(v) > 200)} log2.debug("event extra truncated (%s bytes)", len(raw), level=logging.WARNING) return {**kept, "truncated": True, "size": len(raw)} def _levelno(level): """Return a stdlib level number from an int or a level name ("WARNING").""" if isinstance(level, int): return level value = logging.getLevelName(str(level).upper()) return value if isinstance(value, int) else logging.INFO def _is_excluded(client_ip): """Return True if LOG_EXCLUDED_IP prevents storing an event from this IP.""" try: excluded = get_configuration(appname="home", keyname="LOG_EXCLUDED_IP").split() except Exception: return False return "*" in excluded or (client_ip and client_ip in excluded) class Log2: """Single call-site logging API: transient debug lines and retained events.""" # ---- ambient context --------------------------------------------- def reset(self, **fields): """Replace the ambient context (entry points: view, ajax, api, cli, celery).""" ambient = {k: "" for k in AMBIENT_FIELDS} ambient["request_id"] = uuid.uuid4().hex[:16] ambient.update({k: v for k, v in fields.items() if v is not None}) return _ambient.set(ambient) def enter(self, **fields): """HTTP entry point (start_view, start_ajax, start_api): refine the context set by request_started. Keeps its request_id, clears the actor fields, applies fields.""" ambient = {k: "" for k in AMBIENT_FIELDS} ambient.update(_ambient.get() or {}) for key in ACTOR_FIELDS: ambient[key] = "" ambient.update({k: v for k, v in fields.items() if v is not None}) if not ambient["request_id"]: ambient["request_id"] = uuid.uuid4().hex[:16] return _ambient.set(ambient) def bind(self, **fields): """Merge fields into the ambient context.""" ambient = dict(_ambient.get() or {}) ambient.update({k: v for k, v in fields.items() if v is not None}) return _ambient.set(ambient) def restore(self, token): """Restore the ambient context saved by a previous reset()/bind() token.""" try: _ambient.reset(token) except Exception: pass def bind_aaa(self, aaa): """Bind actor fields from an aaa dict. client_ip only fills an empty one: on HTTP the request's own IP wins over aaa["user_ip"], cached in the session.""" if not aaa: return impersonate_by = aaa.get("impersonate_by", "") or "" if impersonate_by: actor, effective = impersonate_by, aaa.get("username", "") else: actor, effective = aaa.get("username", ""), "" if str(actor).startswith("API:"): kind = "apikey" else: kind = "user" if aaa.get("is_authenticated") else "anonymous" client_ip = None if (_ambient.get() or {}).get("client_ip") else aaa.get("user_ip", "") self.bind(actor=actor, actor_effective=effective, actor_kind=kind, client_ip=client_ip) def bind_command(self, command, action): """Reset the ambient context for a management command run.""" try: user = getpass.getuser() except Exception: user = "" return self.reset(channel="cli", origin=f"{command} {action}".strip(), actor=user, actor_kind="system") # fmt: skip def ambient(self): """Return a copy of the current ambient context.""" return dict(_ambient.get() or {}) # ---- transient --------------------------------------------------- def debug(self, msg, *args, level=logging.DEBUG, extra=None): """Emit a transient line on cavaliba.debug.; never stored.""" level = _levelno(level) module = sys._getframe(1).f_globals.get("__name__", "") dlogger = logging.getLogger(f"cavaliba.debug.{module}") if not dlogger.isEnabledFor(level): return ambient = dict(_ambient.get() or {}) if extra: ambient["extra"] = extra dlogger.log(level, msg, *args, extra={"ambient": ambient}, stacklevel=2) # ---- retained ---------------------------------------------------- def access(self, action, **kw): """Retained read/access event: success, DEBUG by default.""" return self._emit(action, SUCCESS, logging.DEBUG, **kw) def ok(self, action, **kw): """Retained operation success: INFO by default.""" return self._emit(action, SUCCESS, logging.INFO, **kw) def fail(self, action, **kw): """Retained operation failure: ERROR by default.""" return self._emit(action, FAILURE, logging.ERROR, **kw) def deny(self, action, **kw): """Retained authorization denial: WARNING by default.""" return self._emit(action, DENIED, logging.WARNING, **kw) @contextlib.contextmanager def timed(self, action, **kw): """Time a block; emit ok() with duration_ms, or fail() and re-raise. Yields a dict: keys set in the block (count, extra, ...) override kw. level= applies to the success only: a failure keeps fail()'s ERROR.""" fields = {} t0 = time.monotonic() try: yield fields except Exception as e: ms = int((time.monotonic() - t0) * 1000) fkw = {k: v for k, v in {**kw, **fields}.items() if k != "level"} fkw.setdefault("reason", "exception") fkw.setdefault("message", f"{type(e).__name__}: {e}") self.fail(action, duration_ms=ms, **fkw) _telemetry_timed(action, ms) raise ms = int((time.monotonic() - t0) * 1000) self.ok(action, duration_ms=ms, **{**kw, **fields}) _telemetry_timed(action, ms) def _emit(self, action, outcome, default_level, target=None, reason="", count=None, message="", extra=None, level=None, duration_ms=None, domain=None, category=""): # fmt: skip """Build, store and export one retained event; return it as a dict.""" level = default_level if level is None else _levelno(level) target_type, target_id = _target(target) event = {k: "" for k in AMBIENT_FIELDS} event.update(_ambient.get() or {}) if domain is not None: event["domain"] = domain if not event.get("channel"): event["channel"] = "system" event.update( created=timezone.now(), level=logging.getLevelName(level), severity=level, action=action or "", category=category or "", outcome=outcome, reason=reason or "", target_type=target_type, target_id=target_id, count=count, duration_ms=duration_ms, message=str(message or ""), extra=_safe_extra(extra), ) for key, size in _column_limits().items(): value = event.get(key) if isinstance(value, str) and len(value) > size: event[key] = value[:size] if getattr(settings, "CAVALIBA_LOG_DB", True) and not _is_excluded(event["client_ip"]): row = {k: v for k, v in event.items() if k in _MODEL_FIELDS} try: event["id"] = CavalibaEvent.objects.create(**row).id except Exception as e: event["db_error"] = str(e)[:200] print(f"log2: event not stored: {e}", file=sys.stderr) _event_logger.log(level, event["message"] or event["action"], extra={"event": event}) _telemetry_event(event["level"]) return event _MODEL_FIELDS = {f.name for f in CavalibaEvent._meta.get_fields() if f.name != "id"} def _telemetry_event(level): """Count a retained event in the telemetry counter buffer (key log.event, id=level): in-memory only, flushed once per request; never raises. log2.debug() is not counted.""" try: from app_data.telemetry import telemetry_bump telemetry_bump("log.event", id=str(level).lower()) except Exception: pass def _telemetry_timed(action, ms): """Feed a log2.timed() duration to telemetry as timer timed.; never raises. Imported lazily: app_data.telemetry imports this module.""" try: from app_data.telemetry import telemetry telemetry(f"timed.{action}", type="timer", value=ms) except Exception: pass def reset_on_request_started(sender=None, environ=None, **kwargs): """request_started receiver: fresh ambient context for each HTTP request, so a log2 call made before start_view/start_api never inherits the previous request's actor.""" environ = environ or {} path = environ.get("PATH_INFO", "") forwarded = environ.get("HTTP_X_FORWARDED_FOR", "") client_ip = (environ.get("HTTP_X_REAL_IP") or forwarded.split(",")[0] or environ.get("REMOTE_ADDR", "")) # fmt: skip log2.reset(channel="api" if path.startswith("/api/") else "web", origin=path, client_ip=client_ip.strip(), request_id=environ.get("HTTP_X_REQUEST_ID", "")[:64] or None) # fmt: skip log2 = Log2() # ---------------- # CavalibaEvent - count / list / purge # ---------------- def event_count(): """Return the number of stored events.""" return CavalibaEvent.objects.count() def event_list(start=0, size=100): """Return stored events, newest first, rows [start, start+size).""" return CavalibaEvent.objects.all().order_by("-id")[start : start + size] def event_purge_all(): """Delete every stored event; return the count removed.""" count = CavalibaEvent.objects.all().delete()[0] log2.ok("log.purge_all", count=count) return count def event_purge_level(level, keep_days): """Delete events of one level older than keep_days (all if 0); return the count.""" qs = CavalibaEvent.objects.filter(level=level) if keep_days > 0: qs = qs.filter(created__lte=timezone.now() - timedelta(days=keep_days)) return qs.delete()[0] def event_purge(): """Apply LOG_KEEP_DAYS_ and LOG_MAX_ENTRIES retention; return the count removed.""" count = 0 for level in ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"): keep_days = int(get_configuration(appname="home", keyname=f"LOG_KEEP_DAYS_{level}")) count += event_purge_level(level, keep_days) keep_entries = int(get_configuration(appname="home", keyname="LOG_MAX_ENTRIES")) count += _trim_to(keep_entries) log2.ok("log.purge", count=count) return count def _oldest_ids_cut(qs, n): """Return the id of the n-th oldest row of qs, or None if qs has fewer rows.""" return next(iter(qs.order_by("id").values_list("id", flat=True)[n - 1 : n]), None) def _trim_to(keep_entries): """Delete the oldest events above keep_entries, DEBUG ones first, so frequent low-level events don't push out older warnings and errors. Returns the count removed.""" excess = CavalibaEvent.objects.count() - keep_entries if excess <= 0: return 0 debug = CavalibaEvent.objects.filter(level="DEBUG") cut = _oldest_ids_cut(debug, excess) if cut is not None: return debug.filter(id__lte=cut).delete()[0] removed = debug.delete()[0] cut = _oldest_ids_cut(CavalibaEvent.objects.all(), excess - removed) if cut is not None: removed += CavalibaEvent.objects.filter(id__lte=cut).delete()[0] return removed # ---------------- # CavalibaEvent - filtering for the UI and export # ---------------- EVENT_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL") EVENT_OUTCOMES = (SUCCESS, FAILURE, DENIED) EVENT_CHANNELS = ("web", "api", "cli", "celery", "system") # query param -> max length kept EVENT_FILTER_PARAMS = { "level": 10, "outcome": 10, "channel": 10, "domain": 64, "action": 128, "actor": 256, "target": 300, "request_id": 64, "since": 10, "until": 10, "q": 200, } def event_filter_params(querydict): """Return the cleaned event filters found in a GET/POST querydict.""" params = {} for key, size in EVENT_FILTER_PARAMS.items(): value = (querydict.get(key) or "").strip()[:size] if value: params[key] = value if params.get("level", "").upper() not in EVENT_LEVELS: params.pop("level", None) if params.get("outcome") not in EVENT_OUTCOMES: params.pop("outcome", None) if params.get("channel") not in EVENT_CHANNELS: params.pop("channel", None) return params def _parse_day(value): """Return an aware datetime at 00:00 for a YYYY-MM-DD string, or None.""" try: day = datetime.strptime(value, "%Y-%m-%d") except (TypeError, ValueError): return None return timezone.make_aware(day) def event_queryset(params): """Return CavalibaEvent rows matching cleaned filters, newest first (Q objects only).""" qs = CavalibaEvent.objects.all() if params.get("level"): qs = qs.filter(severity__gte=logging.getLevelName(params["level"].upper())) for key in ("outcome", "channel", "domain", "request_id"): if params.get(key): qs = qs.filter(**{key: params[key]}) if params.get("action"): qs = qs.filter(action__istartswith=params["action"]) if params.get("actor"): actor = params["actor"] qs = qs.filter(Q(actor__istartswith=actor) | Q(actor_effective__istartswith=actor)) if params.get("target"): target = params["target"] if ":" in target: ttype, tid = target.split(":", 1) qs = qs.filter(target_type=ttype) if tid: qs = qs.filter(target_id__istartswith=tid) else: qs = qs.filter(target_id__istartswith=target) since = _parse_day(params.get("since")) if since: qs = qs.filter(created__gte=since) until = _parse_day(params.get("until")) if until: qs = qs.filter(created__lt=until + timedelta(days=1)) if params.get("q"): q = params["q"] qs = qs.filter(Q(message__icontains=q) | Q(target_id__icontains=q) | Q(reason=q)) return qs.order_by("-id") def event_as_dict(event): """Return a stored event as a plain dict (export, detail view).""" data = {f: getattr(event, f) for f in ["id", *sorted(_MODEL_FIELDS)]} data["created"] = event.created.isoformat() return data # ---------------- # CavalibaEvent - export (shared by the log page and /api/log/export/) # ---------------- EVENT_EXPORT_MAX = 100000 EVENT_EXPORT_COLUMNS = ( "id", "created", "level", "channel", "domain", "origin", "actor", "actor_effective", "actor_kind", "client_ip", "action", "outcome", "reason", "category", "target_type", "target_id", "count", "duration_ms", "request_id", "message", "extra", ) # fmt: skip class _Echo: """File-like object whose write() returns its input, for csv streaming.""" def write(self, value): """Return value unchanged.""" return value def _csv_cell(value): """Neutralise spreadsheet formulas: a text cell starting with = + - @ (or a tab / CR) gets a leading quote, so a spreadsheet shows it as text instead of evaluating it.""" if isinstance(value, str) and value[:1] in ("=", "+", "-", "@", "\t", "\r"): return "'" + value return value def _export_csv(rows): """Yield CSV lines (header first) for exported events.""" writer = csv.writer(_Echo()) yield writer.writerow(EVENT_EXPORT_COLUMNS) for event in rows: data = event_as_dict(event) data["extra"] = json.dumps(data["extra"], ensure_ascii=False, default=str) yield writer.writerow([_csv_cell(data.get(c, "")) for c in EVENT_EXPORT_COLUMNS]) def _export_json(rows): """Yield a JSON array of exported events, one object at a time.""" yield "[\n" first = True for event in rows: prefix = "" if first else ",\n" first = False yield prefix + json.dumps(event_as_dict(event), ensure_ascii=False, default=str) yield "\n]\n" def event_export_response(params, fmt): """Return a streamed CSV or JSON download of the events matching params (max EVENT_EXPORT_MAX rows, newest first). fmt must be "csv" or "json".""" rows = event_queryset(params)[:EVENT_EXPORT_MAX].iterator(chunk_size=2000) if fmt == "csv": stream, content_type = _export_csv(rows), "text/csv; charset=utf-8" else: stream, content_type = _export_json(rows), "application/json" stamp = timezone.now().strftime("%Y%m%d-%H%M%S") response = StreamingHttpResponse(stream, content_type=content_type) response["Content-Disposition"] = f'attachment; filename="cavaliba-log-{stamp}.{fmt}"' return response