feat(logging): more structure

tests/unit/test_logging.py

@@ -1,9 +1,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
-import json
 import logging
 import logging.config
-import threading
 import uuid
 
 from unittest import mock
@@ -15,28 +13,69 @@
 from warehouse import logging as wlogging
 
 
-class TestStructlogFormatter:
- def test_warehouse_logger_no_renderer(self):
- formatter = wlogging.StructlogFormatter()
- record = logging.LogRecord(
- "warehouse.request", logging.INFO, None, None, "the message", None, None
+class TestGunicornAccessLogParsing:
+ def test_parse_gunicorn_access_log_success(self):
+ access_log_line = (
+ "192.168.1.1 - - "
+ '[11/Aug/2025:21:01:13 +0000] "GET /pypi/b5ee/json HTTP/1.1" 404 24 '
+ '"-" "dependabot-core/0.325.1 excon/1.2.5 ruby/3.4.5 (x86_64-linux)"'
  )
 
- assert formatter.format(record) == "the message"
-
- def test_non_warehouse_logger_renders(self):
- formatter = wlogging.StructlogFormatter()
- record = logging.LogRecord(
- "another.logger", logging.INFO, None, None, "the message", None, None
+ event_dict = {"logger": "gunicorn.access", "event": access_log_line}
+
+ result = wlogging._parse_gunicorn_access_log(None, None, event_dict)
+
+ assert result["event"] == "http_request"
+ assert result["remote_addr"] == "192.168.1.1"
+ assert result["user"] is None
+ assert result["timestamp"] == "11/Aug/2025:21:01:13 +0000"
+ assert result["request"] == "GET /pypi/b5ee/json HTTP/1.1"
+ assert result["method"] == "GET"
+ assert result["path"] == "/pypi/b5ee/json"
+ assert result["protocol"] == "HTTP/1.1"
+ assert result["status"] == 404
+ assert result["size"] == 24
+ assert result["referrer"] is None
+ assert "dependabot-core" in result["user_agent"]
+
+ def test_parse_gunicorn_access_log_with_referrer(self):
+ access_log_line = (
+ "192.168.1.1 - - "
+ '[12/Aug/2025:10:30:45 +0000] "POST /simple/upload HTTP/1.1" 200 500 '
+ '"https://pypi.org/project/test/" "Mozilla/5.0 (compatible; test)"'
  )
 
- assert json.loads(formatter.format(record)) == {
- "logger": "another.logger",
- "level": "INFO",
- "event": "the message",
- "thread": threading.get_ident(),
+ event_dict = {"logger": "gunicorn.access", "event": access_log_line}
+
+ result = wlogging._parse_gunicorn_access_log(None, None, event_dict)
+
+ assert result["remote_addr"] == "192.168.1.1"
+ assert result["method"] == "POST"
+ assert result["path"] == "/simple/upload"
+ assert result["status"] == 200
+ assert result["size"] == 500
+ assert result["referrer"] == "https://pypi.org/project/test/"
+ assert result["user_agent"] == "Mozilla/5.0 (compatible; test)"
+
+ def test_parse_gunicorn_access_log_unparsable(self):
+ event_dict = {
+ "logger": "gunicorn.access",
+ "event": "this is not a valid access log format",
  }
 
+ result = wlogging._parse_gunicorn_access_log(None, None, event_dict)
+
+ # Should return unchanged if unparsable
+ assert result["event"] == "this is not a valid access log format"
+
+ def test_parse_gunicorn_access_log_non_access_log(self):
+ event_dict = {"logger": "some.other.logger", "event": "Some message"}
+
+ result = wlogging._parse_gunicorn_access_log(None, None, event_dict)
+
+ # Should return unchanged for non-access logs
+ assert result == event_dict
+
 
 def test_create_id(monkeypatch):
  uuid4 = pretend.call_recorder(lambda: "a fake uuid")
@@ -82,13 +121,17 @@ def test_includeme(monkeypatch, settings, expected_level):
  "version": 1,
  "disable_existing_loggers": False,
  "formatters": {
- "structlog": {"()": "warehouse.logging.StructlogFormatter"}
+ "structlog_formatter": {
+ "()": structlog.stdlib.ProcessorFormatter,
+ "processor": mock.ANY,
+ "foreign_pre_chain": mock.ANY,
+ }
  },
  "handlers": {
  "primary": {
  "class": "logging.StreamHandler",
  "stream": "ext://sys.stdout",
- "formatter": "structlog",
+ "formatter": "structlog_formatter",
  },
  },
  "loggers": {
@@ -119,10 +162,12 @@ def test_includeme(monkeypatch, settings, expected_level):
  structlog.stdlib.filter_by_level,
  structlog.stdlib.add_logger_name,
  structlog.stdlib.add_log_level,
- mock.ANY,
- mock.ANY,
+ mock.ANY, # PositionalArgumentsFormatter
+ mock.ANY, # TimeStamper
+ mock.ANY, # StackInfoRenderer
  structlog.processors.format_exc_info,
- wlogging.RENDERER,
+ mock.ANY, # _add_datadog_context
+ structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
  ],
  logger_factory=mock.ANY,
  wrapper_class=structlog.stdlib.BoundLogger,
@@ -135,6 +180,10 @@ def test_includeme(monkeypatch, settings, expected_level):
  )
  assert isinstance(
  configure.calls[0].kwargs["processors"][4],
+ structlog.processors.TimeStamper,
+ )
+ assert isinstance(
+ configure.calls[0].kwargs["processors"][5],
  structlog.processors.StackInfoRenderer,
  )
  assert isinstance(

warehouse/logging.py

@@ -1,59 +1,158 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import logging.config
-import threading
+import os
+import re
 import uuid
 
 import structlog
 
 request_logger = structlog.get_logger("warehouse.request")
 
-RENDERER = structlog.processors.JSONRenderer()
+# Determine if we're in development mode
+DEV_MODE = os.environ.get("WAREHOUSE_ENV") == "development"
 
-
-class StructlogFormatter(logging.Formatter):
- def format(self, record):
- # TODO: Figure out a better way of handling this besides just looking
- # at the logger name, ideally this would have some way to
- # really differentiate between log items which were logged by
- # structlog and which were not.
- if not record.name.startswith("warehouse."):
- # TODO: Is there a better way to handle this? Maybe we can figure
- # out a way to pass this through the structlog processors
- # instead of manually duplicating the side effects here?
- event_dict = {
- "logger": record.name,
- "level": record.levelname,
- "event": record.msg,
- "thread": threading.get_ident(),
- }
- record.msg = RENDERER(None, record.levelname, event_dict)
-
- return super().format(record)
+# Choose renderer based on environment
+RENDERER: structlog.dev.ConsoleRenderer | structlog.processors.JSONRenderer
+if DEV_MODE:
+ RENDERER = structlog.dev.ConsoleRenderer(colors=True)
+else:
+ RENDERER = structlog.processors.JSONRenderer()
 
 
 def _create_id(request):
  return str(uuid.uuid4())
 
 
+def _add_datadog_context(logger, method_name, event_dict):
+ """Add Datadog trace context if available"""
+ try:
+ import ddtrace
+
+ span = ddtrace.tracer.current_span()
+ if span:
+ event_dict["dd.trace_id"] = str(span.trace_id)
+ event_dict["dd.span_id"] = str(span.span_id)
+ event_dict["dd.service"] = span.service
+ # deployment metadata
+ event_dict["dd.env"] = os.environ.get("DD_ENV", "development")
+ event_dict["dd.version"] = os.environ.get("DD_VERSION", "unknown")
+ except (ImportError, AttributeError):
+ pass
+ return event_dict
+
+
+def _parse_gunicorn_access_log(logger, method_name, event_dict):
+ """Parse Gunicorn logs into structlog ((only access logs)."""
+ if event_dict.get("logger") != "gunicorn.access":
+ return event_dict
+
+ message = event_dict.get("event", "")
+
+ # based on
+ # https://albersdevelopment.net/2019/08/15/using-structlog-with-gunicorn/
+ # and friends
+ # Combined log format:
+ # host - user [time] "request" status size "referer" "user-agent"
+ pattern = re.compile(
+ r"(?P<remote_addr>\S+) \S+ (?P<user>\S+) "
+ r'\[(?P<timestamp>.+?)\] "(?P<request>.+?)" '
+ r"(?P<status>\d+) (?P<size>\S+) "
+ r'"(?P<referrer>.*?)" "(?P<user_agent>.*?)"'
+ )
+
+ match = pattern.match(message)
+ if not match:
+ return event_dict
+
+ fields = match.groupdict()
+
+ # sanitize
+ fields["user"] = None if fields["user"] == "-" else fields["user"]
+ fields["status"] = int(fields["status"])
+ fields["size"] = 0 if fields["size"] == "-" else int(fields["size"])
+ fields["referrer"] = None if fields["referrer"] == "-" else fields["referrer"]
+
+ # Parse "GET /path HTTP/1.1" into separate fields
+ request_parts = fields["request"].split(" ", 2)
+ if len(request_parts) >= 2:
+ fields["method"] = request_parts[0]
+ fields["path"] = request_parts[1]
+ if len(request_parts) == 3:
+ fields["protocol"] = request_parts[2]
+
+ event_dict.update(fields)
+ event_dict["event"] = "http_request"
+ return event_dict
+
+
+def configure_celery_logging(logfile: str | None = None, loglevel: int = logging.INFO):
+ """Configure unified structlog logging for Celery that handles all log types."""
+ processors: list = [
+ structlog.contextvars.merge_contextvars,
+ structlog.processors.TimeStamper(fmt="iso"),
+ structlog.stdlib.add_log_level,
+ structlog.processors.StackInfoRenderer(),
+ structlog.processors.format_exc_info,
+ _add_datadog_context,
+ ]
+ formatter = structlog.stdlib.ProcessorFormatter(
+ processor=RENDERER,
+ foreign_pre_chain=processors,
+ )
+
+ handler = logging.FileHandler(logfile) if logfile else logging.StreamHandler()
+ handler.setFormatter(formatter)
+
+ root = logging.getLogger()
+ root.handlers.clear()
+ root.addHandler(handler)
+ root.setLevel(loglevel)
+
+ structlog.configure(
+ processors=processors
+ + [
+ structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
+ ],
+ logger_factory=structlog.stdlib.LoggerFactory(),
+ wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
+ cache_logger_on_first_use=True,
+ )
+
+
 def _create_logger(request):
  # This has to use **{} instead of just a kwarg because request.id is not
  # an allowed kwarg name.
  return request_logger.bind(**{"request.id": request.id})
 
 
 def includeme(config):
+ # non structlog thigns
+ foreign_pre_chain: list = [
+ structlog.stdlib.add_log_level,
+ structlog.stdlib.add_logger_name,
+ structlog.processors.TimeStamper(fmt="iso"),
+ _add_datadog_context,
+ _parse_gunicorn_access_log,
+ ]
+
  # Configure the standard library logging
  logging.config.dictConfig(
  {
  "version": 1,
  "disable_existing_loggers": False,
- "formatters": {"structlog": {"()": "warehouse.logging.StructlogFormatter"}},
+ "formatters": {
+ "structlog_formatter": {
+ "()": structlog.stdlib.ProcessorFormatter,
+ "processor": RENDERER,
+ "foreign_pre_chain": foreign_pre_chain,
+ }
+ },
  "handlers": {
  "primary": {
  "class": "logging.StreamHandler",
  "stream": "ext://sys.stdout",
- "formatter": "structlog",
+ "formatter": "structlog_formatter",
  },
  },
  "loggers": {
@@ -88,9 +187,11 @@ def includeme(config):
  structlog.stdlib.add_logger_name,
  structlog.stdlib.add_log_level,
  structlog.stdlib.PositionalArgumentsFormatter(),
+ structlog.processors.TimeStamper(fmt="iso"),
  structlog.processors.StackInfoRenderer(),
  structlog.processors.format_exc_info,
- RENDERER,
+ _add_datadog_context,
+ structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
  ],
  logger_factory=structlog.stdlib.LoggerFactory(),
  wrapper_class=structlog.stdlib.BoundLogger,

warehouse/tasks.py

@@ -14,9 +14,11 @@
 import celery.backends.redis
 import pyramid.scripting
 import pyramid_retry
+import structlog
 import transaction
 import venusian
 
+from celery import signals
 from kombu import Queue
 from pyramid.threadlocal import get_current_request
 
@@ -36,6 +38,23 @@
 logger = logging.getLogger(__name__)
 
 
+# Celery signal handlers for unified structlog configuration
+# https://github.com/hynek/structlog/issues/287
+# https://www.structlog.org/en/stable/frameworks.html#celery
+@signals.after_setup_logger.connect
+def on_after_setup_logger(logger, loglevel, logfile, *args, **kwargs):
+ """Override Celery's default logging behavior w/ unified structlog configuration."""
+ from warehouse.logging import configure_celery_logging
+
+ configure_celery_logging(logfile, loglevel)
+
+
+@signals.task_prerun.connect
+def on_task_prerun(sender, task_id, task, **_):
+ """Bind task metadata to contextvars for all logs within the task."""
+ structlog.contextvars.bind_contextvars(task_id=task_id, task_name=task.name)
+
+
 class TLSRedisBackend(celery.backends.redis.RedisBackend):
  def _params_from_url(self, url, defaults):
  params = super()._params_from_url(url, defaults)
@@ -301,6 +320,10 @@ def includeme(config):
  REDBEAT_REDIS_URL=s["celery.scheduler_url"],
  # Silence deprecation warning on startup
  broker_connection_retry_on_startup=False,
+ # Disable Celery's logger hijacking for unified structlog control
+ worker_hijack_root_logger=False,
+ worker_log_format="%(message)s",
+ worker_task_log_format="%(message)s",
  )
  config.registry["celery.app"].Task = WarehouseTask
  config.registry["celery.app"].pyramid_config = config