Merge pull request #542 from mdekstrand/feature/fix-process-logging

Fix subprocess logging

lenskit/lenskit/logging/monitor.py

@@ -210,6 +210,6 @@ def _handle_log_message(self):
             logger = structlog.get_logger(name)
             data = json.loads(data)
             method = getattr(logger, data["method"])
-            method(data["event"])
+            method(**data["event"])
         else:
             _log.error("invalid log backend")

lenskit/lenskit/logging/processors.py

@@ -2,6 +2,7 @@
 LensKit logging processors.
 """
 
+import multiprocessing as mp
 from datetime import datetime
 from typing import Any
 
@@ -32,3 +33,17 @@ def format_timestamp(logger: Any, method: str, event_dict: EventDict) -> EventDi
         return event_dict
     else:
         return event_dict
+
+
+def add_process_info(logger: Any, method: str, event_dict: EventDict) -> EventDict:
+    """
+    Add process info if it does not exist.
+    """
+
+    proc = mp.current_process()
+    if "pid" not in event_dict:
+        event_dict["pid"] = proc.pid
+    if "pname" not in event_dict:
+        event_dict["pname"] = proc.name
+
+    return event_dict

lenskit/lenskit/logging/worker.py

@@ -23,8 +23,10 @@
 
 from .config import CORE_PROCESSORS, active_logging_config
 from .monitor import get_monitor
+from .processors import add_process_info
 
 _active_context: WorkerContext | None = None
+_log = structlog.stdlib.get_logger(__name__)
 
 
 @dataclass
@@ -82,9 +84,13 @@ def start(self):
         root.setLevel(self.config.level)
 
         structlog.configure(
-            CORE_PROCESSORS + [self._log_handler.send_structlog],
+            [add_process_info]
+            + CORE_PROCESSORS
+            + [structlog.processors.ExceptionPrettyPrinter(), self._log_handler.send_structlog],
+            wrapper_class=structlog.make_filtering_bound_logger(self.config.level),
             logger_factory=structlog.stdlib.LoggerFactory(),
         )
+        _log.debug("log context activated")
 
     def shutdown(self):
         root = getLogger()
@@ -119,6 +125,7 @@ def handle(self, record: LogRecord) -> LogRecord | bool:  # type: ignore
         # update messages for copyability
         if not hasattr(record, "message"):
             record.message = record.msg % record.args
+
         record.exc_info = None
         record.exc_text = None
         record.stack_info = None

lenskit/lenskit/parallel/pool.py

@@ -71,8 +71,8 @@ def __init__(self, *args: Any, **kwargs: Any):
 
     @override
     def run(self):
-        log = _log.bind(pid=self.pid, pname=self.name)
         with WorkerContext(self._log_config):
+            log = _log.bind(pid=self.pid, pname=self.name)
             log.info("multiprocessing worker started")
             super().run()
 

lenskit/lenskit/parallel/worker.py

@@ -50,7 +50,7 @@ def initalize(cfg: WorkerConfig, ctx: ModelData) -> None:
         log.error("deserialization failed: %s", e)
         raise e
 
-    log.debug("worker %d ready (process %s)", os.getpid(), mp.current_process())
+    log.debug("worker %d ready (process %s)", os.getpid(), mp.current_process().name)
 
 
 def worker(arg: Any) -> Any: