feat(package)!: Emit structured JSON logs.

components/clp-py-utils/clp_py_utils/clp_logging.py

@@ -1,8 +1,14 @@
+"""Shared logging helpers for CLP Python components."""
+
 import logging
 import os
 import pathlib
 from typing import get_args, Literal
 
+import structlog
+from structlog.stdlib import BoundLogger
+from structlog.typing import Processor
+
 LoggingLevel = Literal[
     "INFO",
     "DEBUG",
@@ -12,12 +18,107 @@
     "CRITICAL",
 ]
 
+# Processor chain for events emitted through structlog loggers before they reach
+# `ProcessorFormatter`.
+_STRUCTLOG_PROCESSORS: tuple[Processor, ...] = (
+    structlog.stdlib.filter_by_level,
+    structlog.contextvars.merge_contextvars,
+    structlog.stdlib.add_logger_name,
+    structlog.stdlib.add_log_level,
+    structlog.stdlib.PositionalArgumentsFormatter(),
+    structlog.processors.TimeStamper(fmt="iso", utc=True, key="timestamp"),
+    structlog.processors.StackInfoRenderer(),
+    structlog.processors.format_exc_info,
+    structlog.processors.UnicodeDecoder(),
+    structlog.processors.CallsiteParameterAdder(
+        {
+            structlog.processors.CallsiteParameter.FILENAME,
+            structlog.processors.CallsiteParameter.FUNC_NAME,
+            structlog.processors.CallsiteParameter.LINENO,
+        }
+    ),
+)
+
+# Processor chain for stdlib `LogRecord` objects that did not originate from structlog.
+_FOREIGN_PRE_CHAIN: tuple[Processor, ...] = (
+    structlog.contextvars.merge_contextvars,
+    structlog.stdlib.add_logger_name,
+    structlog.stdlib.add_log_level,
+    structlog.stdlib.ExtraAdder(),
+    structlog.stdlib.PositionalArgumentsFormatter(),
+    structlog.processors.TimeStamper(fmt="iso", utc=True, key="timestamp"),
+    structlog.processors.StackInfoRenderer(),
+    structlog.processors.format_exc_info,
+    structlog.processors.UnicodeDecoder(),
+    structlog.processors.CallsiteParameterAdder(
+        {
+            structlog.processors.CallsiteParameter.FILENAME,
+            structlog.processors.CallsiteParameter.FUNC_NAME,
+            structlog.processors.CallsiteParameter.LINENO,
+        }
+    ),
+)
+
+
+def configure_structlog() -> None:
+    """Configure structlog with CLP's default processor chain if it is not already configured."""
+    if not structlog.is_configured():
+        structlog.configure(
+            processors=[
+                *_STRUCTLOG_PROCESSORS,
+                structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
+            ],
+            logger_factory=structlog.stdlib.LoggerFactory(),
+            wrapper_class=BoundLogger,
+            cache_logger_on_first_use=True,
+        )
+
+
+def get_structlog_logger(name: str) -> BoundLogger:
+    """
+    Configure CLP's structured logging defaults and return a structlog logger for a component.
+
+    CLP uses structlog's stdlib integration, so this first configures the stdlib logger's
+    handlers with structlog's `ProcessorFormatter`. Use the returned structlog logger for log calls.
+
+    :param name: Name of the logger to create or retrieve.
+    :return: A structlog logger configured to emit through CLP's JSON formatter.
+    """
+    configure_structlog()
+    stdlib_logger = get_logger(name)
+    configure_logging(stdlib_logger, name)
+    return structlog.stdlib.get_logger(name)
+
+
+def get_logging_formatter() -> logging.Formatter:
+    """:return: A JSON log formatter configured with CLP's structlog processors."""
+    return structlog.stdlib.ProcessorFormatter(
+        # foreign_pre_chain is run for stdlib LogRecord objects that do not go through
+        # structlog's processor chain.
+        foreign_pre_chain=_FOREIGN_PRE_CHAIN,
+        processors=[
+            structlog.stdlib.ProcessorFormatter.remove_processors_meta,
+            structlog.processors.JSONRenderer(),
+        ],
+    )
+
 
-def get_logging_formatter():
-    return logging.Formatter("%(asctime)s %(name)s [%(levelname)s] %(message)s")
+def set_json_formatter_on_handlers(logger: logging.Logger) -> None:
+    """
+    Set CLP's JSON log formatter on all handlers currently attached to the logger.
+
+    :param logger: Logger whose handlers should use CLP's JSON formatter.
+    """
+    formatter = get_logging_formatter()
+    for handler in logger.handlers:
+        handler.setFormatter(formatter)
 
 
-def get_logger(name: str):
+def get_logger(name: str) -> logging.Logger:
+    """
+    :param name: Name of the logger to create or retrieve.
+    :return: A logger configured with CLP's JSON console formatter.
+    """
     logger = logging.getLogger(name)
     # Setup console logging
     logging_console_handler = logging.StreamHandler()
@@ -28,12 +129,19 @@ def get_logger(name: str):
     return logger
 
 
-def set_logging_level(logger: logging.Logger, level: str | None):
+def set_logging_level(logger: logging.Logger, level: str | None) -> None:
+    """
+    Set a logger to the requested logging level, defaulting to INFO when the level is unset or
+    invalid.
+
+    :param logger: Logger whose level should be updated.
+    :param level: Requested logging level, or `None` to use INFO.
+    """
     if level is None:
         logger.setLevel(logging.INFO)
         return
     if level not in get_args(LoggingLevel):
-        logger.warning(f"Invalid logging level: {level}, using INFO as default")
+        logger.warning("Invalid logging level: %s, using INFO as default", level)
         logger.setLevel(logging.INFO)
         return
 
@@ -43,7 +151,7 @@ def set_logging_level(logger: logging.Logger, level: str | None):
 def configure_logging(
     logger: logging.Logger,
     component_name: str,
-):
+) -> None:
     """
     Configures file logging and the logging level for a logger using environment variables.
 

components/clp-py-utils/pyproject.toml

@@ -18,6 +18,7 @@ dependencies = [
     "result>=0.17.0",
     "sqlalchemy>=2.0.46",
     "StrEnum>=0.4.15",
+    "structlog>=26.1.0",
 ]
 
 [build-system]

components/job-orchestration/job_orchestration/executor/compress/celery.py

@@ -1,5 +1,9 @@
-from celery import Celery
-from celery.signals import worker_process_init, worker_process_shutdown
+"""Celery app configuration for compression task execution."""
+
+import logging
+
+from celery import Celery, signals
+from clp_py_utils.clp_logging import set_json_formatter_on_handlers
 from clp_py_utils.telemetry import init_telemetry, shutdown_telemetry
 
 from job_orchestration.executor.compress import celeryconfig
@@ -8,12 +12,25 @@
 app.config_from_object(celeryconfig)
 
 
-@worker_process_init.connect
+@signals.after_setup_logger.connect
+@signals.after_setup_task_logger.connect
+def setup_json_logging(logger: logging.Logger | None = None, **_: object) -> None:
+    """
+    Use CLP's JSON formatter for loggers configured by Celery.
+
+    :param logger: Logger configured by Celery, if one was provided by the signal.
+    :param _: Additional Celery signal keyword arguments. Unused.
+    """
+    if logger is not None:
+        set_json_formatter_on_handlers(logger)
+
+
+@signals.worker_process_init.connect
 def setup_telemetry(**kwargs) -> None:
     init_telemetry()
 
 
-@worker_process_shutdown.connect
+@signals.worker_process_shutdown.connect
 def teardown_telemetry(**kwargs) -> None:
     shutdown_telemetry()
 

components/job-orchestration/job_orchestration/executor/query/celery.py

@@ -1,9 +1,28 @@
-from celery import Celery
+"""Celery app configuration for query task execution."""
+
+import logging
+
+from celery import Celery, signals
+from clp_py_utils.clp_logging import set_json_formatter_on_handlers
 
 from job_orchestration.executor.query import celeryconfig
 
 app = Celery("query")
 app.config_from_object(celeryconfig)
 
+
+@signals.after_setup_logger.connect
+@signals.after_setup_task_logger.connect
+def setup_json_logging(logger: logging.Logger | None = None, **_: object) -> None:
+    """
+    Use CLP's JSON formatter for loggers configured by Celery.
+
+    :param logger: Logger configured by Celery, if one was provided by the signal.
+    :param _: Additional Celery signal keyword arguments. Unused.
+    """
+    if logger is not None:
+        set_json_formatter_on_handlers(logger)
+
+
 if "__main__" == __name__:
     app.start()

docs/src/dev-docs/index.md

@@ -96,6 +96,7 @@ tooling-gh-workflows
 
 design-project-structure
 design-deployment-orchestration
+logging
 
 design-kv-ir-streams/index
 design-metadata-db