From 3663206744e358240b948b94001c93b65d6efedc Mon Sep 17 00:00:00 2001 From: Tyler Mathis <35553152+tsmathis@users.noreply.github.com> Date: Mon, 12 Oct 2026 12:12:50 -0700 Subject: [PATCH 1/2] structured logging for contribs backend servies - temp mp_logging module for testing --- mpcontribs-api/Dockerfile | 3 +- mpcontribs-api/gunicorn.conf.py | 17 +- mpcontribs-api/main.py | 14 +- mpcontribs-api/mp_logging/__init__.py | 39 ++ mpcontribs-api/mp_logging/config.py | 173 +++++++++ mpcontribs-api/mp_logging/context.py | 93 +++++ mpcontribs-api/mp_logging/flask_ext.py | 33 ++ mpcontribs-api/mp_logging/formatters.py | 350 ++++++++++++++++++ mpcontribs-api/mpcontribs/api/__init__.py | 78 ++-- mpcontribs-api/mpcontribs/api/core.py | 6 +- .../mpcontribs/api/notebooks/__init__.py | 2 +- .../mpcontribs/api/notebooks/views.py | 15 +- mpcontribs-kernel-gateway/Dockerfile | 7 +- .../jupyter_kernel_gateway_config.py | 32 ++ .../mp_logging/__init__.py | 39 ++ .../mp_logging/config.py | 173 +++++++++ .../mp_logging/context.py | 93 +++++ .../mp_logging/formatters.py | 350 ++++++++++++++++++ mpcontribs-kernel-gateway/start.sh | 4 +- mpcontribs-portal/Dockerfile | 3 +- mpcontribs-portal/gunicorn.conf.py | 15 +- mpcontribs-portal/mp_logging/__init__.py | 39 ++ mpcontribs-portal/mp_logging/config.py | 173 +++++++++ mpcontribs-portal/mp_logging/context.py | 93 +++++ mpcontribs-portal/mp_logging/django_ext.py | 37 ++ mpcontribs-portal/mp_logging/formatters.py | 350 ++++++++++++++++++ mpcontribs-portal/settings.py | 25 ++ 27 files changed, 2216 insertions(+), 40 deletions(-) create mode 100644 mpcontribs-api/mp_logging/__init__.py create mode 100644 mpcontribs-api/mp_logging/config.py create mode 100644 mpcontribs-api/mp_logging/context.py create mode 100644 mpcontribs-api/mp_logging/flask_ext.py create mode 100644 mpcontribs-api/mp_logging/formatters.py create mode 100644 mpcontribs-kernel-gateway/jupyter_kernel_gateway_config.py create mode 100644 mpcontribs-kernel-gateway/mp_logging/__init__.py create mode 100644 mpcontribs-kernel-gateway/mp_logging/config.py create mode 100644 mpcontribs-kernel-gateway/mp_logging/context.py create mode 100644 mpcontribs-kernel-gateway/mp_logging/formatters.py create mode 100644 mpcontribs-portal/mp_logging/__init__.py create mode 100644 mpcontribs-portal/mp_logging/config.py create mode 100644 mpcontribs-portal/mp_logging/context.py create mode 100644 mpcontribs-portal/mp_logging/django_ext.py create mode 100644 mpcontribs-portal/mp_logging/formatters.py diff --git a/mpcontribs-api/Dockerfile b/mpcontribs-api/Dockerfile index 749c63a755..8511bcb207 100644 --- a/mpcontribs-api/Dockerfile +++ b/mpcontribs-api/Dockerfile @@ -42,6 +42,7 @@ COPY main.py . COPY maintenance.py . COPY docker-entrypoint.sh . COPY gunicorn.conf.py . +COPY mp_logging mp_logging RUN chmod +x main.py scripts/start.sh scripts/start_rq.sh docker-entrypoint.sh ARG VERSION @@ -52,7 +53,7 @@ ENV DD_SERVICE=contribs-apis \ DD_TRACE_OTEL_ENABLED=false \ DD_MAIN_PACKAGE=mpcontribs-api -LABEL com.datadoghq.ad.logs='[{"source": "gunicorn", "service": "contribs-apis"}]' +LABEL com.datadoghq.ad.logs='[{"source": "mp-backend", "service": "contribs-apis"}]' EXPOSE 10000 10002 10003 10005 20000 ENTRYPOINT ["/app/docker-entrypoint.sh"] CMD ["/usr/bin/supervisord", "-c", "supervisord.conf"] diff --git a/mpcontribs-api/gunicorn.conf.py b/mpcontribs-api/gunicorn.conf.py index d6d1693c1c..36d8b2fbe6 100644 --- a/mpcontribs-api/gunicorn.conf.py +++ b/mpcontribs-api/gunicorn.conf.py @@ -1,15 +1,26 @@ import ddtrace.auto # noqa: F401 import os +from mp_logging import gunicorn_logconfig_dict + bind = "0.0.0.0:{}".format(os.getenv("API_PORT")) worker_class = "gevent" workers = os.getenv("NWORKERS") statsd_host = "{}:8125".format(os.getenv("DD_AGENT_HOST")) + +# gunicorn's own error and access logs go through the same structured logging +# config as the application (see `mp_logging`). Applied in the master before +# forking; the access log formatter maps gunicorn's atoms to Datadog's +# standard HTTP attributes, so the format string itself is only a fallback. +logconfig_dict = gunicorn_logconfig_dict( + app_loggers=("mpcontribs",), + # supervisord sets FLASK_DEBUG to "True" / "False" + debug=os.getenv("FLASK_DEBUG", "").strip().lower() in ("1", "true", "yes"), +) accesslog = "-" errorlog = "-" -access_log_format = '{}/{}: %(h)s %(t)s %(m)s %(U)s?%(q)s %(H)s %(s)s %(b)s "%(f)s" "%(a)s" %(D)s %(p)s %({{x-consumer-id}}i)s'.format( - os.getenv("SUPERVISOR_GROUP_NAME"), os.getenv("SUPERVISOR_PROCESS_NAME") -) +access_log_format = '%(h)s %(m)s %(U)s %(q)s %(s)s %(B)s %(D)s "%(f)s" "%(a)s"' + max_requests = os.getenv("MAX_REQUESTS") max_requests_jitter = os.getenv("MAX_REQUESTS_JITTER") proc_name = os.getenv("SUPERVISOR_PROCESS_NAME") diff --git a/mpcontribs-api/main.py b/mpcontribs-api/main.py index 2ec96e9fc6..6eaa4ca344 100755 --- a/mpcontribs-api/main.py +++ b/mpcontribs-api/main.py @@ -7,6 +7,9 @@ from supervisor.options import ClientOptions from supervisor.supervisorctl import Controller +from mp_logging import configure_logging + +configure_logging(app_loggers=("__main__",)) logger = logging.getLogger(__name__) client = boto3.client('ecs') @@ -26,7 +29,7 @@ def start(program): if metadata_uri: # in AWS Fargate with potentially multiple tasks - logger.debug("METADATA URI", metadata_uri) + logger.debug("Task metadata", extra={"metadata_uri": metadata_uri}) r = requests.get(metadata_uri) labels = r.json()["Labels"] prefix = "com.amazonaws.ecs" @@ -42,7 +45,14 @@ def start(program): ntasks += int(v == version) start_rq = ntasks == 1 # this task included in metadata response - logger.info(f"TASKS {ntasks}/{len(tasks)} -> START RQ {start_rq}") + logger.info( + "RQ workers decision", + extra={ + "tasks.current": ntasks, + "tasks.total": len(tasks), + "start_rq": start_rq, + }, + ) if start_rq: start("rq:*") diff --git a/mpcontribs-api/mp_logging/__init__.py b/mpcontribs-api/mp_logging/__init__.py new file mode 100644 index 0000000000..8facdbc939 --- /dev/null +++ b/mpcontribs-api/mp_logging/__init__.py @@ -0,0 +1,39 @@ +from __future__ import annotations + +from .config import ( + ACCESS_LOGGERS, + HANDLER, + build_logging_config, + configure_logging, + gunicorn_logconfig_dict, +) +from .context import ( + RequestContextFilter, + bind_request_context, + get_request_context, + request_fields, + reset_request_context, +) +from .formatters import ( + LOG_RECORD_BUILTIN_ATTRS, + ConsoleFormatter, + DatadogStructuredFormatter, + access_log_fields, +) + +__all__ = [ + "ACCESS_LOGGERS", + "HANDLER", + "LOG_RECORD_BUILTIN_ATTRS", + "ConsoleFormatter", + "DatadogStructuredFormatter", + "RequestContextFilter", + "access_log_fields", + "bind_request_context", + "build_logging_config", + "configure_logging", + "get_request_context", + "gunicorn_logconfig_dict", + "request_fields", + "reset_request_context", +] diff --git a/mpcontribs-api/mp_logging/config.py b/mpcontribs-api/mp_logging/config.py new file mode 100644 index 0000000000..4149465706 --- /dev/null +++ b/mpcontribs-api/mp_logging/config.py @@ -0,0 +1,173 @@ +"""`logging.config.dictConfig` configuration shared by MP backend processes.""" + +from __future__ import annotations + +import copy +import logging +import logging.config +import os +from collections.abc import Iterable, Mapping +from typing import Any, Literal + +from .formatters import ( + ASGI_ACCESS_LOGGER, + GUNICORN_ACCESS_LOGGER, + TORNADO_ACCESS_LOGGER, +) + +LogFormat = Literal["json", "console"] + +# Names are prefixed so they don't clash with the names other tools use when +# this config is merged into theirs (e.g. traitlets' "console" formatter) +HANDLER = "mp_stderr" + +# This package's import path (it is copied into each service's own package) +_PKG = __name__.rsplit(".", 1)[0] + +# Loggers that write access logs: always INFO, straight to the handler +ACCESS_LOGGERS = (GUNICORN_ACCESS_LOGGER, ASGI_ACCESS_LOGGER) + +# output key: LogRecord attribute (or "message" / "timestamp") +FMT_KEYS = { + "level": "levelname", + "message": "message", + "timestamp": "timestamp", + "logger.name": "name", + "logger.module": "module", + "logger.function": "funcName", + "logger.line": "lineno", + "logger.thread_name": "threadName", + "process.pid": "process", +} + +_BASE_CONFIG: dict[str, Any] = { + "version": 1, + "disable_existing_loggers": False, + "filters": { + "request_context": {"()": f"{_PKG}.context.RequestContextFilter"}, + }, + "formatters": { + "mp_json": { + "()": f"{_PKG}.formatters.DatadogStructuredFormatter", + "fmt_keys": FMT_KEYS, + }, + "mp_console": {"()": f"{_PKG}.formatters.ConsoleFormatter"}, + }, + "handlers": { + "mp_stderr": { + "class": "logging.StreamHandler", + "formatter": "mp_json", + "filters": ["request_context"], + "stream": "ext://sys.stderr", + }, + }, + "loggers": { + # gunicorn's (and uvicorn's) own loggers don't propagate, so they need + # the handler directly + "gunicorn.error": {"level": "INFO", "handlers": [HANDLER], "propagate": False}, + "uvicorn.error": {"level": "INFO", "handlers": [HANDLER], "propagate": False}, + }, + # Everything else (third-party libraries, captured warnings): + # WARNING and up (overridable with MP_LOG_LEVEL) + "root": {"level": "WARNING", "handlers": [HANDLER]}, +} + + +def _level_name(level: int | str) -> str: + if isinstance(level, int): + return logging.getLevelName(level) + name = level.strip().upper() + if name == "WARN": + name = "WARNING" + if not isinstance(logging.getLevelName(name), int): + raise ValueError(f"Unknown log level {level!r}") + return name + + +def build_logging_config( + *, + app_loggers: Iterable[str] = (), + access_loggers: Iterable[str] = ACCESS_LOGGERS, + extra_loggers: Mapping[str, Mapping[str, Any]] | None = None, + fmt: LogFormat | None = None, + level: int | str | None = None, + app_level: int | str | None = None, + debug: bool = False, +) -> dict[str, Any]: + """The `logging.config.dictConfig` dict for a service. + + Arguments left as None come from env vars. + + Args: + app_loggers : the service's own loggers (e.g. "mpcontribs"), logged at + `app_level` instead of the root level + access_loggers : loggers that write access logs (INFO, not propagated) + extra_loggers : extra per-logger config, applied last + fmt : "json" or "console" (`MP_LOG_FORMAT`, default "json") + level : root level (`MP_LOG_LEVEL`, default WARNING) + app_level : level for `app_loggers` (`MP_APP_LOG_LEVEL`, default + INFO, or DEBUG if `debug`) + debug : log the service's own loggers at DEBUG unless `app_level` + is set + + Returns: + dict for `logging.config.dictConfig` (a fresh copy, safe to modify) + """ + fmt = fmt or os.environ.get("MP_LOG_FORMAT") or "json" # type: ignore[assignment] + if fmt not in ("json", "console"): + raise ValueError(f"Unknown log format {fmt!r}, expected 'json' or 'console'") + level = _level_name(level or os.environ.get("MP_LOG_LEVEL") or "WARNING") + app_level = _level_name( + app_level + or os.environ.get("MP_APP_LOG_LEVEL") + or ("DEBUG" if debug else "INFO") + ) + + config = copy.deepcopy(_BASE_CONFIG) + config["handlers"][HANDLER]["formatter"] = f"mp_{fmt}" + config["root"]["level"] = level + loggers = config["loggers"] + for name in access_loggers: + loggers[name] = {"level": "INFO", "handlers": [HANDLER], "propagate": False} + for name in app_loggers: + loggers.setdefault(name, {})["level"] = app_level + for name, overrides in (extra_loggers or {}).items(): + loggers.setdefault(name, {}).update(copy.deepcopy(dict(overrides))) + return config + + +def configure_logging(**kwargs: Any) -> dict[str, Any]: + """Configure logging for the process. Safe to call more than once. + + Python warnings are captured into the `py.warnings` logger, so they are + formatted like everything else. See `build_logging_config` for the + arguments. + + Returns: + the applied config + """ + config = build_logging_config(**kwargs) + logging.config.dictConfig(config) + logging.captureWarnings(True) + return config + + +def gunicorn_logconfig_dict(**kwargs: Any) -> dict[str, Any]: + """Config for gunicorn's `logconfig_dict` setting (in `gunicorn.conf.py`). + + Applied by gunicorn in the master before forking, so gunicorn's own error + and access logs use the same handlers and formatters as the application. + Only imports this package, not the application. + """ + return build_logging_config(**kwargs) + + +__all__ = [ + "ACCESS_LOGGERS", + "FMT_KEYS", + "HANDLER", + "TORNADO_ACCESS_LOGGER", + "build_logging_config", + "configure_logging", + "gunicorn_logconfig_dict", +] diff --git a/mpcontribs-api/mp_logging/context.py b/mpcontribs-api/mp_logging/context.py new file mode 100644 index 0000000000..617e228810 --- /dev/null +++ b/mpcontribs-api/mp_logging/context.py @@ -0,0 +1,93 @@ +"""Per-request logging context, independent of the web framework. + +Framework glue (`flask_ext`, `django_ext`, `asgi`) binds the current +request's fields with `bind_request_context` at the start of a request and +resets them at the end. `RequestContextFilter` adds them to every record +logged meanwhile, including records from libraries. + +The context is a `contextvars.ContextVar`, so it is per-greenlet under gevent +and per-task under asyncio. +""" + +from __future__ import annotations + +import logging +from collections.abc import Mapping +from contextvars import ContextVar, Token +from typing import Any + +from .formatters import path_fields, process_name + +_REQUEST_CONTEXT: ContextVar[Mapping[str, Any] | None] = ContextVar( + "mp_logging_request_context", default=None +) + + +def _header(headers: Mapping[str, str], name: str) -> str | None: + """A request header from a case-insensitive or lower-cased mapping.""" + value = headers.get(name) + if value is None: + value = headers.get(name.lower()) + return value or None + + +def request_fields( + method: str | None, path: str | None, headers: Mapping[str, str] +) -> dict[str, Any]: + """Fields describing a request, from its method, path and headers. + + The Kong gateway forwards the consumer in `X-Consumer-Id` and the request + id in `Kong-Request-Id` (or `X-Request-Id` if set upstream). + """ + fields = { + "http.method": method, + "http.url_details.path": path, + **path_fields(path), + "usr.id": _header(headers, "X-Consumer-Id"), + "http.request_id": _header(headers, "X-Request-Id") + or _header(headers, "Kong-Request-Id"), + } + return {k: v for k, v in fields.items() if v} + + +def bind_request_context(fields: Mapping[str, Any]) -> Token: + """Set the current request's fields. Pass the token to `reset_request_context`.""" + return _REQUEST_CONTEXT.set(dict(fields)) + + +def reset_request_context(token: Token | None) -> None: + """Restore the context from before `bind_request_context`. Never raises.""" + if token is None: + return + try: + _REQUEST_CONTEXT.reset(token) + except (ValueError, RuntimeError): + # Token from another context (e.g. teardown ran elsewhere): just clear + _REQUEST_CONTEXT.set(None) + + +def get_request_context() -> Mapping[str, Any] | None: + """The current request's fields, or None outside a request.""" + return _REQUEST_CONTEXT.get() + + +class RequestContextFilter(logging.Filter): + """Add the process name, and the current request's context, to every record. + + Attached to handlers (not loggers), so it also applies to records from + libraries that reach the service's handlers. Outside a request (startup, + background work) records only get the process name. Never drops a record. + """ + + def filter(self, record: logging.LogRecord) -> bool: + # The supervisor program (e.g. contribs-api, micro-worker), on every record + if not hasattr(record, "process.name") and (name := process_name()): + setattr(record, "process.name", name) + context = _REQUEST_CONTEXT.get() + if not context: + return True + for key, value in context.items(): + # Fields set explicitly at the call site win + if not hasattr(record, key): + setattr(record, key, value) + return True diff --git a/mpcontribs-api/mp_logging/flask_ext.py b/mpcontribs-api/mp_logging/flask_ext.py new file mode 100644 index 0000000000..4cd7f91cc3 --- /dev/null +++ b/mpcontribs-api/mp_logging/flask_ext.py @@ -0,0 +1,33 @@ +"""Request logging context for Flask apps.""" + +from __future__ import annotations + +from typing import Any + +from flask import Flask, g, request + +from .context import ( + bind_request_context, + request_fields, + reset_request_context, +) + +_G_KEY = "_mp_logging_token" + + +def init_app(app: Flask) -> None: + """Add the current request's fields to every record logged during it. + + The access log itself comes from gunicorn (`gunicorn.access`). + """ + + @app.before_request + def _bind_logging_context() -> None: + token = bind_request_context( + request_fields(request.method, request.path, request.headers) + ) + setattr(g, _G_KEY, token) + + @app.teardown_request + def _reset_logging_context(exc: Any = None) -> None: + reset_request_context(g.pop(_G_KEY, None)) diff --git a/mpcontribs-api/mp_logging/formatters.py b/mpcontribs-api/mp_logging/formatters.py new file mode 100644 index 0000000000..dd2441b8fd --- /dev/null +++ b/mpcontribs-api/mp_logging/formatters.py @@ -0,0 +1,350 @@ +"""Log formatters: Datadog-friendly JSON for deployments, readable text for dev. + +Adapted from `mp_ui.core.logging.formatters` (MP frontend), which is itself +based on the dagster pipelines' `common.logging.formatters`. + +Every attribute on a record that isn't a builtin `LogRecord` attribute is +emitted as a field. That includes `extra=` fields, the request context added +by `RequestContextFilter`, and the `dd.*` trace-correlation attributes ddtrace +adds when `DD_LOGS_INJECTION=true`. + +Access-log records are turned into Datadog's standard HTTP attributes: + +- gunicorn (`gunicorn.access`) and `mp_logging.asgi` (`asgi.access`) log each + request as `logger.info(format, atoms)`, with gunicorn's atom names +- tornado (`tornado.access`, e.g. the Jupyter kernel gateway) logs + `"%d %s %.2fms"` with (status, "METHOD URI (IP)", milliseconds) +""" + +from __future__ import annotations + +import datetime as dt +import json +import logging +import os +import re +from collections.abc import Mapping +from typing import Any + +# Attributes every LogRecord has. Anything else on a record is a structured field. +# These names also can't be used as `extra=` keys (logging raises KeyError). +LOG_RECORD_BUILTIN_ATTRS = frozenset( + { + "args", + "asctime", + "created", + "exc_info", + "exc_text", + "filename", + "funcName", + "levelname", + "levelno", + "lineno", + "module", + "msecs", + "message", + "msg", + "name", + "pathname", + "process", + "processName", + "relativeCreated", + "stack_info", + "thread", + "threadName", + "taskName", + } +) + +GUNICORN_ACCESS_LOGGER = "gunicorn.access" +ASGI_ACCESS_LOGGER = "asgi.access" +TORNADO_ACCESS_LOGGER = "tornado.access" +# Loggers whose records carry gunicorn-style access atoms +ATOM_ACCESS_LOGGERS = frozenset({GUNICORN_ACCESS_LOGGER, ASGI_ACCESS_LOGGER}) + +# Same patterns as the "Parse URL Path root" / "First two elements from URL +# Path" grok parsers of the previous Datadog pipeline, so facets keep their values: +# /materials/summary/mp-149 -> path_root "/materials", path_first_two "/materials/summary" +_PATH_ROOT = re.compile(r"([^/]*/){1}[^/]+") +_PATH_FIRST_TWO = re.compile(r"([^/]*/){2}[^/]+") + +# tornado.web.RequestHandler._request_summary(): "GET /api/kernels (127.0.0.1)" +_TORNADO_SUMMARY = re.compile(r"^(?P\S+) (?P\S+) \((?P[^)]*)\)$") + + +def path_fields(path: str | None) -> dict[str, str]: + """`http.url_details.path_root` / `path_first_two` for a URL path (if they exist).""" + fields: dict[str, str] = {} + if not path: + return fields + if match := _PATH_ROOT.match(path): + fields["http.url_details.path_root"] = match.group(0) + if match := _PATH_FIRST_TWO.match(path): + fields["http.url_details.path_first_two"] = match.group(0) + return fields + + +def status_category(status: int | None) -> str | None: + """Datadog status for an HTTP status code: info (1xx-3xx), warning (4xx), error (5xx).""" + if status is None: + return None + if status >= 500: + return "error" + if status >= 400: + return "warning" + return "info" + + +# `level` given to access records, by status category +_ACCESS_LEVELS = {"info": "INFO", "warning": "WARNING", "error": "ERROR"} + + +def process_name() -> str | None: + """The supervisor program this process runs under (e.g. contribs-api), if any.""" + return os.environ.get("SUPERVISOR_PROCESS_NAME") or None + + +def _atom(atoms: Mapping[str, Any], key: str) -> Any: + """An access-log atom, or None if it's missing or "-".""" + try: + value = atoms[key] + except KeyError: + return None + return None if value in ("-", "", None) else value + + +def _as_int(value: Any) -> int | None: + try: + return int(value) + except (TypeError, ValueError): + return None + + +def _client(atoms: Mapping[str, Any]) -> tuple[Any, int | None]: + """Client IP and port. gunicorn gives the port as an environ atom; + the ASGI middleware gives it as its own atom.""" + ip = _atom(atoms, "h") + port = _as_int(_atom(atoms, "{remote_port}e")) + return ip, port + + +def _http_fields( + *, + method: Any, + path: Any, + query: Any, + status: int | None, + duration_ns: int | None, +) -> dict[str, Any]: + category = status_category(status) + return { + # Access records take their level from the response status, so 4xx/5xx + # responses show as warnings/errors + "level": _ACCESS_LEVELS.get(category) if category else None, + "http.method": method, + # Full URL for a Datadog URL parser, which fills `http.url_details` + # (including `queryString` as an object of parameters) + "http.url": f"{path}?{query}" if query else path, + "http.url_details.path": path, + **path_fields(path), + "http.status_code": status, + "http.status_category": category, + # Datadog's standard duration attribute is in nanoseconds + "duration": duration_ns, + } + + +def _atom_access_fields(atoms: Mapping[str, Any]) -> dict[str, Any]: + method = _atom(atoms, "m") + path = _atom(atoms, "U") + status = _as_int(_atom(atoms, "s")) + duration_us = _as_int(_atom(atoms, "D")) + ip, port = _client(atoms) + return { + "message": f"{method} {path} {status}", + **_http_fields( + method=method, + path=path, + query=_atom(atoms, "q"), + status=status, + duration_ns=duration_us * 1000 if duration_us is not None else None, + ), + "http.version": _atom(atoms, "H"), + "http.referer": _atom(atoms, "f"), + "http.useragent": _atom(atoms, "a"), + "http.request_id": _atom(atoms, "{x-request-id}i") + or _atom(atoms, "{kong-request-id}i"), + "network.client.ip": ip, + "network.client.port": port, + "network.bytes_written": _as_int(_atom(atoms, "B")), + # Kong consumer, as forwarded by the gateway + "usr.id": _atom(atoms, "{x-consumer-id}i"), + "usr.groups": _atom(atoms, "{x-consumer-groups}o") + or _atom(atoms, "{x-consumer-groups}i") + or _atom(atoms, "{x-authenticated-groups}i"), + # Dash callback name (web only; absent for APIs) + "dash.callback": _atom(atoms, "{x-callback-name}o"), + "process.pid": _as_int(str(_atom(atoms, "p") or "").strip("<>")), + "process.name": process_name(), + } + + +def _tornado_access_fields(args: tuple[Any, ...]) -> dict[str, Any] | None: + status, summary, ms = args + match = _TORNADO_SUMMARY.match(str(summary)) + if match is None: + return None + method, uri, ip = match.group("method", "uri", "ip") + path, _, query = uri.partition("?") + status = _as_int(status) + try: + duration_ns = int(float(ms) * 1_000_000) + except (TypeError, ValueError): + duration_ns = None + return { + "message": f"{method} {path} {status}", + **_http_fields( + method=method, + path=path, + query=query, + status=status, + duration_ns=duration_ns, + ), + "network.client.ip": ip or None, + "process.name": process_name(), + } + + +def access_log_fields(record: logging.LogRecord) -> dict[str, Any] | None: + """Structured fields for an access-log record. + + Field names follow Datadog's standard attributes so its HTTP facets work + without remapping. + + Returns: + dict of fields (including "message"), or None if `record` isn't an + access-log record. + """ + fields: dict[str, Any] | None = None + if record.name in ATOM_ACCESS_LOGGERS and isinstance(record.args, Mapping): + fields = _atom_access_fields(record.args) + elif ( + record.name == TORNADO_ACCESS_LOGGER + and isinstance(record.args, tuple) + and len(record.args) == 3 + ): + fields = _tornado_access_fields(record.args) + if fields is None: + return None + return {k: v for k, v in fields.items() if v is not None} + + +# Attributes some frameworks add to their records, renamed to Datadog's +# standard attributes (e.g. Django's `django.request` / `django.server`) +_RENAMED_ATTRS = {"status_code": "http.status_code"} +# Attributes some frameworks add that aren't useful as fields (Django's +# request/server objects, whose useful parts are already in the request +# context; uvicorn's ANSI-colored copy of the message) +_DROPPED_ATTRS = frozenset({"request", "server_time", "color_message"}) + + +def extra_fields(record: logging.LogRecord) -> dict[str, Any]: + """Non-builtin attributes of a record (`extra=`, filters, ddtrace).""" + return { + _RENAMED_ATTRS.get(key, key): val + for key, val in record.__dict__.items() + if key not in LOG_RECORD_BUILTIN_ATTRS and key not in _DROPPED_ATTRS + } + + +class DatadogStructuredFormatter(logging.Formatter): + """One JSON object per record, using Datadog's standard attribute names.""" + + def __init__(self, *, fmt_keys: dict[str, str] | None = None) -> None: + """Create the formatter. + + Args: + fmt_keys (dict or None) : output key -> LogRecord attribute, or + "message" / "timestamp" for the formatted message and the + ISO-8601 UTC time. + """ + super().__init__() + self.fmt_keys = fmt_keys if fmt_keys is not None else {} + + def format(self, record: logging.LogRecord) -> str: + return json.dumps(self._prepare_log_dict(record), default=str) + + def _prepare_log_dict(self, record: logging.LogRecord) -> dict[str, Any]: + access = access_log_fields(record) + always_fields: dict[str, Any] = { + "message": ( + access.pop("message") + if access is not None + else record.getMessage().rstrip() + ), + "timestamp": dt.datetime.fromtimestamp( + record.created, tz=dt.timezone.utc + ).isoformat(), + } + + # `exc_info` is (None, None, None) when `logger.exception` is called + # outside an `except` block + if record.exc_info and record.exc_info[0] is not None: + exc_type, exc_value, _ = record.exc_info + # Datadog's standard error attributes + always_fields["error.kind"] = exc_type.__name__ + always_fields["error.message"] = str(exc_value) + always_fields["error.stack"] = self.formatException(record.exc_info) + + # Only present when a call site passes `stack_info=True` + if record.stack_info is not None: + always_fields["error.stack_info"] = self.formatStack(record.stack_info) + + message = { + key: ( + msg_val + if (msg_val := always_fields.pop(attr, None)) is not None + else getattr(record, attr, None) + ) + for key, attr in self.fmt_keys.items() + } + message.update(always_fields) + if access is not None: + message.update(access) + message.update(extra_fields(record)) + return message + + +class ConsoleFormatter(logging.Formatter): + """Readable single-line records for local development. + + `2026-10-08 12:00:00,000 WARNING [mpcontribs.api] Message key=value ...` + """ + + def __init__( + self, + fmt: str = "%(asctime)s %(levelname)-8s [%(name)s] %(message)s", + datefmt: str | None = None, + ) -> None: + super().__init__(fmt=fmt, datefmt=datefmt) + + def format(self, record: logging.LogRecord) -> str: + access = access_log_fields(record) + if access is not None: + # Format a copy, so other handlers still see the original record + record = logging.makeLogRecord(record.__dict__) + record.msg, record.args = access.pop("message"), None + if level := access.pop("level", None): + record.levelname = level + text = super().format(record) + + fields = {**(access or {}), **extra_fields(record)} + # ddtrace's injected ids are noise locally + fields = { + k: v for k, v in fields.items() if not k.startswith("dd.") and v is not None + } + if fields: + pairs = " ".join(f"{k}={v}" for k, v in fields.items()) + first, sep, rest = text.partition("\n") + text = f"{first} {pairs}{sep}{rest}" + return text diff --git a/mpcontribs-api/mpcontribs/api/__init__.py b/mpcontribs-api/mpcontribs/api/__init__.py index e61583b4b8..0a70ce26fa 100644 --- a/mpcontribs-api/mpcontribs/api/__init__.py +++ b/mpcontribs-api/mpcontribs/api/__init__.py @@ -58,19 +58,38 @@ ) -class CustomLoggerAdapter(logging.LoggerAdapter): - def process(self, msg, kwargs): - prefix = self.extra.get("prefix") - return f"[{prefix}] {msg}" if prefix else msg, kwargs +def get_logger(name): + """Logger for `name`. + Formatting, levels and the per-request/process fields (`process.name`, + `usr.id`, ...) come from the structured logging config (`mp_logging`, + applied in `configure_logging` and gunicorn.conf.py), so this is a plain + `logging.getLogger`. Kept for existing imports. + """ + return logging.getLogger(name) -def get_logger(name): - logger = logging.getLogger(name) - process = os.environ.get("SUPERVISOR_PROCESS_NAME") - group = os.environ.get("SUPERVISOR_GROUP_NAME") - cfg = {"prefix": f"{group}/{process}"} if process and group else {} - logger.setLevel("DEBUG" if os.environ.get("FLASK_DEBUG") else "INFO") - return CustomLoggerAdapter(logger, cfg) + +def flask_debug(): + """FLASK_DEBUG as a bool (supervisord sets it to "True" / "False").""" + return os.environ.get("FLASK_DEBUG", "").strip().lower() in ("1", "true", "yes") + + +def configure_logging(): + """Structured (Datadog JSON) logging for this process, see `mp_logging`. + + `mp_logging` is a top-level module next to the app (not part of the + `mpcontribs.api` package). If it isn't importable (e.g. the package is + used outside the service image), the process' existing config is kept. + """ + try: + from mp_logging import configure_logging as _configure + except ImportError: + return None + return _configure( + # rq.worker: job start/finish messages (`flask rq worker`) + app_loggers=("mpcontribs", "__main__", "rq.worker"), + debug=flask_debug(), + ) logger = get_logger(__name__) @@ -125,7 +144,7 @@ def send_email(to, subject, html): ses_client.starttls() ses_client.login(os.environ["SMTP_USERNAME"], os.environ["SMTP_PASSWORD"]) ses_client.send_message(msg) - logger.warning(f"Email with subject `{subject}` sent to `{to}`") + logger.warning("Email sent", extra={"email.subject": subject, "email.to": to}) def get_collections(db): @@ -166,7 +185,7 @@ def get_kernels(): try: r = requests.get(get_kernel_endpoint(), timeout=2) except (ConnectionError, Timeout): - logger.warning("Kernel Gateway NOT AVAILABLE") + logger.warning("Kernel Gateway not available") return None response = r.json() @@ -174,7 +193,13 @@ def get_kernels(): idx = int(os.environ.get("DEPLOYMENT", 1)) if len(response) < nkernels * (idx + 1): - logger.error("NOT ENOUGH KERNELS AVAILABLE") + logger.error( + "Not enough kernels available", + extra={ + "kernels.available": len(response), + "kernels.required": nkernels * (idx + 1), + }, + ) return None return {kernel["id"]: None for kernel in response[idx : idx + 3]} @@ -192,6 +217,7 @@ def get_consumer(): def create_app(): """create flask app""" + configure_logging() app = Flask(__name__) app.config.from_pyfile("config.py", silent=True) app.config["USTS"] = URLSafeTimedSerializer(app.secret_key) @@ -201,7 +227,7 @@ def create_app(): app.json.sort_keys = False MPCONTRIBS_API_HOST = app.config.get("MPCONTRIBS_API_HOST") app.config["TEMPLATE"]["schemes"] = ["http"] if app.debug else ["https"] - logger.info("database: " + app.config["MPCONTRIBS_DB"]) + logger.info("Using database", extra={"db.name": app.config["MPCONTRIBS_DB"]}) Compress(app) Marshmallow(app) MongoEngine(app) @@ -223,8 +249,8 @@ def create_app(): module_path = ".".join(["mpcontribs", "api", collection, "views"]) try: module = import_module(module_path) - except ModuleNotFoundError as ex: - logger.error(f"API module {module_path}: {ex}") + except ModuleNotFoundError: + logger.exception("API module not found", extra={"module": module_path}) continue try: @@ -232,9 +258,12 @@ def create_app(): app.register_blueprint(blueprint, url_prefix="/" + collection) klass = getattr(module, collection.capitalize() + "View") register_class(app, klass, name=collection) - logger.info(f"{collection} registered") - except AttributeError as ex: - logger.error(f"Failed to register {module_path}: {collection} {ex}") + logger.info("Collection registered", extra={"collection": collection}) + except AttributeError: + logger.exception( + "Failed to register collection", + extra={"module": module_path, "collection": collection}, + ) if getattr(app, "kernels", None): from mpcontribs.api.notebooks.views import rq @@ -248,10 +277,17 @@ def healthcheck(): app.register_blueprint(sse, url_prefix="/stream") app.add_url_rule("/healthcheck", view_func=healthcheck) + try: + from mp_logging import flask_ext + except ImportError: + pass + else: + flask_ext.init_app(app) + @app.after_request def add_header(response): response.headers["X-Consumer-Id"] = request.headers.get("X-Consumer-Id") return response - logger.info("app created.") + logger.info("App created") return app diff --git a/mpcontribs-api/mpcontribs/api/core.py b/mpcontribs-api/mpcontribs/api/core.py index ff24418d64..73d5000dec 100644 --- a/mpcontribs-api/mpcontribs/api/core.py +++ b/mpcontribs-api/mpcontribs/api/core.py @@ -452,7 +452,11 @@ def __init_subclass__(cls, **kwargs): with open(file_path, "w") as f: yaml.dump(spec, f) logger.debug( - f"{cls.tags[0]}.{method.__name__} written to {file_path}" + "API spec written", + extra={ + "spec": f"{cls.tags[0]}.{method.__name__}", + "file_path": file_path, + }, ) def get_groups(self, request): diff --git a/mpcontribs-api/mpcontribs/api/notebooks/__init__.py b/mpcontribs-api/mpcontribs/api/notebooks/__init__.py index 2db2955a69..106ca150bc 100644 --- a/mpcontribs-api/mpcontribs/api/notebooks/__init__.py +++ b/mpcontribs-api/mpcontribs/api/notebooks/__init__.py @@ -8,7 +8,7 @@ def run_cells(kernel_id, cid, cells): - logger.debug(f"running {cid} on {kernel_id}") + logger.debug("Running cells", extra={"contribution": cid, "kernel_id": kernel_id}) ws = create_kernel_connection(kernel_id) outputs = {} diff --git a/mpcontribs-api/mpcontribs/api/notebooks/views.py b/mpcontribs-api/mpcontribs/api/notebooks/views.py index e3bf88af8c..90e37611cc 100644 --- a/mpcontribs-api/mpcontribs/api/notebooks/views.py +++ b/mpcontribs-api/mpcontribs/api/notebooks/views.py @@ -68,9 +68,14 @@ def execute_cells(cid, cells): current_app.kernels[kernel_id] = None return outputs else: - logger.warning(f"{kernel_id} busy with {running_cid}") + logger.warning( + "Kernel busy", + extra={"kernel_id": kernel_id, "contribution": running_cid}, + ) - logger.warning("WAITING for a kernel to become available") + logger.warning( + "Waiting for a kernel to become available", extra={"tries": ntries} + ) sleep(5) ntries += 1 @@ -186,12 +191,14 @@ def make(projects=None, cids=None, force=False): nb = Notebooks.objects.get(id=document.notebook.id) nb.delete() document.update(unset__notebook="") - logger.debug(f"Notebook {document.notebook.id} deleted.") + logger.debug( + "Notebook deleted", extra={"notebook": str(document.notebook.id)} + ) except DoesNotExist: pass cid = str(document.id) - logger.debug(f"prep notebook for {cid} ...") + logger.debug("Preparing notebook", extra={"contribution": cid}) document.reload("tables", "structures", "attachments") cells = [ diff --git a/mpcontribs-kernel-gateway/Dockerfile b/mpcontribs-kernel-gateway/Dockerfile index 04de9bc1d2..a7eb3572e4 100644 --- a/mpcontribs-kernel-gateway/Dockerfile +++ b/mpcontribs-kernel-gateway/Dockerfile @@ -42,12 +42,15 @@ ENV PATH=/root/.local/bin:$PATH \ DD_LOGS_INJECTION=true \ DD_TRACE_SAMPLE_RATE=1 \ DD_TRACE_OTEL_ENABLED=false \ - DD_PROFILING_LOCK_ENABLED=false + DD_PROFILING_LOCK_ENABLED=false \ + JUPYTER_CONFIG_DIR=/app EXPOSE 10100 COPY make_seed.py . COPY start.sh . COPY healthcheck.sh . +COPY jupyter_kernel_gateway_config.py . +COPY mp_logging mp_logging RUN python make_seed.py && chmod +x start.sh healthcheck.sh -LABEL com.datadoghq.ad.logs='[{"source": "gunicorn", "service": "contribs-kernel-gateway"}]' +LABEL com.datadoghq.ad.logs='[{"source": "mp-backend", "service": "contribs-kernel-gateway"}]' CMD ./start.sh diff --git a/mpcontribs-kernel-gateway/jupyter_kernel_gateway_config.py b/mpcontribs-kernel-gateway/jupyter_kernel_gateway_config.py new file mode 100644 index 0000000000..66f345067c --- /dev/null +++ b/mpcontribs-kernel-gateway/jupyter_kernel_gateway_config.py @@ -0,0 +1,32 @@ +"""Kernel gateway config: structured (Datadog JSON) logging, see `mp_logging`. + +Loaded by `jupyter kernelgateway` from `JUPYTER_CONFIG_DIR` (set in the +Dockerfile to this file's directory). +""" + +import os +import sys + +# mp_logging is copied next to this file +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from mp_logging import HANDLER, build_logging_config # noqa: E402 + +c = get_config() # noqa: F821 + +_logging = build_logging_config( + app_loggers=("KernelGatewayApp",), + extra_loggers={ + # tornado's request log (logs successes at INFO, 4xx at WARNING, 5xx at ERROR) + "tornado.access": {"level": "INFO", "handlers": [HANDLER], "propagate": False}, + }, +) +# traitlets merges this into its default logging config, whose "console" +# handler the app's own logger (KernelGatewayApp) writes to: give it the same +# formatter and filter +_logging["handlers"]["console"] = { + "formatter": _logging["handlers"][HANDLER]["formatter"], + "filters": ["request_context"], +} +c.Application.logging_config = _logging +c.Application.log_level = os.environ.get("MP_APP_LOG_LEVEL", "INFO") diff --git a/mpcontribs-kernel-gateway/mp_logging/__init__.py b/mpcontribs-kernel-gateway/mp_logging/__init__.py new file mode 100644 index 0000000000..8facdbc939 --- /dev/null +++ b/mpcontribs-kernel-gateway/mp_logging/__init__.py @@ -0,0 +1,39 @@ +from __future__ import annotations + +from .config import ( + ACCESS_LOGGERS, + HANDLER, + build_logging_config, + configure_logging, + gunicorn_logconfig_dict, +) +from .context import ( + RequestContextFilter, + bind_request_context, + get_request_context, + request_fields, + reset_request_context, +) +from .formatters import ( + LOG_RECORD_BUILTIN_ATTRS, + ConsoleFormatter, + DatadogStructuredFormatter, + access_log_fields, +) + +__all__ = [ + "ACCESS_LOGGERS", + "HANDLER", + "LOG_RECORD_BUILTIN_ATTRS", + "ConsoleFormatter", + "DatadogStructuredFormatter", + "RequestContextFilter", + "access_log_fields", + "bind_request_context", + "build_logging_config", + "configure_logging", + "get_request_context", + "gunicorn_logconfig_dict", + "request_fields", + "reset_request_context", +] diff --git a/mpcontribs-kernel-gateway/mp_logging/config.py b/mpcontribs-kernel-gateway/mp_logging/config.py new file mode 100644 index 0000000000..4149465706 --- /dev/null +++ b/mpcontribs-kernel-gateway/mp_logging/config.py @@ -0,0 +1,173 @@ +"""`logging.config.dictConfig` configuration shared by MP backend processes.""" + +from __future__ import annotations + +import copy +import logging +import logging.config +import os +from collections.abc import Iterable, Mapping +from typing import Any, Literal + +from .formatters import ( + ASGI_ACCESS_LOGGER, + GUNICORN_ACCESS_LOGGER, + TORNADO_ACCESS_LOGGER, +) + +LogFormat = Literal["json", "console"] + +# Names are prefixed so they don't clash with the names other tools use when +# this config is merged into theirs (e.g. traitlets' "console" formatter) +HANDLER = "mp_stderr" + +# This package's import path (it is copied into each service's own package) +_PKG = __name__.rsplit(".", 1)[0] + +# Loggers that write access logs: always INFO, straight to the handler +ACCESS_LOGGERS = (GUNICORN_ACCESS_LOGGER, ASGI_ACCESS_LOGGER) + +# output key: LogRecord attribute (or "message" / "timestamp") +FMT_KEYS = { + "level": "levelname", + "message": "message", + "timestamp": "timestamp", + "logger.name": "name", + "logger.module": "module", + "logger.function": "funcName", + "logger.line": "lineno", + "logger.thread_name": "threadName", + "process.pid": "process", +} + +_BASE_CONFIG: dict[str, Any] = { + "version": 1, + "disable_existing_loggers": False, + "filters": { + "request_context": {"()": f"{_PKG}.context.RequestContextFilter"}, + }, + "formatters": { + "mp_json": { + "()": f"{_PKG}.formatters.DatadogStructuredFormatter", + "fmt_keys": FMT_KEYS, + }, + "mp_console": {"()": f"{_PKG}.formatters.ConsoleFormatter"}, + }, + "handlers": { + "mp_stderr": { + "class": "logging.StreamHandler", + "formatter": "mp_json", + "filters": ["request_context"], + "stream": "ext://sys.stderr", + }, + }, + "loggers": { + # gunicorn's (and uvicorn's) own loggers don't propagate, so they need + # the handler directly + "gunicorn.error": {"level": "INFO", "handlers": [HANDLER], "propagate": False}, + "uvicorn.error": {"level": "INFO", "handlers": [HANDLER], "propagate": False}, + }, + # Everything else (third-party libraries, captured warnings): + # WARNING and up (overridable with MP_LOG_LEVEL) + "root": {"level": "WARNING", "handlers": [HANDLER]}, +} + + +def _level_name(level: int | str) -> str: + if isinstance(level, int): + return logging.getLevelName(level) + name = level.strip().upper() + if name == "WARN": + name = "WARNING" + if not isinstance(logging.getLevelName(name), int): + raise ValueError(f"Unknown log level {level!r}") + return name + + +def build_logging_config( + *, + app_loggers: Iterable[str] = (), + access_loggers: Iterable[str] = ACCESS_LOGGERS, + extra_loggers: Mapping[str, Mapping[str, Any]] | None = None, + fmt: LogFormat | None = None, + level: int | str | None = None, + app_level: int | str | None = None, + debug: bool = False, +) -> dict[str, Any]: + """The `logging.config.dictConfig` dict for a service. + + Arguments left as None come from env vars. + + Args: + app_loggers : the service's own loggers (e.g. "mpcontribs"), logged at + `app_level` instead of the root level + access_loggers : loggers that write access logs (INFO, not propagated) + extra_loggers : extra per-logger config, applied last + fmt : "json" or "console" (`MP_LOG_FORMAT`, default "json") + level : root level (`MP_LOG_LEVEL`, default WARNING) + app_level : level for `app_loggers` (`MP_APP_LOG_LEVEL`, default + INFO, or DEBUG if `debug`) + debug : log the service's own loggers at DEBUG unless `app_level` + is set + + Returns: + dict for `logging.config.dictConfig` (a fresh copy, safe to modify) + """ + fmt = fmt or os.environ.get("MP_LOG_FORMAT") or "json" # type: ignore[assignment] + if fmt not in ("json", "console"): + raise ValueError(f"Unknown log format {fmt!r}, expected 'json' or 'console'") + level = _level_name(level or os.environ.get("MP_LOG_LEVEL") or "WARNING") + app_level = _level_name( + app_level + or os.environ.get("MP_APP_LOG_LEVEL") + or ("DEBUG" if debug else "INFO") + ) + + config = copy.deepcopy(_BASE_CONFIG) + config["handlers"][HANDLER]["formatter"] = f"mp_{fmt}" + config["root"]["level"] = level + loggers = config["loggers"] + for name in access_loggers: + loggers[name] = {"level": "INFO", "handlers": [HANDLER], "propagate": False} + for name in app_loggers: + loggers.setdefault(name, {})["level"] = app_level + for name, overrides in (extra_loggers or {}).items(): + loggers.setdefault(name, {}).update(copy.deepcopy(dict(overrides))) + return config + + +def configure_logging(**kwargs: Any) -> dict[str, Any]: + """Configure logging for the process. Safe to call more than once. + + Python warnings are captured into the `py.warnings` logger, so they are + formatted like everything else. See `build_logging_config` for the + arguments. + + Returns: + the applied config + """ + config = build_logging_config(**kwargs) + logging.config.dictConfig(config) + logging.captureWarnings(True) + return config + + +def gunicorn_logconfig_dict(**kwargs: Any) -> dict[str, Any]: + """Config for gunicorn's `logconfig_dict` setting (in `gunicorn.conf.py`). + + Applied by gunicorn in the master before forking, so gunicorn's own error + and access logs use the same handlers and formatters as the application. + Only imports this package, not the application. + """ + return build_logging_config(**kwargs) + + +__all__ = [ + "ACCESS_LOGGERS", + "FMT_KEYS", + "HANDLER", + "TORNADO_ACCESS_LOGGER", + "build_logging_config", + "configure_logging", + "gunicorn_logconfig_dict", +] diff --git a/mpcontribs-kernel-gateway/mp_logging/context.py b/mpcontribs-kernel-gateway/mp_logging/context.py new file mode 100644 index 0000000000..617e228810 --- /dev/null +++ b/mpcontribs-kernel-gateway/mp_logging/context.py @@ -0,0 +1,93 @@ +"""Per-request logging context, independent of the web framework. + +Framework glue (`flask_ext`, `django_ext`, `asgi`) binds the current +request's fields with `bind_request_context` at the start of a request and +resets them at the end. `RequestContextFilter` adds them to every record +logged meanwhile, including records from libraries. + +The context is a `contextvars.ContextVar`, so it is per-greenlet under gevent +and per-task under asyncio. +""" + +from __future__ import annotations + +import logging +from collections.abc import Mapping +from contextvars import ContextVar, Token +from typing import Any + +from .formatters import path_fields, process_name + +_REQUEST_CONTEXT: ContextVar[Mapping[str, Any] | None] = ContextVar( + "mp_logging_request_context", default=None +) + + +def _header(headers: Mapping[str, str], name: str) -> str | None: + """A request header from a case-insensitive or lower-cased mapping.""" + value = headers.get(name) + if value is None: + value = headers.get(name.lower()) + return value or None + + +def request_fields( + method: str | None, path: str | None, headers: Mapping[str, str] +) -> dict[str, Any]: + """Fields describing a request, from its method, path and headers. + + The Kong gateway forwards the consumer in `X-Consumer-Id` and the request + id in `Kong-Request-Id` (or `X-Request-Id` if set upstream). + """ + fields = { + "http.method": method, + "http.url_details.path": path, + **path_fields(path), + "usr.id": _header(headers, "X-Consumer-Id"), + "http.request_id": _header(headers, "X-Request-Id") + or _header(headers, "Kong-Request-Id"), + } + return {k: v for k, v in fields.items() if v} + + +def bind_request_context(fields: Mapping[str, Any]) -> Token: + """Set the current request's fields. Pass the token to `reset_request_context`.""" + return _REQUEST_CONTEXT.set(dict(fields)) + + +def reset_request_context(token: Token | None) -> None: + """Restore the context from before `bind_request_context`. Never raises.""" + if token is None: + return + try: + _REQUEST_CONTEXT.reset(token) + except (ValueError, RuntimeError): + # Token from another context (e.g. teardown ran elsewhere): just clear + _REQUEST_CONTEXT.set(None) + + +def get_request_context() -> Mapping[str, Any] | None: + """The current request's fields, or None outside a request.""" + return _REQUEST_CONTEXT.get() + + +class RequestContextFilter(logging.Filter): + """Add the process name, and the current request's context, to every record. + + Attached to handlers (not loggers), so it also applies to records from + libraries that reach the service's handlers. Outside a request (startup, + background work) records only get the process name. Never drops a record. + """ + + def filter(self, record: logging.LogRecord) -> bool: + # The supervisor program (e.g. contribs-api, micro-worker), on every record + if not hasattr(record, "process.name") and (name := process_name()): + setattr(record, "process.name", name) + context = _REQUEST_CONTEXT.get() + if not context: + return True + for key, value in context.items(): + # Fields set explicitly at the call site win + if not hasattr(record, key): + setattr(record, key, value) + return True diff --git a/mpcontribs-kernel-gateway/mp_logging/formatters.py b/mpcontribs-kernel-gateway/mp_logging/formatters.py new file mode 100644 index 0000000000..dd2441b8fd --- /dev/null +++ b/mpcontribs-kernel-gateway/mp_logging/formatters.py @@ -0,0 +1,350 @@ +"""Log formatters: Datadog-friendly JSON for deployments, readable text for dev. + +Adapted from `mp_ui.core.logging.formatters` (MP frontend), which is itself +based on the dagster pipelines' `common.logging.formatters`. + +Every attribute on a record that isn't a builtin `LogRecord` attribute is +emitted as a field. That includes `extra=` fields, the request context added +by `RequestContextFilter`, and the `dd.*` trace-correlation attributes ddtrace +adds when `DD_LOGS_INJECTION=true`. + +Access-log records are turned into Datadog's standard HTTP attributes: + +- gunicorn (`gunicorn.access`) and `mp_logging.asgi` (`asgi.access`) log each + request as `logger.info(format, atoms)`, with gunicorn's atom names +- tornado (`tornado.access`, e.g. the Jupyter kernel gateway) logs + `"%d %s %.2fms"` with (status, "METHOD URI (IP)", milliseconds) +""" + +from __future__ import annotations + +import datetime as dt +import json +import logging +import os +import re +from collections.abc import Mapping +from typing import Any + +# Attributes every LogRecord has. Anything else on a record is a structured field. +# These names also can't be used as `extra=` keys (logging raises KeyError). +LOG_RECORD_BUILTIN_ATTRS = frozenset( + { + "args", + "asctime", + "created", + "exc_info", + "exc_text", + "filename", + "funcName", + "levelname", + "levelno", + "lineno", + "module", + "msecs", + "message", + "msg", + "name", + "pathname", + "process", + "processName", + "relativeCreated", + "stack_info", + "thread", + "threadName", + "taskName", + } +) + +GUNICORN_ACCESS_LOGGER = "gunicorn.access" +ASGI_ACCESS_LOGGER = "asgi.access" +TORNADO_ACCESS_LOGGER = "tornado.access" +# Loggers whose records carry gunicorn-style access atoms +ATOM_ACCESS_LOGGERS = frozenset({GUNICORN_ACCESS_LOGGER, ASGI_ACCESS_LOGGER}) + +# Same patterns as the "Parse URL Path root" / "First two elements from URL +# Path" grok parsers of the previous Datadog pipeline, so facets keep their values: +# /materials/summary/mp-149 -> path_root "/materials", path_first_two "/materials/summary" +_PATH_ROOT = re.compile(r"([^/]*/){1}[^/]+") +_PATH_FIRST_TWO = re.compile(r"([^/]*/){2}[^/]+") + +# tornado.web.RequestHandler._request_summary(): "GET /api/kernels (127.0.0.1)" +_TORNADO_SUMMARY = re.compile(r"^(?P\S+) (?P\S+) \((?P[^)]*)\)$") + + +def path_fields(path: str | None) -> dict[str, str]: + """`http.url_details.path_root` / `path_first_two` for a URL path (if they exist).""" + fields: dict[str, str] = {} + if not path: + return fields + if match := _PATH_ROOT.match(path): + fields["http.url_details.path_root"] = match.group(0) + if match := _PATH_FIRST_TWO.match(path): + fields["http.url_details.path_first_two"] = match.group(0) + return fields + + +def status_category(status: int | None) -> str | None: + """Datadog status for an HTTP status code: info (1xx-3xx), warning (4xx), error (5xx).""" + if status is None: + return None + if status >= 500: + return "error" + if status >= 400: + return "warning" + return "info" + + +# `level` given to access records, by status category +_ACCESS_LEVELS = {"info": "INFO", "warning": "WARNING", "error": "ERROR"} + + +def process_name() -> str | None: + """The supervisor program this process runs under (e.g. contribs-api), if any.""" + return os.environ.get("SUPERVISOR_PROCESS_NAME") or None + + +def _atom(atoms: Mapping[str, Any], key: str) -> Any: + """An access-log atom, or None if it's missing or "-".""" + try: + value = atoms[key] + except KeyError: + return None + return None if value in ("-", "", None) else value + + +def _as_int(value: Any) -> int | None: + try: + return int(value) + except (TypeError, ValueError): + return None + + +def _client(atoms: Mapping[str, Any]) -> tuple[Any, int | None]: + """Client IP and port. gunicorn gives the port as an environ atom; + the ASGI middleware gives it as its own atom.""" + ip = _atom(atoms, "h") + port = _as_int(_atom(atoms, "{remote_port}e")) + return ip, port + + +def _http_fields( + *, + method: Any, + path: Any, + query: Any, + status: int | None, + duration_ns: int | None, +) -> dict[str, Any]: + category = status_category(status) + return { + # Access records take their level from the response status, so 4xx/5xx + # responses show as warnings/errors + "level": _ACCESS_LEVELS.get(category) if category else None, + "http.method": method, + # Full URL for a Datadog URL parser, which fills `http.url_details` + # (including `queryString` as an object of parameters) + "http.url": f"{path}?{query}" if query else path, + "http.url_details.path": path, + **path_fields(path), + "http.status_code": status, + "http.status_category": category, + # Datadog's standard duration attribute is in nanoseconds + "duration": duration_ns, + } + + +def _atom_access_fields(atoms: Mapping[str, Any]) -> dict[str, Any]: + method = _atom(atoms, "m") + path = _atom(atoms, "U") + status = _as_int(_atom(atoms, "s")) + duration_us = _as_int(_atom(atoms, "D")) + ip, port = _client(atoms) + return { + "message": f"{method} {path} {status}", + **_http_fields( + method=method, + path=path, + query=_atom(atoms, "q"), + status=status, + duration_ns=duration_us * 1000 if duration_us is not None else None, + ), + "http.version": _atom(atoms, "H"), + "http.referer": _atom(atoms, "f"), + "http.useragent": _atom(atoms, "a"), + "http.request_id": _atom(atoms, "{x-request-id}i") + or _atom(atoms, "{kong-request-id}i"), + "network.client.ip": ip, + "network.client.port": port, + "network.bytes_written": _as_int(_atom(atoms, "B")), + # Kong consumer, as forwarded by the gateway + "usr.id": _atom(atoms, "{x-consumer-id}i"), + "usr.groups": _atom(atoms, "{x-consumer-groups}o") + or _atom(atoms, "{x-consumer-groups}i") + or _atom(atoms, "{x-authenticated-groups}i"), + # Dash callback name (web only; absent for APIs) + "dash.callback": _atom(atoms, "{x-callback-name}o"), + "process.pid": _as_int(str(_atom(atoms, "p") or "").strip("<>")), + "process.name": process_name(), + } + + +def _tornado_access_fields(args: tuple[Any, ...]) -> dict[str, Any] | None: + status, summary, ms = args + match = _TORNADO_SUMMARY.match(str(summary)) + if match is None: + return None + method, uri, ip = match.group("method", "uri", "ip") + path, _, query = uri.partition("?") + status = _as_int(status) + try: + duration_ns = int(float(ms) * 1_000_000) + except (TypeError, ValueError): + duration_ns = None + return { + "message": f"{method} {path} {status}", + **_http_fields( + method=method, + path=path, + query=query, + status=status, + duration_ns=duration_ns, + ), + "network.client.ip": ip or None, + "process.name": process_name(), + } + + +def access_log_fields(record: logging.LogRecord) -> dict[str, Any] | None: + """Structured fields for an access-log record. + + Field names follow Datadog's standard attributes so its HTTP facets work + without remapping. + + Returns: + dict of fields (including "message"), or None if `record` isn't an + access-log record. + """ + fields: dict[str, Any] | None = None + if record.name in ATOM_ACCESS_LOGGERS and isinstance(record.args, Mapping): + fields = _atom_access_fields(record.args) + elif ( + record.name == TORNADO_ACCESS_LOGGER + and isinstance(record.args, tuple) + and len(record.args) == 3 + ): + fields = _tornado_access_fields(record.args) + if fields is None: + return None + return {k: v for k, v in fields.items() if v is not None} + + +# Attributes some frameworks add to their records, renamed to Datadog's +# standard attributes (e.g. Django's `django.request` / `django.server`) +_RENAMED_ATTRS = {"status_code": "http.status_code"} +# Attributes some frameworks add that aren't useful as fields (Django's +# request/server objects, whose useful parts are already in the request +# context; uvicorn's ANSI-colored copy of the message) +_DROPPED_ATTRS = frozenset({"request", "server_time", "color_message"}) + + +def extra_fields(record: logging.LogRecord) -> dict[str, Any]: + """Non-builtin attributes of a record (`extra=`, filters, ddtrace).""" + return { + _RENAMED_ATTRS.get(key, key): val + for key, val in record.__dict__.items() + if key not in LOG_RECORD_BUILTIN_ATTRS and key not in _DROPPED_ATTRS + } + + +class DatadogStructuredFormatter(logging.Formatter): + """One JSON object per record, using Datadog's standard attribute names.""" + + def __init__(self, *, fmt_keys: dict[str, str] | None = None) -> None: + """Create the formatter. + + Args: + fmt_keys (dict or None) : output key -> LogRecord attribute, or + "message" / "timestamp" for the formatted message and the + ISO-8601 UTC time. + """ + super().__init__() + self.fmt_keys = fmt_keys if fmt_keys is not None else {} + + def format(self, record: logging.LogRecord) -> str: + return json.dumps(self._prepare_log_dict(record), default=str) + + def _prepare_log_dict(self, record: logging.LogRecord) -> dict[str, Any]: + access = access_log_fields(record) + always_fields: dict[str, Any] = { + "message": ( + access.pop("message") + if access is not None + else record.getMessage().rstrip() + ), + "timestamp": dt.datetime.fromtimestamp( + record.created, tz=dt.timezone.utc + ).isoformat(), + } + + # `exc_info` is (None, None, None) when `logger.exception` is called + # outside an `except` block + if record.exc_info and record.exc_info[0] is not None: + exc_type, exc_value, _ = record.exc_info + # Datadog's standard error attributes + always_fields["error.kind"] = exc_type.__name__ + always_fields["error.message"] = str(exc_value) + always_fields["error.stack"] = self.formatException(record.exc_info) + + # Only present when a call site passes `stack_info=True` + if record.stack_info is not None: + always_fields["error.stack_info"] = self.formatStack(record.stack_info) + + message = { + key: ( + msg_val + if (msg_val := always_fields.pop(attr, None)) is not None + else getattr(record, attr, None) + ) + for key, attr in self.fmt_keys.items() + } + message.update(always_fields) + if access is not None: + message.update(access) + message.update(extra_fields(record)) + return message + + +class ConsoleFormatter(logging.Formatter): + """Readable single-line records for local development. + + `2026-10-08 12:00:00,000 WARNING [mpcontribs.api] Message key=value ...` + """ + + def __init__( + self, + fmt: str = "%(asctime)s %(levelname)-8s [%(name)s] %(message)s", + datefmt: str | None = None, + ) -> None: + super().__init__(fmt=fmt, datefmt=datefmt) + + def format(self, record: logging.LogRecord) -> str: + access = access_log_fields(record) + if access is not None: + # Format a copy, so other handlers still see the original record + record = logging.makeLogRecord(record.__dict__) + record.msg, record.args = access.pop("message"), None + if level := access.pop("level", None): + record.levelname = level + text = super().format(record) + + fields = {**(access or {}), **extra_fields(record)} + # ddtrace's injected ids are noise locally + fields = { + k: v for k, v in fields.items() if not k.startswith("dd.") and v is not None + } + if fields: + pairs = " ".join(f"{k}={v}" for k, v in fields.items()) + first, sep, rest = text.partition("\n") + text = f"{first} {pairs}{sep}{rest}" + return text diff --git a/mpcontribs-kernel-gateway/start.sh b/mpcontribs-kernel-gateway/start.sh index f86e4900db..af1e5bd842 100644 --- a/mpcontribs-kernel-gateway/start.sh +++ b/mpcontribs-kernel-gateway/start.sh @@ -10,4 +10,6 @@ if [[ -n "$DD_TRACE_HOST" ]]; then wait-for-it.sh "$DD_TRACE_HOST" -q -s -t 10 && CMD="ddtrace-run $CMD" || echo "WARNING: datadog agent unreachable" fi -exec $CMD --KernelGatewayApp.log_format='%(asctime)s,%(msecs)03d %(levelname)s [%(name)s] [%(module)s:%(lineno)d] - %(message)s' +# Logging (structured JSON, incl. the request log) is configured in +# jupyter_kernel_gateway_config.py (JUPYTER_CONFIG_DIR) +exec $CMD diff --git a/mpcontribs-portal/Dockerfile b/mpcontribs-portal/Dockerfile index df95d147ec..92eca34163 100644 --- a/mpcontribs-portal/Dockerfile +++ b/mpcontribs-portal/Dockerfile @@ -69,6 +69,7 @@ COPY start.sh . COPY maintenance.py . COPY gunicorn.conf.py . COPY healthchecks.py . +COPY mp_logging mp_logging ARG VERSION ENV DD_SERVICE=contribs-portals \ @@ -77,7 +78,7 @@ ENV DD_SERVICE=contribs-portals \ DD_TRACE_HOST=localhost:8126 \ DD_TRACE_OTEL_ENABLED=false -LABEL com.datadoghq.ad.logs='[{"source": "gunicorn", "service": "contribs-portals"}]' +LABEL com.datadoghq.ad.logs='[{"source": "mp-backend", "service": "contribs-portals"}]' EXPOSE 8080 8082 8083 8085 ENTRYPOINT ["/app/docker-entrypoint.sh"] CMD ["/usr/bin/supervisord", "-c", "supervisord.conf"] diff --git a/mpcontribs-portal/gunicorn.conf.py b/mpcontribs-portal/gunicorn.conf.py index 612305305a..75f06fca32 100644 --- a/mpcontribs-portal/gunicorn.conf.py +++ b/mpcontribs-portal/gunicorn.conf.py @@ -1,14 +1,23 @@ import os +from mp_logging import gunicorn_logconfig_dict + bind = "0.0.0.0:{}".format(os.getenv("PORTAL_PORT")) worker_class = "gevent" workers = os.getenv("NWORKERS") statsd_host = "{}:8125".format(os.getenv("DD_AGENT_HOST")) + +# gunicorn's own error and access logs go through the same structured logging +# config as the application (see `mp_logging`). Applied in the master before +# forking; the access log formatter maps gunicorn's atoms to Datadog's +# standard HTTP attributes, so the format string itself is only a fallback. +logconfig_dict = gunicorn_logconfig_dict( + app_loggers=("mpcontribs",), debug=os.getenv("NODE_ENV") == "development" +) accesslog = "-" errorlog = "-" -access_log_format = '{}/{}: %(h)s %(t)s %(m)s %(U)s?%(q)s %(H)s %(s)s %(b)s "%(f)s" "%(a)s" %(D)s %(p)s %({{x-consumer-id}}i)s'.format( - os.getenv("SUPERVISOR_GROUP_NAME"), os.getenv("SUPERVISOR_PROCESS_NAME") -) +access_log_format = '%(h)s %(m)s %(U)s %(q)s %(s)s %(B)s %(D)s "%(f)s" "%(a)s"' + max_requests = os.getenv("MAX_REQUESTS") max_requests_jitter = os.getenv("MAX_REQUESTS_JITTER") proc_name = os.getenv("SUPERVISOR_PROCESS_NAME") diff --git a/mpcontribs-portal/mp_logging/__init__.py b/mpcontribs-portal/mp_logging/__init__.py new file mode 100644 index 0000000000..8facdbc939 --- /dev/null +++ b/mpcontribs-portal/mp_logging/__init__.py @@ -0,0 +1,39 @@ +from __future__ import annotations + +from .config import ( + ACCESS_LOGGERS, + HANDLER, + build_logging_config, + configure_logging, + gunicorn_logconfig_dict, +) +from .context import ( + RequestContextFilter, + bind_request_context, + get_request_context, + request_fields, + reset_request_context, +) +from .formatters import ( + LOG_RECORD_BUILTIN_ATTRS, + ConsoleFormatter, + DatadogStructuredFormatter, + access_log_fields, +) + +__all__ = [ + "ACCESS_LOGGERS", + "HANDLER", + "LOG_RECORD_BUILTIN_ATTRS", + "ConsoleFormatter", + "DatadogStructuredFormatter", + "RequestContextFilter", + "access_log_fields", + "bind_request_context", + "build_logging_config", + "configure_logging", + "get_request_context", + "gunicorn_logconfig_dict", + "request_fields", + "reset_request_context", +] diff --git a/mpcontribs-portal/mp_logging/config.py b/mpcontribs-portal/mp_logging/config.py new file mode 100644 index 0000000000..4149465706 --- /dev/null +++ b/mpcontribs-portal/mp_logging/config.py @@ -0,0 +1,173 @@ +"""`logging.config.dictConfig` configuration shared by MP backend processes.""" + +from __future__ import annotations + +import copy +import logging +import logging.config +import os +from collections.abc import Iterable, Mapping +from typing import Any, Literal + +from .formatters import ( + ASGI_ACCESS_LOGGER, + GUNICORN_ACCESS_LOGGER, + TORNADO_ACCESS_LOGGER, +) + +LogFormat = Literal["json", "console"] + +# Names are prefixed so they don't clash with the names other tools use when +# this config is merged into theirs (e.g. traitlets' "console" formatter) +HANDLER = "mp_stderr" + +# This package's import path (it is copied into each service's own package) +_PKG = __name__.rsplit(".", 1)[0] + +# Loggers that write access logs: always INFO, straight to the handler +ACCESS_LOGGERS = (GUNICORN_ACCESS_LOGGER, ASGI_ACCESS_LOGGER) + +# output key: LogRecord attribute (or "message" / "timestamp") +FMT_KEYS = { + "level": "levelname", + "message": "message", + "timestamp": "timestamp", + "logger.name": "name", + "logger.module": "module", + "logger.function": "funcName", + "logger.line": "lineno", + "logger.thread_name": "threadName", + "process.pid": "process", +} + +_BASE_CONFIG: dict[str, Any] = { + "version": 1, + "disable_existing_loggers": False, + "filters": { + "request_context": {"()": f"{_PKG}.context.RequestContextFilter"}, + }, + "formatters": { + "mp_json": { + "()": f"{_PKG}.formatters.DatadogStructuredFormatter", + "fmt_keys": FMT_KEYS, + }, + "mp_console": {"()": f"{_PKG}.formatters.ConsoleFormatter"}, + }, + "handlers": { + "mp_stderr": { + "class": "logging.StreamHandler", + "formatter": "mp_json", + "filters": ["request_context"], + "stream": "ext://sys.stderr", + }, + }, + "loggers": { + # gunicorn's (and uvicorn's) own loggers don't propagate, so they need + # the handler directly + "gunicorn.error": {"level": "INFO", "handlers": [HANDLER], "propagate": False}, + "uvicorn.error": {"level": "INFO", "handlers": [HANDLER], "propagate": False}, + }, + # Everything else (third-party libraries, captured warnings): + # WARNING and up (overridable with MP_LOG_LEVEL) + "root": {"level": "WARNING", "handlers": [HANDLER]}, +} + + +def _level_name(level: int | str) -> str: + if isinstance(level, int): + return logging.getLevelName(level) + name = level.strip().upper() + if name == "WARN": + name = "WARNING" + if not isinstance(logging.getLevelName(name), int): + raise ValueError(f"Unknown log level {level!r}") + return name + + +def build_logging_config( + *, + app_loggers: Iterable[str] = (), + access_loggers: Iterable[str] = ACCESS_LOGGERS, + extra_loggers: Mapping[str, Mapping[str, Any]] | None = None, + fmt: LogFormat | None = None, + level: int | str | None = None, + app_level: int | str | None = None, + debug: bool = False, +) -> dict[str, Any]: + """The `logging.config.dictConfig` dict for a service. + + Arguments left as None come from env vars. + + Args: + app_loggers : the service's own loggers (e.g. "mpcontribs"), logged at + `app_level` instead of the root level + access_loggers : loggers that write access logs (INFO, not propagated) + extra_loggers : extra per-logger config, applied last + fmt : "json" or "console" (`MP_LOG_FORMAT`, default "json") + level : root level (`MP_LOG_LEVEL`, default WARNING) + app_level : level for `app_loggers` (`MP_APP_LOG_LEVEL`, default + INFO, or DEBUG if `debug`) + debug : log the service's own loggers at DEBUG unless `app_level` + is set + + Returns: + dict for `logging.config.dictConfig` (a fresh copy, safe to modify) + """ + fmt = fmt or os.environ.get("MP_LOG_FORMAT") or "json" # type: ignore[assignment] + if fmt not in ("json", "console"): + raise ValueError(f"Unknown log format {fmt!r}, expected 'json' or 'console'") + level = _level_name(level or os.environ.get("MP_LOG_LEVEL") or "WARNING") + app_level = _level_name( + app_level + or os.environ.get("MP_APP_LOG_LEVEL") + or ("DEBUG" if debug else "INFO") + ) + + config = copy.deepcopy(_BASE_CONFIG) + config["handlers"][HANDLER]["formatter"] = f"mp_{fmt}" + config["root"]["level"] = level + loggers = config["loggers"] + for name in access_loggers: + loggers[name] = {"level": "INFO", "handlers": [HANDLER], "propagate": False} + for name in app_loggers: + loggers.setdefault(name, {})["level"] = app_level + for name, overrides in (extra_loggers or {}).items(): + loggers.setdefault(name, {}).update(copy.deepcopy(dict(overrides))) + return config + + +def configure_logging(**kwargs: Any) -> dict[str, Any]: + """Configure logging for the process. Safe to call more than once. + + Python warnings are captured into the `py.warnings` logger, so they are + formatted like everything else. See `build_logging_config` for the + arguments. + + Returns: + the applied config + """ + config = build_logging_config(**kwargs) + logging.config.dictConfig(config) + logging.captureWarnings(True) + return config + + +def gunicorn_logconfig_dict(**kwargs: Any) -> dict[str, Any]: + """Config for gunicorn's `logconfig_dict` setting (in `gunicorn.conf.py`). + + Applied by gunicorn in the master before forking, so gunicorn's own error + and access logs use the same handlers and formatters as the application. + Only imports this package, not the application. + """ + return build_logging_config(**kwargs) + + +__all__ = [ + "ACCESS_LOGGERS", + "FMT_KEYS", + "HANDLER", + "TORNADO_ACCESS_LOGGER", + "build_logging_config", + "configure_logging", + "gunicorn_logconfig_dict", +] diff --git a/mpcontribs-portal/mp_logging/context.py b/mpcontribs-portal/mp_logging/context.py new file mode 100644 index 0000000000..617e228810 --- /dev/null +++ b/mpcontribs-portal/mp_logging/context.py @@ -0,0 +1,93 @@ +"""Per-request logging context, independent of the web framework. + +Framework glue (`flask_ext`, `django_ext`, `asgi`) binds the current +request's fields with `bind_request_context` at the start of a request and +resets them at the end. `RequestContextFilter` adds them to every record +logged meanwhile, including records from libraries. + +The context is a `contextvars.ContextVar`, so it is per-greenlet under gevent +and per-task under asyncio. +""" + +from __future__ import annotations + +import logging +from collections.abc import Mapping +from contextvars import ContextVar, Token +from typing import Any + +from .formatters import path_fields, process_name + +_REQUEST_CONTEXT: ContextVar[Mapping[str, Any] | None] = ContextVar( + "mp_logging_request_context", default=None +) + + +def _header(headers: Mapping[str, str], name: str) -> str | None: + """A request header from a case-insensitive or lower-cased mapping.""" + value = headers.get(name) + if value is None: + value = headers.get(name.lower()) + return value or None + + +def request_fields( + method: str | None, path: str | None, headers: Mapping[str, str] +) -> dict[str, Any]: + """Fields describing a request, from its method, path and headers. + + The Kong gateway forwards the consumer in `X-Consumer-Id` and the request + id in `Kong-Request-Id` (or `X-Request-Id` if set upstream). + """ + fields = { + "http.method": method, + "http.url_details.path": path, + **path_fields(path), + "usr.id": _header(headers, "X-Consumer-Id"), + "http.request_id": _header(headers, "X-Request-Id") + or _header(headers, "Kong-Request-Id"), + } + return {k: v for k, v in fields.items() if v} + + +def bind_request_context(fields: Mapping[str, Any]) -> Token: + """Set the current request's fields. Pass the token to `reset_request_context`.""" + return _REQUEST_CONTEXT.set(dict(fields)) + + +def reset_request_context(token: Token | None) -> None: + """Restore the context from before `bind_request_context`. Never raises.""" + if token is None: + return + try: + _REQUEST_CONTEXT.reset(token) + except (ValueError, RuntimeError): + # Token from another context (e.g. teardown ran elsewhere): just clear + _REQUEST_CONTEXT.set(None) + + +def get_request_context() -> Mapping[str, Any] | None: + """The current request's fields, or None outside a request.""" + return _REQUEST_CONTEXT.get() + + +class RequestContextFilter(logging.Filter): + """Add the process name, and the current request's context, to every record. + + Attached to handlers (not loggers), so it also applies to records from + libraries that reach the service's handlers. Outside a request (startup, + background work) records only get the process name. Never drops a record. + """ + + def filter(self, record: logging.LogRecord) -> bool: + # The supervisor program (e.g. contribs-api, micro-worker), on every record + if not hasattr(record, "process.name") and (name := process_name()): + setattr(record, "process.name", name) + context = _REQUEST_CONTEXT.get() + if not context: + return True + for key, value in context.items(): + # Fields set explicitly at the call site win + if not hasattr(record, key): + setattr(record, key, value) + return True diff --git a/mpcontribs-portal/mp_logging/django_ext.py b/mpcontribs-portal/mp_logging/django_ext.py new file mode 100644 index 0000000000..5b7031c1e9 --- /dev/null +++ b/mpcontribs-portal/mp_logging/django_ext.py @@ -0,0 +1,37 @@ +"""Request logging context for Django apps. + +Add to `MIDDLEWARE` (first, so it covers the other middleware): + +```python +MIDDLEWARE = ("mp_logging.django_ext.RequestContextMiddleware", ...) +``` + +The access log itself comes from gunicorn (`gunicorn.access`). +""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import Any + +from .context import ( + bind_request_context, + request_fields, + reset_request_context, +) + + +class RequestContextMiddleware: + """Add the current request's fields to every record logged during it.""" + + def __init__(self, get_response: Callable[[Any], Any]) -> None: + self.get_response = get_response + + def __call__(self, request: Any) -> Any: + token = bind_request_context( + request_fields(request.method, request.path, request.headers) + ) + try: + return self.get_response(request) + finally: + reset_request_context(token) diff --git a/mpcontribs-portal/mp_logging/formatters.py b/mpcontribs-portal/mp_logging/formatters.py new file mode 100644 index 0000000000..dd2441b8fd --- /dev/null +++ b/mpcontribs-portal/mp_logging/formatters.py @@ -0,0 +1,350 @@ +"""Log formatters: Datadog-friendly JSON for deployments, readable text for dev. + +Adapted from `mp_ui.core.logging.formatters` (MP frontend), which is itself +based on the dagster pipelines' `common.logging.formatters`. + +Every attribute on a record that isn't a builtin `LogRecord` attribute is +emitted as a field. That includes `extra=` fields, the request context added +by `RequestContextFilter`, and the `dd.*` trace-correlation attributes ddtrace +adds when `DD_LOGS_INJECTION=true`. + +Access-log records are turned into Datadog's standard HTTP attributes: + +- gunicorn (`gunicorn.access`) and `mp_logging.asgi` (`asgi.access`) log each + request as `logger.info(format, atoms)`, with gunicorn's atom names +- tornado (`tornado.access`, e.g. the Jupyter kernel gateway) logs + `"%d %s %.2fms"` with (status, "METHOD URI (IP)", milliseconds) +""" + +from __future__ import annotations + +import datetime as dt +import json +import logging +import os +import re +from collections.abc import Mapping +from typing import Any + +# Attributes every LogRecord has. Anything else on a record is a structured field. +# These names also can't be used as `extra=` keys (logging raises KeyError). +LOG_RECORD_BUILTIN_ATTRS = frozenset( + { + "args", + "asctime", + "created", + "exc_info", + "exc_text", + "filename", + "funcName", + "levelname", + "levelno", + "lineno", + "module", + "msecs", + "message", + "msg", + "name", + "pathname", + "process", + "processName", + "relativeCreated", + "stack_info", + "thread", + "threadName", + "taskName", + } +) + +GUNICORN_ACCESS_LOGGER = "gunicorn.access" +ASGI_ACCESS_LOGGER = "asgi.access" +TORNADO_ACCESS_LOGGER = "tornado.access" +# Loggers whose records carry gunicorn-style access atoms +ATOM_ACCESS_LOGGERS = frozenset({GUNICORN_ACCESS_LOGGER, ASGI_ACCESS_LOGGER}) + +# Same patterns as the "Parse URL Path root" / "First two elements from URL +# Path" grok parsers of the previous Datadog pipeline, so facets keep their values: +# /materials/summary/mp-149 -> path_root "/materials", path_first_two "/materials/summary" +_PATH_ROOT = re.compile(r"([^/]*/){1}[^/]+") +_PATH_FIRST_TWO = re.compile(r"([^/]*/){2}[^/]+") + +# tornado.web.RequestHandler._request_summary(): "GET /api/kernels (127.0.0.1)" +_TORNADO_SUMMARY = re.compile(r"^(?P\S+) (?P\S+) \((?P[^)]*)\)$") + + +def path_fields(path: str | None) -> dict[str, str]: + """`http.url_details.path_root` / `path_first_two` for a URL path (if they exist).""" + fields: dict[str, str] = {} + if not path: + return fields + if match := _PATH_ROOT.match(path): + fields["http.url_details.path_root"] = match.group(0) + if match := _PATH_FIRST_TWO.match(path): + fields["http.url_details.path_first_two"] = match.group(0) + return fields + + +def status_category(status: int | None) -> str | None: + """Datadog status for an HTTP status code: info (1xx-3xx), warning (4xx), error (5xx).""" + if status is None: + return None + if status >= 500: + return "error" + if status >= 400: + return "warning" + return "info" + + +# `level` given to access records, by status category +_ACCESS_LEVELS = {"info": "INFO", "warning": "WARNING", "error": "ERROR"} + + +def process_name() -> str | None: + """The supervisor program this process runs under (e.g. contribs-api), if any.""" + return os.environ.get("SUPERVISOR_PROCESS_NAME") or None + + +def _atom(atoms: Mapping[str, Any], key: str) -> Any: + """An access-log atom, or None if it's missing or "-".""" + try: + value = atoms[key] + except KeyError: + return None + return None if value in ("-", "", None) else value + + +def _as_int(value: Any) -> int | None: + try: + return int(value) + except (TypeError, ValueError): + return None + + +def _client(atoms: Mapping[str, Any]) -> tuple[Any, int | None]: + """Client IP and port. gunicorn gives the port as an environ atom; + the ASGI middleware gives it as its own atom.""" + ip = _atom(atoms, "h") + port = _as_int(_atom(atoms, "{remote_port}e")) + return ip, port + + +def _http_fields( + *, + method: Any, + path: Any, + query: Any, + status: int | None, + duration_ns: int | None, +) -> dict[str, Any]: + category = status_category(status) + return { + # Access records take their level from the response status, so 4xx/5xx + # responses show as warnings/errors + "level": _ACCESS_LEVELS.get(category) if category else None, + "http.method": method, + # Full URL for a Datadog URL parser, which fills `http.url_details` + # (including `queryString` as an object of parameters) + "http.url": f"{path}?{query}" if query else path, + "http.url_details.path": path, + **path_fields(path), + "http.status_code": status, + "http.status_category": category, + # Datadog's standard duration attribute is in nanoseconds + "duration": duration_ns, + } + + +def _atom_access_fields(atoms: Mapping[str, Any]) -> dict[str, Any]: + method = _atom(atoms, "m") + path = _atom(atoms, "U") + status = _as_int(_atom(atoms, "s")) + duration_us = _as_int(_atom(atoms, "D")) + ip, port = _client(atoms) + return { + "message": f"{method} {path} {status}", + **_http_fields( + method=method, + path=path, + query=_atom(atoms, "q"), + status=status, + duration_ns=duration_us * 1000 if duration_us is not None else None, + ), + "http.version": _atom(atoms, "H"), + "http.referer": _atom(atoms, "f"), + "http.useragent": _atom(atoms, "a"), + "http.request_id": _atom(atoms, "{x-request-id}i") + or _atom(atoms, "{kong-request-id}i"), + "network.client.ip": ip, + "network.client.port": port, + "network.bytes_written": _as_int(_atom(atoms, "B")), + # Kong consumer, as forwarded by the gateway + "usr.id": _atom(atoms, "{x-consumer-id}i"), + "usr.groups": _atom(atoms, "{x-consumer-groups}o") + or _atom(atoms, "{x-consumer-groups}i") + or _atom(atoms, "{x-authenticated-groups}i"), + # Dash callback name (web only; absent for APIs) + "dash.callback": _atom(atoms, "{x-callback-name}o"), + "process.pid": _as_int(str(_atom(atoms, "p") or "").strip("<>")), + "process.name": process_name(), + } + + +def _tornado_access_fields(args: tuple[Any, ...]) -> dict[str, Any] | None: + status, summary, ms = args + match = _TORNADO_SUMMARY.match(str(summary)) + if match is None: + return None + method, uri, ip = match.group("method", "uri", "ip") + path, _, query = uri.partition("?") + status = _as_int(status) + try: + duration_ns = int(float(ms) * 1_000_000) + except (TypeError, ValueError): + duration_ns = None + return { + "message": f"{method} {path} {status}", + **_http_fields( + method=method, + path=path, + query=query, + status=status, + duration_ns=duration_ns, + ), + "network.client.ip": ip or None, + "process.name": process_name(), + } + + +def access_log_fields(record: logging.LogRecord) -> dict[str, Any] | None: + """Structured fields for an access-log record. + + Field names follow Datadog's standard attributes so its HTTP facets work + without remapping. + + Returns: + dict of fields (including "message"), or None if `record` isn't an + access-log record. + """ + fields: dict[str, Any] | None = None + if record.name in ATOM_ACCESS_LOGGERS and isinstance(record.args, Mapping): + fields = _atom_access_fields(record.args) + elif ( + record.name == TORNADO_ACCESS_LOGGER + and isinstance(record.args, tuple) + and len(record.args) == 3 + ): + fields = _tornado_access_fields(record.args) + if fields is None: + return None + return {k: v for k, v in fields.items() if v is not None} + + +# Attributes some frameworks add to their records, renamed to Datadog's +# standard attributes (e.g. Django's `django.request` / `django.server`) +_RENAMED_ATTRS = {"status_code": "http.status_code"} +# Attributes some frameworks add that aren't useful as fields (Django's +# request/server objects, whose useful parts are already in the request +# context; uvicorn's ANSI-colored copy of the message) +_DROPPED_ATTRS = frozenset({"request", "server_time", "color_message"}) + + +def extra_fields(record: logging.LogRecord) -> dict[str, Any]: + """Non-builtin attributes of a record (`extra=`, filters, ddtrace).""" + return { + _RENAMED_ATTRS.get(key, key): val + for key, val in record.__dict__.items() + if key not in LOG_RECORD_BUILTIN_ATTRS and key not in _DROPPED_ATTRS + } + + +class DatadogStructuredFormatter(logging.Formatter): + """One JSON object per record, using Datadog's standard attribute names.""" + + def __init__(self, *, fmt_keys: dict[str, str] | None = None) -> None: + """Create the formatter. + + Args: + fmt_keys (dict or None) : output key -> LogRecord attribute, or + "message" / "timestamp" for the formatted message and the + ISO-8601 UTC time. + """ + super().__init__() + self.fmt_keys = fmt_keys if fmt_keys is not None else {} + + def format(self, record: logging.LogRecord) -> str: + return json.dumps(self._prepare_log_dict(record), default=str) + + def _prepare_log_dict(self, record: logging.LogRecord) -> dict[str, Any]: + access = access_log_fields(record) + always_fields: dict[str, Any] = { + "message": ( + access.pop("message") + if access is not None + else record.getMessage().rstrip() + ), + "timestamp": dt.datetime.fromtimestamp( + record.created, tz=dt.timezone.utc + ).isoformat(), + } + + # `exc_info` is (None, None, None) when `logger.exception` is called + # outside an `except` block + if record.exc_info and record.exc_info[0] is not None: + exc_type, exc_value, _ = record.exc_info + # Datadog's standard error attributes + always_fields["error.kind"] = exc_type.__name__ + always_fields["error.message"] = str(exc_value) + always_fields["error.stack"] = self.formatException(record.exc_info) + + # Only present when a call site passes `stack_info=True` + if record.stack_info is not None: + always_fields["error.stack_info"] = self.formatStack(record.stack_info) + + message = { + key: ( + msg_val + if (msg_val := always_fields.pop(attr, None)) is not None + else getattr(record, attr, None) + ) + for key, attr in self.fmt_keys.items() + } + message.update(always_fields) + if access is not None: + message.update(access) + message.update(extra_fields(record)) + return message + + +class ConsoleFormatter(logging.Formatter): + """Readable single-line records for local development. + + `2026-10-08 12:00:00,000 WARNING [mpcontribs.api] Message key=value ...` + """ + + def __init__( + self, + fmt: str = "%(asctime)s %(levelname)-8s [%(name)s] %(message)s", + datefmt: str | None = None, + ) -> None: + super().__init__(fmt=fmt, datefmt=datefmt) + + def format(self, record: logging.LogRecord) -> str: + access = access_log_fields(record) + if access is not None: + # Format a copy, so other handlers still see the original record + record = logging.makeLogRecord(record.__dict__) + record.msg, record.args = access.pop("message"), None + if level := access.pop("level", None): + record.levelname = level + text = super().format(record) + + fields = {**(access or {}), **extra_fields(record)} + # ddtrace's injected ids are noise locally + fields = { + k: v for k, v in fields.items() if not k.startswith("dd.") and v is not None + } + if fields: + pairs = " ".join(f"{k}={v}" for k, v in fields.items()) + first, sep, rest = text.partition("\n") + text = f"{first} {pairs}{sep}{rest}" + return text diff --git a/mpcontribs-portal/settings.py b/mpcontribs-portal/settings.py index 373d89bd6a..0ca6603eb8 100644 --- a/mpcontribs-portal/settings.py +++ b/mpcontribs-portal/settings.py @@ -22,6 +22,7 @@ ] MIDDLEWARE = ( + "mp_logging.django_ext.RequestContextMiddleware", "django.contrib.sessions.middleware.SessionMiddleware", "django.middleware.common.CommonMiddleware", "django.middleware.clickjacking.XFrameOptionsMiddleware", @@ -78,3 +79,27 @@ APPEND_SLASH = False SECURE_REFERRER_POLICY = "unsafe-url" + + +def _logging_config(): + """Structured (Datadog JSON) logging, see `mp_logging`. + + Replaces Django's default handlers (its `django` / `django.server` loggers + would otherwise also write plain text). The access log comes from + gunicorn (gunicorn.conf.py). + """ + from mp_logging import build_logging_config + + return build_logging_config( + app_loggers=("mpcontribs",), + debug=DEBUG, + extra_loggers={ + # Django's own loggers: propagate to the root handler at INFO + "django": {"level": "INFO", "handlers": [], "propagate": True}, + "django.server": {"level": "INFO", "handlers": [], "propagate": True}, + }, + ) + + +LOGGING_CONFIG = "logging.config.dictConfig" +LOGGING = _logging_config() From 9a77b59d64106e1d738805a7a6ba86c602d90bd4 Mon Sep 17 00:00:00 2001 From: Tyler Mathis <35553152+tsmathis@users.noreply.github.com> Date: Wed, 14 Oct 2026 10:54:29 -0700 Subject: [PATCH 2/2] logging: better support for route decoration, add query param parsing, pass referer fields --- mpcontribs-api/mp_logging/__init__.py | 2 + mpcontribs-api/mp_logging/context.py | 18 +++- mpcontribs-api/mp_logging/flask_ext.py | 15 ++- mpcontribs-api/mp_logging/formatters.py | 101 +++++++++++++++--- .../mp_logging/__init__.py | 2 + .../mp_logging/context.py | 18 +++- .../mp_logging/formatters.py | 101 +++++++++++++++--- mpcontribs-portal/mp_logging/__init__.py | 2 + mpcontribs-portal/mp_logging/context.py | 18 +++- mpcontribs-portal/mp_logging/django_ext.py | 28 ++++- mpcontribs-portal/mp_logging/formatters.py | 101 +++++++++++++++--- 11 files changed, 343 insertions(+), 63 deletions(-) diff --git a/mpcontribs-api/mp_logging/__init__.py b/mpcontribs-api/mp_logging/__init__.py index 8facdbc939..d96b6da4c1 100644 --- a/mpcontribs-api/mp_logging/__init__.py +++ b/mpcontribs-api/mp_logging/__init__.py @@ -13,6 +13,7 @@ get_request_context, request_fields, reset_request_context, + update_request_context, ) from .formatters import ( LOG_RECORD_BUILTIN_ATTRS, @@ -36,4 +37,5 @@ "gunicorn_logconfig_dict", "request_fields", "reset_request_context", + "update_request_context", ] diff --git a/mpcontribs-api/mp_logging/context.py b/mpcontribs-api/mp_logging/context.py index 617e228810..dec10709c7 100644 --- a/mpcontribs-api/mp_logging/context.py +++ b/mpcontribs-api/mp_logging/context.py @@ -16,7 +16,7 @@ from contextvars import ContextVar, Token from typing import Any -from .formatters import path_fields, process_name +from .formatters import process_name _REQUEST_CONTEXT: ContextVar[Mapping[str, Any] | None] = ContextVar( "mp_logging_request_context", default=None @@ -32,7 +32,10 @@ def _header(headers: Mapping[str, str], name: str) -> str | None: def request_fields( - method: str | None, path: str | None, headers: Mapping[str, str] + method: str | None, + path: str | None, + headers: Mapping[str, str], + route: str | None = None, ) -> dict[str, Any]: """Fields describing a request, from its method, path and headers. @@ -42,7 +45,8 @@ def request_fields( fields = { "http.method": method, "http.url_details.path": path, - **path_fields(path), + # The matched route template, when known at bind time (Flask, Django) + "http.route": route, "usr.id": _header(headers, "X-Consumer-Id"), "http.request_id": _header(headers, "X-Request-Id") or _header(headers, "Kong-Request-Id"), @@ -66,6 +70,14 @@ def reset_request_context(token: Token | None) -> None: _REQUEST_CONTEXT.set(None) +def update_request_context(fields: Mapping[str, Any]) -> None: + """Add fields to the current request's context (e.g. the route, once matched).""" + context = _REQUEST_CONTEXT.get() + if context is None: + return + _REQUEST_CONTEXT.set({**context, **{k: v for k, v in fields.items() if v}}) + + def get_request_context() -> Mapping[str, Any] | None: """The current request's fields, or None outside a request.""" return _REQUEST_CONTEXT.get() diff --git a/mpcontribs-api/mp_logging/flask_ext.py b/mpcontribs-api/mp_logging/flask_ext.py index 4cd7f91cc3..6a6dcab7c2 100644 --- a/mpcontribs-api/mp_logging/flask_ext.py +++ b/mpcontribs-api/mp_logging/flask_ext.py @@ -11,20 +11,31 @@ request_fields, reset_request_context, ) +from .formatters import ENVIRON_ROUTE _G_KEY = "_mp_logging_token" +def _route() -> str | None: + """The matched URL rule, e.g. `/projects/` (None if unmatched).""" + rule = request.url_rule + return rule.rule if rule is not None else None + + def init_app(app: Flask) -> None: """Add the current request's fields to every record logged during it. - The access log itself comes from gunicorn (`gunicorn.access`). + The access log itself comes from gunicorn (`gunicorn.access`); the matched + route reaches it through the WSGI environ (`ENVIRON_ROUTE`). """ @app.before_request def _bind_logging_context() -> None: + route = _route() + if route: + request.environ[ENVIRON_ROUTE] = route token = bind_request_context( - request_fields(request.method, request.path, request.headers) + request_fields(request.method, request.path, request.headers, route=route) ) setattr(g, _G_KEY, token) diff --git a/mpcontribs-api/mp_logging/formatters.py b/mpcontribs-api/mp_logging/formatters.py index dd2441b8fd..b9c305f2f4 100644 --- a/mpcontribs-api/mp_logging/formatters.py +++ b/mpcontribs-api/mp_logging/formatters.py @@ -25,6 +25,7 @@ import re from collections.abc import Mapping from typing import Any +from urllib.parse import parse_qsl, urlsplit # Attributes every LogRecord has. Anything else on a record is a structured field. # These names also can't be used as `extra=` keys (logging raises KeyError). @@ -62,28 +63,91 @@ # Loggers whose records carry gunicorn-style access atoms ATOM_ACCESS_LOGGERS = frozenset({GUNICORN_ACCESS_LOGGER, ASGI_ACCESS_LOGGER}) -# Same patterns as the "Parse URL Path root" / "First two elements from URL -# Path" grok parsers of the previous Datadog pipeline, so facets keep their values: -# /materials/summary/mp-149 -> path_root "/materials", path_first_two "/materials/summary" -_PATH_ROOT = re.compile(r"([^/]*/){1}[^/]+") -_PATH_FIRST_TWO = re.compile(r"([^/]*/){2}[^/]+") +# The matched route template (e.g. `/materials/summary/{material_id}/`), set by +# the framework glue: an atom of the ASGI middleware, or a WSGI environ entry +# for gunicorn (environ entries are lowercased `{}e` atoms) +ROUTE_ATOM = "http.route" +ENVIRON_ROUTE = "mp_logging.http.route" +_ENVIRON_ROUTE_ATOM = "{mp_logging.http.route}e" + +# Caps on URL-derived fields: query keys are client-controlled and every one +# becomes an attribute (and potential facet) in Datadog +MAX_URL_LENGTH = 1024 +MAX_QUERY_PARAMS = 20 +MAX_QUERY_KEY_LENGTH = 64 +MAX_QUERY_VALUE_LENGTH = 256 # tornado.web.RequestHandler._request_summary(): "GET /api/kernels (127.0.0.1)" _TORNADO_SUMMARY = re.compile(r"^(?P\S+) (?P\S+) \((?P[^)]*)\)$") -def path_fields(path: str | None) -> dict[str, str]: - """`http.url_details.path_root` / `path_first_two` for a URL path (if they exist).""" - fields: dict[str, str] = {} +def _cap(value: str, limit: int) -> tuple[str, bool]: + return (value[:limit], True) if len(value) > limit else (value, False) + + +def url_field(path: str | None, query: str | None) -> str | None: + """`http.url`: path and query string, capped at `MAX_URL_LENGTH`.""" if not path: - return fields - if match := _PATH_ROOT.match(path): - fields["http.url_details.path_root"] = match.group(0) - if match := _PATH_FIRST_TWO.match(path): - fields["http.url_details.path_first_two"] = match.group(0) + return None + url = f"{path}?{query}" if query else path + return url[:MAX_URL_LENGTH] + + +def query_fields(query: str | None) -> dict[str, Any]: + """`http.url_details.queryString`: the decoded query parameters. + + Values are URL-decoded (`%2C` -> `,`); a repeated key gives a list. At + most `MAX_QUERY_PARAMS` parameters, keys capped at `MAX_QUERY_KEY_LENGTH` + and values at `MAX_QUERY_VALUE_LENGTH` characters; if anything was cut, + `http.url_details.queryString_truncated` is true. + """ + if not query: + return {} + try: + pairs = parse_qsl(query, keep_blank_values=True) + except ValueError: + return {} + params: dict[str, Any] = {} + truncated = False + for key, value in pairs: + key, cut_key = _cap(key, MAX_QUERY_KEY_LENGTH) + value, cut_value = _cap(value, MAX_QUERY_VALUE_LENGTH) + truncated = truncated or cut_key or cut_value + if key in params: + existing = params[key] + params[key] = ( + [*existing, value] if isinstance(existing, list) else [existing, value] + ) + elif len(params) < MAX_QUERY_PARAMS: + params[key] = value + else: + truncated = True + fields: dict[str, Any] = {} + if params: + fields["http.url_details.queryString"] = params + if truncated: + fields["http.url_details.queryString_truncated"] = True return fields +def referer_fields(referer: str | None) -> dict[str, Any]: + """`http.referer_details.{scheme,host,port,path}` from the Referer header.""" + if not referer: + return {} + try: + parts = urlsplit(referer) + port = parts.port + except ValueError: + return {} + fields = { + "http.referer_details.scheme": parts.scheme or None, + "http.referer_details.host": parts.hostname or None, + "http.referer_details.port": port, + "http.referer_details.path": parts.path[:MAX_URL_LENGTH] or None, + } + return {k: v for k, v in fields.items() if v is not None} + + def status_category(status: int | None) -> str | None: """Datadog status for an HTTP status code: info (1xx-3xx), warning (4xx), error (5xx).""" if status is None: @@ -133,6 +197,7 @@ def _http_fields( method: Any, path: Any, query: Any, + route: Any, status: int | None, duration_ns: int | None, ) -> dict[str, Any]: @@ -142,11 +207,10 @@ def _http_fields( # responses show as warnings/errors "level": _ACCESS_LEVELS.get(category) if category else None, "http.method": method, - # Full URL for a Datadog URL parser, which fills `http.url_details` - # (including `queryString` as an object of parameters) - "http.url": f"{path}?{query}" if query else path, + "http.url": url_field(path, query), "http.url_details.path": path, - **path_fields(path), + **query_fields(query), + "http.route": route, "http.status_code": status, "http.status_category": category, # Datadog's standard duration attribute is in nanoseconds @@ -166,11 +230,13 @@ def _atom_access_fields(atoms: Mapping[str, Any]) -> dict[str, Any]: method=method, path=path, query=_atom(atoms, "q"), + route=_atom(atoms, ROUTE_ATOM) or _atom(atoms, _ENVIRON_ROUTE_ATOM), status=status, duration_ns=duration_us * 1000 if duration_us is not None else None, ), "http.version": _atom(atoms, "H"), "http.referer": _atom(atoms, "f"), + **referer_fields(_atom(atoms, "f")), "http.useragent": _atom(atoms, "a"), "http.request_id": _atom(atoms, "{x-request-id}i") or _atom(atoms, "{kong-request-id}i"), @@ -207,6 +273,7 @@ def _tornado_access_fields(args: tuple[Any, ...]) -> dict[str, Any] | None: method=method, path=path, query=query, + route=None, status=status, duration_ns=duration_ns, ), diff --git a/mpcontribs-kernel-gateway/mp_logging/__init__.py b/mpcontribs-kernel-gateway/mp_logging/__init__.py index 8facdbc939..d96b6da4c1 100644 --- a/mpcontribs-kernel-gateway/mp_logging/__init__.py +++ b/mpcontribs-kernel-gateway/mp_logging/__init__.py @@ -13,6 +13,7 @@ get_request_context, request_fields, reset_request_context, + update_request_context, ) from .formatters import ( LOG_RECORD_BUILTIN_ATTRS, @@ -36,4 +37,5 @@ "gunicorn_logconfig_dict", "request_fields", "reset_request_context", + "update_request_context", ] diff --git a/mpcontribs-kernel-gateway/mp_logging/context.py b/mpcontribs-kernel-gateway/mp_logging/context.py index 617e228810..dec10709c7 100644 --- a/mpcontribs-kernel-gateway/mp_logging/context.py +++ b/mpcontribs-kernel-gateway/mp_logging/context.py @@ -16,7 +16,7 @@ from contextvars import ContextVar, Token from typing import Any -from .formatters import path_fields, process_name +from .formatters import process_name _REQUEST_CONTEXT: ContextVar[Mapping[str, Any] | None] = ContextVar( "mp_logging_request_context", default=None @@ -32,7 +32,10 @@ def _header(headers: Mapping[str, str], name: str) -> str | None: def request_fields( - method: str | None, path: str | None, headers: Mapping[str, str] + method: str | None, + path: str | None, + headers: Mapping[str, str], + route: str | None = None, ) -> dict[str, Any]: """Fields describing a request, from its method, path and headers. @@ -42,7 +45,8 @@ def request_fields( fields = { "http.method": method, "http.url_details.path": path, - **path_fields(path), + # The matched route template, when known at bind time (Flask, Django) + "http.route": route, "usr.id": _header(headers, "X-Consumer-Id"), "http.request_id": _header(headers, "X-Request-Id") or _header(headers, "Kong-Request-Id"), @@ -66,6 +70,14 @@ def reset_request_context(token: Token | None) -> None: _REQUEST_CONTEXT.set(None) +def update_request_context(fields: Mapping[str, Any]) -> None: + """Add fields to the current request's context (e.g. the route, once matched).""" + context = _REQUEST_CONTEXT.get() + if context is None: + return + _REQUEST_CONTEXT.set({**context, **{k: v for k, v in fields.items() if v}}) + + def get_request_context() -> Mapping[str, Any] | None: """The current request's fields, or None outside a request.""" return _REQUEST_CONTEXT.get() diff --git a/mpcontribs-kernel-gateway/mp_logging/formatters.py b/mpcontribs-kernel-gateway/mp_logging/formatters.py index dd2441b8fd..b9c305f2f4 100644 --- a/mpcontribs-kernel-gateway/mp_logging/formatters.py +++ b/mpcontribs-kernel-gateway/mp_logging/formatters.py @@ -25,6 +25,7 @@ import re from collections.abc import Mapping from typing import Any +from urllib.parse import parse_qsl, urlsplit # Attributes every LogRecord has. Anything else on a record is a structured field. # These names also can't be used as `extra=` keys (logging raises KeyError). @@ -62,28 +63,91 @@ # Loggers whose records carry gunicorn-style access atoms ATOM_ACCESS_LOGGERS = frozenset({GUNICORN_ACCESS_LOGGER, ASGI_ACCESS_LOGGER}) -# Same patterns as the "Parse URL Path root" / "First two elements from URL -# Path" grok parsers of the previous Datadog pipeline, so facets keep their values: -# /materials/summary/mp-149 -> path_root "/materials", path_first_two "/materials/summary" -_PATH_ROOT = re.compile(r"([^/]*/){1}[^/]+") -_PATH_FIRST_TWO = re.compile(r"([^/]*/){2}[^/]+") +# The matched route template (e.g. `/materials/summary/{material_id}/`), set by +# the framework glue: an atom of the ASGI middleware, or a WSGI environ entry +# for gunicorn (environ entries are lowercased `{}e` atoms) +ROUTE_ATOM = "http.route" +ENVIRON_ROUTE = "mp_logging.http.route" +_ENVIRON_ROUTE_ATOM = "{mp_logging.http.route}e" + +# Caps on URL-derived fields: query keys are client-controlled and every one +# becomes an attribute (and potential facet) in Datadog +MAX_URL_LENGTH = 1024 +MAX_QUERY_PARAMS = 20 +MAX_QUERY_KEY_LENGTH = 64 +MAX_QUERY_VALUE_LENGTH = 256 # tornado.web.RequestHandler._request_summary(): "GET /api/kernels (127.0.0.1)" _TORNADO_SUMMARY = re.compile(r"^(?P\S+) (?P\S+) \((?P[^)]*)\)$") -def path_fields(path: str | None) -> dict[str, str]: - """`http.url_details.path_root` / `path_first_two` for a URL path (if they exist).""" - fields: dict[str, str] = {} +def _cap(value: str, limit: int) -> tuple[str, bool]: + return (value[:limit], True) if len(value) > limit else (value, False) + + +def url_field(path: str | None, query: str | None) -> str | None: + """`http.url`: path and query string, capped at `MAX_URL_LENGTH`.""" if not path: - return fields - if match := _PATH_ROOT.match(path): - fields["http.url_details.path_root"] = match.group(0) - if match := _PATH_FIRST_TWO.match(path): - fields["http.url_details.path_first_two"] = match.group(0) + return None + url = f"{path}?{query}" if query else path + return url[:MAX_URL_LENGTH] + + +def query_fields(query: str | None) -> dict[str, Any]: + """`http.url_details.queryString`: the decoded query parameters. + + Values are URL-decoded (`%2C` -> `,`); a repeated key gives a list. At + most `MAX_QUERY_PARAMS` parameters, keys capped at `MAX_QUERY_KEY_LENGTH` + and values at `MAX_QUERY_VALUE_LENGTH` characters; if anything was cut, + `http.url_details.queryString_truncated` is true. + """ + if not query: + return {} + try: + pairs = parse_qsl(query, keep_blank_values=True) + except ValueError: + return {} + params: dict[str, Any] = {} + truncated = False + for key, value in pairs: + key, cut_key = _cap(key, MAX_QUERY_KEY_LENGTH) + value, cut_value = _cap(value, MAX_QUERY_VALUE_LENGTH) + truncated = truncated or cut_key or cut_value + if key in params: + existing = params[key] + params[key] = ( + [*existing, value] if isinstance(existing, list) else [existing, value] + ) + elif len(params) < MAX_QUERY_PARAMS: + params[key] = value + else: + truncated = True + fields: dict[str, Any] = {} + if params: + fields["http.url_details.queryString"] = params + if truncated: + fields["http.url_details.queryString_truncated"] = True return fields +def referer_fields(referer: str | None) -> dict[str, Any]: + """`http.referer_details.{scheme,host,port,path}` from the Referer header.""" + if not referer: + return {} + try: + parts = urlsplit(referer) + port = parts.port + except ValueError: + return {} + fields = { + "http.referer_details.scheme": parts.scheme or None, + "http.referer_details.host": parts.hostname or None, + "http.referer_details.port": port, + "http.referer_details.path": parts.path[:MAX_URL_LENGTH] or None, + } + return {k: v for k, v in fields.items() if v is not None} + + def status_category(status: int | None) -> str | None: """Datadog status for an HTTP status code: info (1xx-3xx), warning (4xx), error (5xx).""" if status is None: @@ -133,6 +197,7 @@ def _http_fields( method: Any, path: Any, query: Any, + route: Any, status: int | None, duration_ns: int | None, ) -> dict[str, Any]: @@ -142,11 +207,10 @@ def _http_fields( # responses show as warnings/errors "level": _ACCESS_LEVELS.get(category) if category else None, "http.method": method, - # Full URL for a Datadog URL parser, which fills `http.url_details` - # (including `queryString` as an object of parameters) - "http.url": f"{path}?{query}" if query else path, + "http.url": url_field(path, query), "http.url_details.path": path, - **path_fields(path), + **query_fields(query), + "http.route": route, "http.status_code": status, "http.status_category": category, # Datadog's standard duration attribute is in nanoseconds @@ -166,11 +230,13 @@ def _atom_access_fields(atoms: Mapping[str, Any]) -> dict[str, Any]: method=method, path=path, query=_atom(atoms, "q"), + route=_atom(atoms, ROUTE_ATOM) or _atom(atoms, _ENVIRON_ROUTE_ATOM), status=status, duration_ns=duration_us * 1000 if duration_us is not None else None, ), "http.version": _atom(atoms, "H"), "http.referer": _atom(atoms, "f"), + **referer_fields(_atom(atoms, "f")), "http.useragent": _atom(atoms, "a"), "http.request_id": _atom(atoms, "{x-request-id}i") or _atom(atoms, "{kong-request-id}i"), @@ -207,6 +273,7 @@ def _tornado_access_fields(args: tuple[Any, ...]) -> dict[str, Any] | None: method=method, path=path, query=query, + route=None, status=status, duration_ns=duration_ns, ), diff --git a/mpcontribs-portal/mp_logging/__init__.py b/mpcontribs-portal/mp_logging/__init__.py index 8facdbc939..d96b6da4c1 100644 --- a/mpcontribs-portal/mp_logging/__init__.py +++ b/mpcontribs-portal/mp_logging/__init__.py @@ -13,6 +13,7 @@ get_request_context, request_fields, reset_request_context, + update_request_context, ) from .formatters import ( LOG_RECORD_BUILTIN_ATTRS, @@ -36,4 +37,5 @@ "gunicorn_logconfig_dict", "request_fields", "reset_request_context", + "update_request_context", ] diff --git a/mpcontribs-portal/mp_logging/context.py b/mpcontribs-portal/mp_logging/context.py index 617e228810..dec10709c7 100644 --- a/mpcontribs-portal/mp_logging/context.py +++ b/mpcontribs-portal/mp_logging/context.py @@ -16,7 +16,7 @@ from contextvars import ContextVar, Token from typing import Any -from .formatters import path_fields, process_name +from .formatters import process_name _REQUEST_CONTEXT: ContextVar[Mapping[str, Any] | None] = ContextVar( "mp_logging_request_context", default=None @@ -32,7 +32,10 @@ def _header(headers: Mapping[str, str], name: str) -> str | None: def request_fields( - method: str | None, path: str | None, headers: Mapping[str, str] + method: str | None, + path: str | None, + headers: Mapping[str, str], + route: str | None = None, ) -> dict[str, Any]: """Fields describing a request, from its method, path and headers. @@ -42,7 +45,8 @@ def request_fields( fields = { "http.method": method, "http.url_details.path": path, - **path_fields(path), + # The matched route template, when known at bind time (Flask, Django) + "http.route": route, "usr.id": _header(headers, "X-Consumer-Id"), "http.request_id": _header(headers, "X-Request-Id") or _header(headers, "Kong-Request-Id"), @@ -66,6 +70,14 @@ def reset_request_context(token: Token | None) -> None: _REQUEST_CONTEXT.set(None) +def update_request_context(fields: Mapping[str, Any]) -> None: + """Add fields to the current request's context (e.g. the route, once matched).""" + context = _REQUEST_CONTEXT.get() + if context is None: + return + _REQUEST_CONTEXT.set({**context, **{k: v for k, v in fields.items() if v}}) + + def get_request_context() -> Mapping[str, Any] | None: """The current request's fields, or None outside a request.""" return _REQUEST_CONTEXT.get() diff --git a/mpcontribs-portal/mp_logging/django_ext.py b/mpcontribs-portal/mp_logging/django_ext.py index 5b7031c1e9..78da593a40 100644 --- a/mpcontribs-portal/mp_logging/django_ext.py +++ b/mpcontribs-portal/mp_logging/django_ext.py @@ -6,7 +6,8 @@ MIDDLEWARE = ("mp_logging.django_ext.RequestContextMiddleware", ...) ``` -The access log itself comes from gunicorn (`gunicorn.access`). +The access log itself comes from gunicorn (`gunicorn.access`); the matched +route reaches it through the WSGI environ (`ENVIRON_ROUTE`). """ from __future__ import annotations @@ -18,7 +19,22 @@ bind_request_context, request_fields, reset_request_context, + update_request_context, ) +from .formatters import ENVIRON_ROUTE + + +def _route(request: Any) -> str | None: + """The matched URL pattern, e.g. `projects//` or `^search/?$`. + + Only known after URL resolution, i.e. from `process_view` on. + """ + match = getattr(request, "resolver_match", None) + route = getattr(match, "route", None) + if not route: + return None + # Django's patterns have no leading slash; match the request path's form + return route if route.startswith(("/", "^")) else f"/{route}" class RequestContextMiddleware: @@ -35,3 +51,13 @@ def __call__(self, request: Any) -> Any: return self.get_response(request) finally: reset_request_context(token) + + def process_view( + self, request: Any, view_func: Any, view_args: Any, view_kwargs: Any + ) -> None: + """Add the matched route, once Django has resolved the URL.""" + route = _route(request) + if route: + update_request_context({"http.route": route}) + # request.META is the WSGI environ that gunicorn's access log reads + request.META[ENVIRON_ROUTE] = route diff --git a/mpcontribs-portal/mp_logging/formatters.py b/mpcontribs-portal/mp_logging/formatters.py index dd2441b8fd..b9c305f2f4 100644 --- a/mpcontribs-portal/mp_logging/formatters.py +++ b/mpcontribs-portal/mp_logging/formatters.py @@ -25,6 +25,7 @@ import re from collections.abc import Mapping from typing import Any +from urllib.parse import parse_qsl, urlsplit # Attributes every LogRecord has. Anything else on a record is a structured field. # These names also can't be used as `extra=` keys (logging raises KeyError). @@ -62,28 +63,91 @@ # Loggers whose records carry gunicorn-style access atoms ATOM_ACCESS_LOGGERS = frozenset({GUNICORN_ACCESS_LOGGER, ASGI_ACCESS_LOGGER}) -# Same patterns as the "Parse URL Path root" / "First two elements from URL -# Path" grok parsers of the previous Datadog pipeline, so facets keep their values: -# /materials/summary/mp-149 -> path_root "/materials", path_first_two "/materials/summary" -_PATH_ROOT = re.compile(r"([^/]*/){1}[^/]+") -_PATH_FIRST_TWO = re.compile(r"([^/]*/){2}[^/]+") +# The matched route template (e.g. `/materials/summary/{material_id}/`), set by +# the framework glue: an atom of the ASGI middleware, or a WSGI environ entry +# for gunicorn (environ entries are lowercased `{}e` atoms) +ROUTE_ATOM = "http.route" +ENVIRON_ROUTE = "mp_logging.http.route" +_ENVIRON_ROUTE_ATOM = "{mp_logging.http.route}e" + +# Caps on URL-derived fields: query keys are client-controlled and every one +# becomes an attribute (and potential facet) in Datadog +MAX_URL_LENGTH = 1024 +MAX_QUERY_PARAMS = 20 +MAX_QUERY_KEY_LENGTH = 64 +MAX_QUERY_VALUE_LENGTH = 256 # tornado.web.RequestHandler._request_summary(): "GET /api/kernels (127.0.0.1)" _TORNADO_SUMMARY = re.compile(r"^(?P\S+) (?P\S+) \((?P[^)]*)\)$") -def path_fields(path: str | None) -> dict[str, str]: - """`http.url_details.path_root` / `path_first_two` for a URL path (if they exist).""" - fields: dict[str, str] = {} +def _cap(value: str, limit: int) -> tuple[str, bool]: + return (value[:limit], True) if len(value) > limit else (value, False) + + +def url_field(path: str | None, query: str | None) -> str | None: + """`http.url`: path and query string, capped at `MAX_URL_LENGTH`.""" if not path: - return fields - if match := _PATH_ROOT.match(path): - fields["http.url_details.path_root"] = match.group(0) - if match := _PATH_FIRST_TWO.match(path): - fields["http.url_details.path_first_two"] = match.group(0) + return None + url = f"{path}?{query}" if query else path + return url[:MAX_URL_LENGTH] + + +def query_fields(query: str | None) -> dict[str, Any]: + """`http.url_details.queryString`: the decoded query parameters. + + Values are URL-decoded (`%2C` -> `,`); a repeated key gives a list. At + most `MAX_QUERY_PARAMS` parameters, keys capped at `MAX_QUERY_KEY_LENGTH` + and values at `MAX_QUERY_VALUE_LENGTH` characters; if anything was cut, + `http.url_details.queryString_truncated` is true. + """ + if not query: + return {} + try: + pairs = parse_qsl(query, keep_blank_values=True) + except ValueError: + return {} + params: dict[str, Any] = {} + truncated = False + for key, value in pairs: + key, cut_key = _cap(key, MAX_QUERY_KEY_LENGTH) + value, cut_value = _cap(value, MAX_QUERY_VALUE_LENGTH) + truncated = truncated or cut_key or cut_value + if key in params: + existing = params[key] + params[key] = ( + [*existing, value] if isinstance(existing, list) else [existing, value] + ) + elif len(params) < MAX_QUERY_PARAMS: + params[key] = value + else: + truncated = True + fields: dict[str, Any] = {} + if params: + fields["http.url_details.queryString"] = params + if truncated: + fields["http.url_details.queryString_truncated"] = True return fields +def referer_fields(referer: str | None) -> dict[str, Any]: + """`http.referer_details.{scheme,host,port,path}` from the Referer header.""" + if not referer: + return {} + try: + parts = urlsplit(referer) + port = parts.port + except ValueError: + return {} + fields = { + "http.referer_details.scheme": parts.scheme or None, + "http.referer_details.host": parts.hostname or None, + "http.referer_details.port": port, + "http.referer_details.path": parts.path[:MAX_URL_LENGTH] or None, + } + return {k: v for k, v in fields.items() if v is not None} + + def status_category(status: int | None) -> str | None: """Datadog status for an HTTP status code: info (1xx-3xx), warning (4xx), error (5xx).""" if status is None: @@ -133,6 +197,7 @@ def _http_fields( method: Any, path: Any, query: Any, + route: Any, status: int | None, duration_ns: int | None, ) -> dict[str, Any]: @@ -142,11 +207,10 @@ def _http_fields( # responses show as warnings/errors "level": _ACCESS_LEVELS.get(category) if category else None, "http.method": method, - # Full URL for a Datadog URL parser, which fills `http.url_details` - # (including `queryString` as an object of parameters) - "http.url": f"{path}?{query}" if query else path, + "http.url": url_field(path, query), "http.url_details.path": path, - **path_fields(path), + **query_fields(query), + "http.route": route, "http.status_code": status, "http.status_category": category, # Datadog's standard duration attribute is in nanoseconds @@ -166,11 +230,13 @@ def _atom_access_fields(atoms: Mapping[str, Any]) -> dict[str, Any]: method=method, path=path, query=_atom(atoms, "q"), + route=_atom(atoms, ROUTE_ATOM) or _atom(atoms, _ENVIRON_ROUTE_ATOM), status=status, duration_ns=duration_us * 1000 if duration_us is not None else None, ), "http.version": _atom(atoms, "H"), "http.referer": _atom(atoms, "f"), + **referer_fields(_atom(atoms, "f")), "http.useragent": _atom(atoms, "a"), "http.request_id": _atom(atoms, "{x-request-id}i") or _atom(atoms, "{kong-request-id}i"), @@ -207,6 +273,7 @@ def _tornado_access_fields(args: tuple[Any, ...]) -> dict[str, Any] | None: method=method, path=path, query=query, + route=None, status=status, duration_ns=duration_ns, ),