feature: support logfmt log formatting

Author: ClemDoum

datashare-python/datashare_python/config.py

@@ -1,3 +1,4 @@
+from enum import StrEnum
 from pathlib import Path
 from typing import Literal
 
@@ -78,13 +79,21 @@ async def to_client(self) -> TemporalClient:
 LogLevel = Literal["CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG"]
 
 
+class LogFormat(StrEnum):
+    JSON = "json"
+    LOGFMT = "logfmt"
+    DEFAULT = "default"
+
+
 class LoggingConfig(BaseModel):
-    log_in_json: bool = False
+    format: LogFormat = LogFormat.DEFAULT
     loggers: dict[str, LogLevel]
 
 
 _DEFAULT_LOGGERS = {datashare_python.__name__: "INFO"}
-_DEFAULT_LOGGING_CONFIG = LoggingConfig(log_in_json=True, loggers=_DEFAULT_LOGGERS)
+_DEFAULT_LOGGING_CONFIG = LoggingConfig(
+    format=LogFormat.DEFAULT, loggers=_DEFAULT_LOGGERS
+)
 
 
 class WorkerConfig(ICIJSettings, BaseModel):

datashare-python/datashare_python/conftest.py

@@ -13,6 +13,7 @@
 
 from datashare_python.config import (
     DatashareClientConfig,
+    LogFormat,
     LoggingConfig,
     TemporalClientConfig,
     WorkerConfig,
@@ -93,7 +94,7 @@ def event_loop(
 @pytest.fixture(scope="session")
 def test_worker_config() -> WorkerConfig:
     logging_config = LoggingConfig(
-        log_in_json=False,
+        format=LogFormat.DEFAULT,
         loggers={
             "datashare_python": "DEBUG",
             "icij_common": "DEBUG",

datashare-python/datashare_python/dependencies.py

@@ -40,7 +40,7 @@ def set_loggers(
     worker_config: WorkerConfig, worker_id: str, loggers: dict[str, LogLevel]
 ) -> None:
     setup_worker_loggers(
-        loggers=loggers, worker_id=worker_id, in_json=worker_config.logging.log_in_json
+        loggers=loggers, worker_id=worker_id, format=worker_config.logging.format
     )
     logger.info("worker loggers ready to log 💬")
 

datashare-python/datashare_python/logging_.py

@@ -1,20 +1,39 @@
 import logging
+import numbers
 import sys
 from copy import copy
+from typing import Any
 
+import orjson
 from icij_common.logging_utils import DATE_FMT, STREAM_HANDLER_FMT
-from pythonjsonlogger.core import RESERVED_ATTRS, BaseJsonFormatter
+from pythonjsonlogger.core import BaseJsonFormatter
 from pythonjsonlogger.orjson import OrjsonFormatter
 from temporalio import activity, workflow
 
-from .config import LogLevel
+from .config import LogFormat, LogLevel
 from .interceptors import get_trace_context
 
+_BASE_ATTRS = [
+    "asctime",
+    "exc_info",
+    "filename",
+    "funcName",
+    "levelname",
+    "levelno",
+    "lineno",
+    "module",
+    "msecs",
+    "message",
+    "msg",
+    "name",
+    "pathname",
+]
 _ACT_LOGGER_ATTRS = ["activity_type", "activity_id", "activity_run_id"]
 _WF_LOGGED_ATTRS = ["workflow_type", "workflow_id", "workflow_run_id"]
 _TRACE_CONTEXT_ATTRS = ["trace_id", "parent_id", "traceparent"]
+
 _LOGGED_ATTRIBUTES = (
-    copy(RESERVED_ATTRS)
+    copy(_BASE_ATTRS)
     + _WF_LOGGED_ATTRS
     + _ACT_LOGGER_ATTRS
     + _TRACE_CONTEXT_ATTRS
@@ -28,15 +47,15 @@
 
 
 def setup_worker_loggers(
-    loggers: dict[str, LogLevel], *, worker_id: str | None, in_json: bool
+    loggers: dict[str, LogLevel], *, worker_id: str | None, format: LogFormat
 ) -> None:
     worker_filter = WorkerFilter(worker_id)
     for logger_name, level_str in loggers.items():
         level = getattr(logging, level_str)
         logger = logging.getLogger(logger_name)
         logger.setLevel(level)
         logger.handlers = []
-        for handler in _get_worker_handlers(level, worker_filter, in_json=in_json):
+        for handler in _get_worker_handlers(level, worker_filter, format=format):
             logger.addHandler(handler)
 
 
@@ -64,23 +83,50 @@ def filter(self, record: logging.LogRecord) -> bool:
 
 
 def _get_worker_handlers(
-    level: int, worker_filter: WorkerFilter, *, in_json: bool
+    level: int, worker_filter: WorkerFilter, *, format: LogFormat
 ) -> list[logging.Handler]:
     stream_handler = logging.StreamHandler(sys.stderr)
-    if in_json:
-        fmt = _json_formatter(datefmt=DATE_FMT)
-    else:
-        if worker_filter.worker_id is not None:
-            fmt = _STREAM_HANDLER_FMT_WITH_WORKER_ID
-        else:
-            fmt = STREAM_HANDLER_FMT
-        fmt = logging.Formatter(fmt, DATE_FMT)
+    match format:
+        case LogFormat.JSON:
+            fmt = _json_formatter(datefmt=DATE_FMT)
+        case LogFormat.LOGFMT:
+            fmt = LogFmtFormatter(datefmt=DATE_FMT)
+        case LogFormat.DEFAULT:
+            if worker_filter.worker_id is not None:
+                fmt = _STREAM_HANDLER_FMT_WITH_WORKER_ID
+            else:
+                fmt = STREAM_HANDLER_FMT
+            fmt = logging.Formatter(fmt, DATE_FMT)
+        case _:
+            raise NotImplementedError(f"invalid log format: {format}")
     stream_handler.setFormatter(fmt)
     stream_handler.setLevel(level)
     stream_handler.addFilter(worker_filter)
     return [stream_handler]
 
 
+class LogFmtFormatter(logging.Formatter):
+    def format(self, record: logging.LogRecord) -> str:
+        logged = dict()
+        if record.exc_info and not record.exc_text:
+            record.exc_text = self.formatException(record.exc_info)
+            logged["exc_info"] = record.exc_text
+        for k, v in record.__dict__.items():
+            if k in _LOGGED_ATTRIBUTES and k != "exc_info":
+                logged[k] = _encode_value(v)
+        return " ".join(f"{k}={v}" for k, v in sorted(logged.items()))
+
+
+def _encode_value(value: Any) -> str:
+    if value is None:
+        return ""
+    if isinstance(value, bool):
+        return "true" if value else "false"
+    if isinstance(value, numbers.Number):
+        return str(value)
+    return orjson.dumps(value).decode()
+
+
 def _json_formatter(datefmt: str) -> BaseJsonFormatter:
     fmt = OrjsonFormatter(  # let's keep logging as fast as possible
         _LOGGED_ATTRIBUTES, datefmt=datefmt

datashare-python/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "datashare-python"
-version = "0.7.3"
+version = "0.8.0"
 description = "Manage Python tasks and local resources in Datashare"
 authors = [
   { name = "Clément Doumouro", email = "cdoumouro@icij.org" },

datashare-python/tests/test_logging.py

@@ -0,0 +1,31 @@
+import logging
+import re
+import sys
+from logging import LogRecord
+
+from datashare_python.logging_ import LogFmtFormatter
+
+
+def test_logfmt_formatter() -> None:
+    # Given
+    fmt = LogFmtFormatter()
+    exc_info = sys.exc_info()
+    record = LogRecord(
+        name="some_logger",
+        level=logging.DEBUG,
+        pathname="/some/path",
+        lineno=2,
+        msg="some message with space in it",
+        args=dict(),
+        exc_info=exc_info,
+    )
+    # When
+    logged = fmt.format(record)
+    # Then
+    expected_logged = (
+        'exc_info=NoneType: None filename="path" funcName= '
+        r'levelname="DEBUG" levelno=10 lineno=2 module="path" msecs=\d+.\d+'
+        ' msg="some message with space in it" name="some_logger" pathname="/some/path"'
+    )
+    expected_logged = re.compile(expected_logged)
+    assert expected_logged.match(logged)