Logs configuration improvements

Signed-off-by: Federico Busetti <[email protected]>

Author: febus982

pyproject.toml

@@ -80,6 +80,7 @@ source = ["src"]
 omit = [
     "src/alembic/*",
     "src/bootstrap/config.py",
+    "src/bootstrap/logs/*",
 ]
 # It's not necessary to configure concurrency here
 # because pytest-cov takes care of that

src/bootstrap/bootstrap.py

@@ -4,8 +4,9 @@
 from domains import init_celery, init_domains
 from pydantic import BaseModel, ConfigDict
 
-from .config import AppConfig, init_logger
+from .config import AppConfig
 from .di_container import Container
+from .logs import init_logger
 from .storage import init_storage
 
 

src/bootstrap/config.py

@@ -1,13 +1,9 @@
-import logging
 from pathlib import Path
-from typing import Dict, List, Literal, Optional
+from typing import Dict, Literal, Optional
 
-import structlog
-from opentelemetry import trace
 from pydantic import BaseModel
 from pydantic_settings import BaseSettings, SettingsConfigDict
 from sqlalchemy_bind_manager import SQLAlchemyAsyncConfig
-from structlog.typing import Processor
 
 TYPE_ENVIRONMENT = Literal["local", "test", "staging", "production"]
 
@@ -65,87 +61,3 @@ class AppConfig(BaseSettings):
             ),
         ),
     )
-
-
-def init_logger(config: AppConfig) -> None:
-    """
-    Configure structlog and stdlib logging with shared handler and formatter.
-
-    :param config: The app configuration
-    :type config: AppConfig
-    :return:
-    """
-    # These processors will be used by both structlog and stdlib logger
-    processors: List[Processor] = [
-        structlog.contextvars.merge_contextvars,
-        structlog.stdlib.add_logger_name,
-        structlog.stdlib.PositionalArgumentsFormatter(),
-        structlog.processors.add_log_level,
-        structlog.processors.StackInfoRenderer(),
-        _add_logging_open_telemetry_spans,
-    ]
-
-    log_level = logging.DEBUG if config.DEBUG else logging.INFO
-    if config.ENVIRONMENT not in ["local", "test"]:
-        processors.append(structlog.stdlib.ProcessorFormatter.remove_processors_meta)
-        processors.append(structlog.processors.TimeStamper(fmt="iso", utc=True))
-        processors.append(structlog.processors.dict_tracebacks)
-        processors.append(structlog.processors.JSONRenderer())
-    else:
-        processors.append(structlog.stdlib.ProcessorFormatter.remove_processors_meta)
-        processors.append(
-            structlog.processors.TimeStamper(fmt="%d-%m-%Y %H:%M:%S", utc=True)
-        )
-        processors.append(structlog.dev.ConsoleRenderer())
-
-    structlog.stdlib.recreate_defaults()
-    """
-    Even if we set the loglevel using the stdlib setLevel later,
-    using make_filtering_bound_logger will filter events before
-    in the chain, producing a performance improvement
-    """
-    structlog.configure(
-        wrapper_class=structlog.make_filtering_bound_logger(log_level),
-        processors=[
-            # This prepares the log events to be handled by stdlib
-            structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
-        ],
-        logger_factory=structlog.stdlib.LoggerFactory(),
-        cache_logger_on_first_use=True,
-    )
-    stdlib_handler = logging.StreamHandler()
-    # Use structlog `ProcessorFormatter` to format all `logging` entries
-    stdlib_handler.setFormatter(
-        structlog.stdlib.ProcessorFormatter(
-            processors=processors,
-        )
-    )
-    stdlib_logger = logging.getLogger()
-    stdlib_logger.handlers.clear()
-    stdlib_logger.addHandler(stdlib_handler)
-    stdlib_logger.setLevel(log_level)
-
-    for _log in ["uvicorn", "uvicorn.error", "uvicorn.access"]:
-        # Clear the log handlers for uvicorn loggers, and enable propagation
-        # so the messages are caught by our root logger and formatted correctly
-        # by structlog. Initial messages from reloader startup are not caught.
-        logging.getLogger(_log).handlers.clear()
-        logging.getLogger(_log).propagate = True
-
-
-def _add_logging_open_telemetry_spans(_, __, event_dict):
-    span = trace.get_current_span()
-    if not span.is_recording():
-        event_dict["span"] = None
-        return event_dict
-
-    ctx = span.get_span_context()
-    parent = getattr(span, "parent", None)
-
-    event_dict["span"] = {
-        "span_id": hex(ctx.span_id),
-        "trace_id": hex(ctx.trace_id),
-        "parent_span_id": None if not parent else hex(parent.span_id),
-    }
-
-    return event_dict

src/bootstrap/logs/__init__.py

@@ -0,0 +1,93 @@
+import logging
+from typing import List
+
+import structlog
+from opentelemetry import trace
+from structlog.typing import EventDict, Processor
+
+from ..config import AppConfig
+from .processors import (
+    add_logging_open_telemetry_spans,
+    drop_color_message_key,
+    extract_from_record,
+)
+
+
+def init_logger(config: AppConfig) -> None:
+    """
+    Configure structlog and stdlib logging with shared handler and formatter.
+
+    :param config: The app configuration
+    :type config: AppConfig
+    :return:
+    """
+    # Strongly inspired by https://gist.github.com/nymous/f138c7f06062b7c43c060bf03759c29e
+
+    # These processors will be used by both structlog and stdlib logger
+    shared_processors: List[Processor] = [
+        structlog.contextvars.merge_contextvars,
+        structlog.stdlib.add_logger_name,
+        structlog.stdlib.add_log_level,
+        structlog.stdlib.PositionalArgumentsFormatter(),
+        structlog.stdlib.ExtraAdder(),
+        drop_color_message_key,
+        add_logging_open_telemetry_spans,
+        structlog.processors.StackInfoRenderer(),
+    ]
+
+    # stdlib_processors are executed before the shared ones, so processors
+    # accessing processor metadata such as `_extract_from_record` must
+    # run here, before `remove_processors_meta`
+    stdlib_processors: List[Processor] = [
+        extract_from_record,
+        structlog.stdlib.ProcessorFormatter.remove_processors_meta,
+    ]
+
+    log_level = logging.DEBUG if config.DEBUG else logging.INFO
+    if config.ENVIRONMENT in ["local", "test"]:
+        shared_processors.append(
+            structlog.processors.TimeStamper(fmt="%d-%m-%Y %H:%M:%S", utc=True)
+        )
+        stdlib_processors.append(structlog.dev.ConsoleRenderer())
+    else:
+        shared_processors.append(structlog.processors.TimeStamper(fmt="iso", utc=True))
+        shared_processors.append(structlog.processors.dict_tracebacks)
+        stdlib_processors.append(structlog.processors.JSONRenderer())
+
+    """
+    Even if we set the loglevel using the stdlib setLevel later,
+    using make_filtering_bound_logger will filter events before
+    in the chain, producing a performance improvement
+    """
+    structlog.configure(
+        wrapper_class=structlog.make_filtering_bound_logger(log_level),
+        processors=[
+            *shared_processors,
+            # This prepares the log events to be handled by stdlib
+            structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
+        ],
+        logger_factory=structlog.stdlib.LoggerFactory(),
+        cache_logger_on_first_use=True,
+    )
+
+    # Create a handler for stdlib logger
+    stdlib_handler = logging.StreamHandler()
+    stdlib_handler.setFormatter(
+        structlog.stdlib.ProcessorFormatter(
+            foreign_pre_chain=shared_processors,
+            processors=stdlib_processors,
+        )
+    )
+
+    # Use structlog to format logs coming from stdlib logger
+    stdlib_logger = logging.getLogger()
+    # stdlib_logger.handlers.clear()
+    stdlib_logger.addHandler(stdlib_handler)
+    stdlib_logger.setLevel(log_level)
+
+    for _log in ["uvicorn", "uvicorn.error", "uvicorn.access"]:
+        # Clear the log handlers for uvicorn loggers, and enable propagation
+        # so the messages are caught by our root logger and formatted correctly
+        # by structlog. Initial messages from reloader startup are not caught.
+        logging.getLogger(_log).handlers.clear()
+        logging.getLogger(_log).propagate = True

src/bootstrap/logs/processors.py

@@ -0,0 +1,39 @@
+from opentelemetry import trace
+from structlog.typing import EventDict
+
+
+def extract_from_record(_, __, event_dict: EventDict) -> EventDict:
+    """
+    Extract thread and process names and add them to the event dict.
+    """
+    record = event_dict["_record"]
+    event_dict["thread_name"] = record.threadName
+    event_dict["process_name"] = record.processName
+    return event_dict
+
+
+def drop_color_message_key(_, __, event_dict: EventDict) -> EventDict:
+    """
+    Uvicorn logs the message a second time in the extra `color_message`, but we don't
+    need it. This processor drops the key from the event dict if it exists.
+    """
+    event_dict.pop("color_message", None)
+    return event_dict
+
+
+def add_logging_open_telemetry_spans(_, __, event_dict: EventDict) -> EventDict:
+    span = trace.get_current_span()
+    if not span.is_recording():
+        event_dict["span"] = None
+        return event_dict
+
+    ctx = span.get_span_context()
+    parent = getattr(span, "parent", None)
+
+    event_dict["span"] = {
+        "span_id": hex(ctx.span_id),
+        "trace_id": hex(ctx.trace_id),
+        "parent_span_id": None if not parent else hex(parent.span_id),
+    }
+
+    return event_dict