feat(logging): add datadog context, celery structure logs

tests/unit/test_logging.py

@@ -119,9 +119,11 @@ 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,
+                mock.ANY,  # _add_datadog_context
                 wlogging.RENDERER,
             ],
             logger_factory=mock.ANY,
@@ -133,14 +135,114 @@ def test_includeme(monkeypatch, settings, expected_level):
         configure.calls[0].kwargs["processors"][3],
         structlog.stdlib.PositionalArgumentsFormatter,
     )
+    # timestamper #18843
     assert isinstance(
         configure.calls[0].kwargs["processors"][4],
+        structlog.processors.TimeStamper,
+    )
+    assert isinstance(
+        configure.calls[0].kwargs["processors"][5],
         structlog.processors.StackInfoRenderer,
     )
+    # _add_datadog_context #18843
+    assert configure.calls[0].kwargs["processors"][7] == wlogging._add_datadog_context
     assert isinstance(
         configure.calls[0].kwargs["logger_factory"], structlog.stdlib.LoggerFactory
     )
     assert config.add_request_method.calls == [
         pretend.call(wlogging._create_id, name="id", reify=True),
         pretend.call(wlogging._create_logger, name="log", reify=True),
     ]
+
+
+def test_add_datadog_context(monkeypatch):
+    monkeypatch.setenv("DD_ENV", "production")
+    monkeypatch.setenv("DD_VERSION", "1.2.3")
+
+    event_dict = {"event": "test"}
+    result = wlogging._add_datadog_context(None, None, event_dict)
+
+    assert result["dd.env"] == "production"
+    assert result["dd.version"] == "1.2.3"
+
+
+def test_add_datadog_context_with_span(monkeypatch):
+    """Test Datadog context with an active span."""
+    import sys
+
+    mock_span = pretend.stub(trace_id=123, span_id=456, service="test-svc")
+    mock_tracer = pretend.stub(current_span=lambda: mock_span)
+    mock_ddtrace = pretend.stub(tracer=mock_tracer)
+
+    sys.modules["ddtrace"] = mock_ddtrace
+    monkeypatch.setenv("DD_ENV", "prod")
+    monkeypatch.setenv("DD_VERSION", "2.0")
+
+    try:
+        event_dict = {"event": "test"}
+        result = wlogging._add_datadog_context(None, None, event_dict)
+
+        assert result["dd.trace_id"] == "123"
+        assert result["dd.span_id"] == "456"
+        assert result["dd.service"] == "test-svc"
+    finally:
+        del sys.modules["ddtrace"]
+
+
+def test_add_datadog_context_no_ddtrace(monkeypatch):
+    """Test Datadog context when ddtrace is not available."""
+    import sys
+    import builtins
+
+    # Block ddtrace import
+    # we just want to simulate ddtrace beiung unavailable
+    # to make coverage happy :)
+    original_import = builtins.__import__
+
+    def mock_import(name, *args, **kwargs):
+        if name == "ddtrace":
+            raise ImportError("No module named 'ddtrace'")
+        return original_import(name, *args, **kwargs)
+
+    # Remove from sys.modules if present
+    if "ddtrace" in sys.modules:
+        del sys.modules["ddtrace"]
+
+    monkeypatch.setattr(builtins, "__import__", mock_import)
+
+    event_dict = {"event": "test"}
+    result = wlogging._add_datadog_context(None, None, event_dict)
+
+    # Should not have any dd fields when import fails
+    assert "dd.trace_id" not in result
+    assert "dd.span_id" not in result
+    assert "dd.service" not in result
+    assert "dd.env" not in result
+    assert "dd.version" not in result
+
+
+def test_configure_celery_logging(monkeypatch):
+    configure = pretend.call_recorder(lambda **kw: None)
+    monkeypatch.setattr(structlog, "configure", configure)
+
+    mock_handler = pretend.stub(setFormatter=pretend.call_recorder(lambda f: None))
+    mock_logger = pretend.stub(
+        handlers=pretend.stub(clear=pretend.call_recorder(lambda: None)),
+        setLevel=pretend.call_recorder(lambda level: None),
+        addHandler=pretend.call_recorder(lambda add_handler: None),
+        removeHandler=pretend.call_recorder(lambda remove_handler: None),
+    )
+    monkeypatch.setattr(logging, "getLogger", lambda: mock_logger)
+    monkeypatch.setattr(logging, "StreamHandler", lambda: mock_handler)
+
+    wlogging.configure_celery_logging()
+
+    # Verify handlers cleared and new one added
+    assert mock_logger.handlers.clear.calls == [pretend.call()]
+    assert len(mock_logger.addHandler.calls) == 1
+    assert mock_logger.setLevel.calls == [pretend.call(logging.INFO)]
+
+    # Verify processors
+    processors = configure.calls[0].kwargs["processors"]
+    assert structlog.contextvars.merge_contextvars in processors
+    assert wlogging._add_datadog_context in processors

tests/unit/test_tasks.py

@@ -524,3 +524,26 @@ def test_includeme(env, ssl, broker_redis_url, expected_url, transport_options):
     assert config.add_request_method.calls == [
         pretend.call(tasks._get_task_from_request, name="task", reify=True)
     ]
+
+
+def test_on_after_setup_logger(monkeypatch):
+    configure_celery_logging = pretend.call_recorder(lambda logfile, loglevel: None)
+    monkeypatch.setattr(
+        "warehouse.logging.configure_celery_logging", configure_celery_logging
+    )
+
+    tasks.on_after_setup_logger("logger", "loglevel", "logfile")
+
+    assert configure_celery_logging.calls == [pretend.call("logfile", "loglevel")]
+
+
+def test_on_task_prerun(monkeypatch):
+    bind_contextvars = pretend.call_recorder(lambda **kw: None)
+    monkeypatch.setattr("structlog.contextvars.bind_contextvars", bind_contextvars)
+
+    task = pretend.stub(name="test.task")
+    tasks.on_task_prerun(None, "task-123", task)
+
+    assert bind_contextvars.calls == [
+        pretend.call(task_id="task-123", task_name="test.task")
+    ]

warehouse/logging.py

@@ -1,6 +1,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import logging.config
+import os
 import threading
 import uuid
 
@@ -36,6 +37,58 @@ 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 configure_celery_logging(logfile: str | None = None, loglevel: int = logging.INFO):
+    """Configure unified structlog logging for Celery that handles all log types."""
+    processors = [
+        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,  # type: ignore[arg-type]
+    )
+
+    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  # type: ignore[arg-type]
+        + [
+            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.
@@ -88,8 +141,10 @@ 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,
+            _add_datadog_context,
             RENDERER,
         ],
         logger_factory=structlog.stdlib.LoggerFactory(),

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
 
@@ -37,6 +39,21 @@
 logger = logging.getLogger(__name__)
 
 
+# Celery signal handlers for unified structlog configuration
+@signals.after_setup_logger.connect
+def on_after_setup_logger(logger, loglevel, logfile, *args, **kwargs):
+    """Override Celery's default logging behavior with unified structlog configuration."""  # noqa: E501
+    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)
@@ -302,6 +319,10 @@ def includeme(config: Configurator) -> None:
         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