From 0c965fab6d59981dae69bf39e1d8b061f6c30250 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 3 Oct 2026 12:02:58 +0100 Subject: [PATCH] Add optional queue-based non-blocking logging ## Summary File, syslog and network handlers can block the thread that logs. ## Changes - `setup_logging(use_queue=True)` puts records on a `queue.SimpleQueue`. A `QueueListener` thread owns the real handlers, and per-handler levels are still respected. - `shutdown_logging()` removes the queue handlers, writes whatever is still queued, stops the listener threads and closes their handlers. It is idempotent and registered with `atexit`. Calling `setup_logging()` again for a logger stops its previous listener, so threads do not leak. - The context filter runs on the queue handler, that is on the logging thread. The listener thread cannot see that thread's context variables. - `logging.handlers.QueueHandler` formats the record and drops `exc_info`, which would fold tracebacks into the message and out of the JSON `exc_info` field. A small subclass only resolves the message arguments, so tracebacks are unchanged. - Documented in the README and changelog, and `shutdown_logging` is exported. ## Trade-offs The queue is unbounded, so a producer that outruns the handlers grows memory. A bounded queue needs a drop or block policy, which is a design decision beyond this issue, so it is not included. ## Verification - Tests cover ordering and completeness after shutdown, captured context, handler levels, tracebacks in JSON, a handler that blocks while the caller does not, reconfiguration, switching the queue off, and idempotent shutdown. - `pytest`: 133 passed (latest dependencies) and 112 passed with 21 skipped (no extras). `ruff` and `mypy --strict` are clean. Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01U2nYHVGDdthQpcwNBPbp8r --- CHANGELOG.md | 3 + README.md | 20 ++++++ pylogkit/__init__.py | 2 + pylogkit/log_setup.py | 105 ++++++++++++++++++++++++--- tests/test_queue.py | 163 ++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 285 insertions(+), 8 deletions(-) create mode 100644 tests/test_queue.py diff --git a/CHANGELOG.md b/CHANGELOG.md index c8d65cc..255a649 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -44,6 +44,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). - `setup_logging()` reads `LOG_LEVEL`, `LOG_FORMAT` (`pretty`, `compact`, `json` or `auto`) and `LOG_FILE` when the matching arguments are not passed. Explicit arguments always override the environment (#22). +- `setup_logging(use_queue=True)` hands records to a background thread through + a queue, so slow handlers do not block the caller, with `shutdown_logging()` + to flush and stop it. It also runs at interpreter exit (#23). ### Changed diff --git a/README.md b/README.md index 086a3ea..964cb19 100644 --- a/README.md +++ b/README.md @@ -129,6 +129,7 @@ log.info("outside") # carries neither | `to_console` | `True` | Log to the console. | | `console_level` | `None` | Console level, defaulting to `level`. | | `console_stream` | `None` | Console stream, for example `sys.stderr`. Defaults to `sys.stdout`. | +| `use_queue` | `False` | Write records on a background thread, so slow handlers do not block your code. See below. | | `use_emoji` | `True` | Show a level emoji in the verbose console layout. Set `False` for terminals that render emoji badly. | | `mode` | `None` | Console layout: `"verbose"` (coloured, with source location and context) or `"compact"` (`[LEVEL] message`). Falls back to `LOG_FORMAT`, then `"verbose"`. | | `use_json` | `None` | Write JSON lines to the console instead. Needs the `json` extra. Falls back to `LOG_FORMAT`. | @@ -158,6 +159,25 @@ Things worth knowing: - Asking for JSON output without `python-json-logger` installed raises an `ImportError` with an install hint. +### Non-blocking logging + +File, syslog and network handlers can stall the thread that logs. With `use_queue=True` the logger +only puts the record on an in-memory queue, and a background thread owns the real handlers and +writes it. The context fields are captured on the calling thread, so they are still correct. + +```python +from pylogkit import setup_logging, shutdown_logging + +logger = setup_logging("app", to_file=True, use_queue=True) +logger.info("returns without waiting for the file") +shutdown_logging() # waits for queued records to be written +``` + +The queue is unbounded, and records are written a moment after the call, so +`shutdown_logging()` is how you wait for them. It also runs automatically when the interpreter +exits, and it is safe to call more than once. After it, the affected loggers have no handlers until +you call `setup_logging()` again. + ### Environment variables Deployments can change logging without code changes. An argument passed to `setup_logging()` always diff --git a/pylogkit/__init__.py b/pylogkit/__init__.py index db67aa6..b24ee54 100644 --- a/pylogkit/__init__.py +++ b/pylogkit/__init__.py @@ -20,6 +20,7 @@ set_log_context, setup_logging, setup_syslog_logger, + shutdown_logging, tqdm_logging, ) @@ -34,5 +35,6 @@ "set_log_context", "setup_logging", "setup_syslog_logger", + "shutdown_logging", "tqdm_logging", ] diff --git a/pylogkit/log_setup.py b/pylogkit/log_setup.py index 656a1ab..5d07cf1 100644 --- a/pylogkit/log_setup.py +++ b/pylogkit/log_setup.py @@ -24,8 +24,11 @@ from __future__ import annotations +import atexit +import copy import logging import os +import queue import socket import sys import time @@ -34,7 +37,13 @@ from contextlib import contextmanager from contextvars import ContextVar from functools import cache, wraps -from logging.handlers import RotatingFileHandler, SysLogHandler, TimedRotatingFileHandler +from logging.handlers import ( + QueueHandler, + QueueListener, + RotatingFileHandler, + SysLogHandler, + TimedRotatingFileHandler, +) from typing import IO, TYPE_CHECKING, Any, ParamSpec, TypeVar if TYPE_CHECKING: @@ -320,15 +329,62 @@ def with_context(self, **context: Any) -> ContextualLoggerAdapter: return ContextualLoggerAdapter(self.logger, {**(self.extra or {}), **context}) +class _QueueHandler(QueueHandler): + """Queue handler that hands records over without losing information. + + :class:`logging.handlers.QueueHandler` formats the record and drops + ``exc_info``, which would move tracebacks into the message and out of the + JSON ``exc_info`` field. This subclass only resolves the message + arguments, so the listener's handlers format the record as usual. + """ + + def prepare(self, record: logging.LogRecord) -> logging.LogRecord: + """Return a copy of ``record`` with ``msg`` and ``args`` already merged.""" + record = copy.copy(record) + record.msg = record.getMessage() + record.args = None + return record + + +# Listeners that own the real handlers of loggers configured with ``use_queue``. +_listeners: dict[str, QueueListener] = {} +_atexit_registered = False + + +def _stop_listener(logger: logging.Logger) -> None: + """Stop the listener of ``logger``, if any, after it has handled queued records.""" + listener = _listeners.pop(logger.name, None) + if listener is not None: + listener.stop() + for handler in listener.handlers: + handler.close() + + def _reset_logger(logger: logging.Logger) -> None: """Remove and close the handlers ``logger`` has. Closing matters for file handlers: dropping them without closing leaks - open file descriptors every time logging is reconfigured. + open file descriptors every time logging is reconfigured. A queue + listener is stopped, and so flushed, after the logger stops feeding it. """ for handler in list(logger.handlers): logger.removeHandler(handler) handler.close() + _stop_listener(logger) + + +def shutdown_logging() -> None: + """Flush and stop the background logging of every ``use_queue`` logger. + + Records still waiting in a queue are written, then the listener threads + stop and their handlers are closed. The affected loggers are left without + handlers. It is safe to call more than once, and it runs automatically + when the interpreter exits, so call it explicitly only when you need the + output to be complete at a particular point, for example before + ``os._exit()`` or before reading a log file. + """ + for name in list(_listeners): + _reset_logger(logging.getLogger(name)) if JsonFormatter is not None: @@ -455,6 +511,7 @@ def setup_logging( syslog_facility: int = SysLogHandler.LOG_USER, console_stream: IO[str] | None = None, use_emoji: bool = True, + use_queue: bool = False, ) -> logging.Logger: """Configure and return a logger with console, file, JSON and syslog output. @@ -518,6 +575,11 @@ def setup_logging( use_emoji: Show a level emoji after the level name in the verbose console layout. Set it to false for terminals and log viewers that render emoji badly. + use_queue: Hand records to a background thread through a queue, so + that slow handlers (files, syslog) do not block the code that logs. + The queue is unbounded. Records are written when the thread gets + to them, so call :func:`shutdown_logging` to wait for the queue to + drain. That happens automatically at interpreter exit. Returns: The configured :class:`logging.Logger`. @@ -546,11 +608,13 @@ def setup_logging( else: formatter = _build_console_formatter(stream) + handlers: list[logging.Handler] = [] + if to_console: console_handler = logging.StreamHandler(stream) console_handler.setLevel((console_level or level).upper()) console_handler.setFormatter(formatter) - logger.addHandler(console_handler) + handlers.append(console_handler) if to_file and file_path: _prepare_log_file(file_path, overwrite) @@ -566,7 +630,7 @@ def setup_logging( file_handler.setLevel((file_level or level).upper()) file_formatter = SmartFieldFormatter(FILE_FORMAT, datefmt=DATE_FORMAT, no_color=True) file_handler.setFormatter(file_formatter) - logger.addHandler(file_handler) + handlers.append(file_handler) if to_json_file and json_file_path: json_formatter = _build_json_formatter() @@ -576,22 +640,47 @@ def setup_logging( ) json_handler.setLevel((json_level or level).upper()) json_handler.setFormatter(json_formatter) - logger.addHandler(json_handler) + handlers.append(json_handler) if to_syslog: syslog_handler = SysLogHandler(address=syslog_address, facility=syslog_facility) syslog_handler.setLevel((syslog_level or level).upper()) syslog_formatter = SmartFieldFormatter(SYSLOG_FORMAT, no_color=True) syslog_handler.setFormatter(syslog_formatter) - logger.addHandler(syslog_handler) + handlers.append(syslog_handler) context_filter = ContextFilter(use_emoji=use_emoji) - for handler in logger.handlers: - handler.addFilter(context_filter) + if use_queue: + _start_queue(logger, handlers, context_filter) + else: + for handler in handlers: + handler.addFilter(context_filter) + logger.addHandler(handler) return logger +def _start_queue( + logger: logging.Logger, handlers: list[logging.Handler], context_filter: ContextFilter +) -> None: + """Route ``logger`` through a queue drained by a listener thread owning ``handlers``. + + The context filter runs on the queue handler, that is in the thread that + logs, because the listener thread has no access to that thread's context. + """ + global _atexit_registered + log_queue: queue.SimpleQueue[logging.LogRecord] = queue.SimpleQueue() + queue_handler = _QueueHandler(log_queue) + queue_handler.addFilter(context_filter) + listener = QueueListener(log_queue, *handlers, respect_handler_level=True) + listener.start() + _listeners[logger.name] = listener + logger.addHandler(queue_handler) + if not _atexit_registered: + atexit.register(shutdown_logging) + _atexit_registered = True + + def log_exception(logger: LoggerLike, msg: str) -> None: """Log ``msg`` at ERROR level with the current exception's traceback. diff --git a/tests/test_queue.py b/tests/test_queue.py new file mode 100644 index 0000000..9ae4818 --- /dev/null +++ b/tests/test_queue.py @@ -0,0 +1,163 @@ +"""Tests for non-blocking logging through ``use_queue``.""" + +import json +import logging +import threading +import time + +import pytest + +from pylogkit import log_context, log_exception, log_setup, setup_logging, shutdown_logging + + +@pytest.fixture(autouse=True) +def _stop_listeners(): + """Never leave a listener thread behind.""" + yield + shutdown_logging() + + +def test_records_reach_the_file_after_shutdown(tmp_path): + """Everything logged before ``shutdown_logging`` is written, in order.""" + path = tmp_path / "queued.log" + logger = setup_logging( + "q-file", to_console=False, to_file=True, file_path=str(path), use_queue=True + ) + + for i in range(200): + logger.info("message %d", i) + shutdown_logging() + + lines = path.read_text(encoding="utf-8").splitlines() + assert [line.rsplit(" - ", 1)[1] for line in lines] == [f"message {i}" for i in range(200)] + + +def test_handlers_are_owned_by_the_listener(): + """The logger itself only holds the queue handler.""" + logger = setup_logging("q-owner", use_queue=True) + + assert len(logger.handlers) == 1 + assert isinstance(logger.handlers[0], logging.handlers.QueueHandler) + + +def test_context_of_the_logging_thread_is_used(tmp_path): + """The listener thread has no context, so it must be captured when logging.""" + path = tmp_path / "ctx.log" + logger = setup_logging( + "q-ctx", to_console=False, to_file=True, file_path=str(path), use_queue=True + ) + + with log_context(request_id="req-q"): + logger.info("inside") + shutdown_logging() + + assert "[request_id=req-q]" in path.read_text(encoding="utf-8") + + +def test_handler_levels_are_respected(tmp_path): + """A handler with a higher level still filters, as it does without a queue.""" + path = tmp_path / "levels.log" + logger = setup_logging( + "q-level", + to_console=False, + to_file=True, + file_path=str(path), + file_level="ERROR", + use_queue=True, + ) + + logger.info("quiet") + logger.error("loud") + shutdown_logging() + + text = path.read_text(encoding="utf-8") + assert "loud" in text + assert "quiet" not in text + + +@pytest.mark.json_logger +def test_exceptions_keep_their_traceback_field(tmp_path): + """The traceback stays in ``exc_info`` instead of being folded into the message.""" + path = tmp_path / "exc.json.log" + logger = setup_logging( + "q-exc", to_console=False, to_json_file=True, json_file_path=str(path), use_queue=True + ) + + try: + raise ValueError("bad value") + except ValueError: + log_exception(logger, "failed") + shutdown_logging() + + record = json.loads(path.read_text(encoding="utf-8").splitlines()[0]) + assert record["message"] == "failed" + assert "ValueError: bad value" in record["exc_info"] + + +def test_slow_handler_does_not_block_the_caller(tmp_path): + """Logging returns at once while the handler takes its time.""" + release = threading.Event() + handled = [] + + class Slow(logging.Handler): + def emit(self, record): + release.wait(5) + handled.append(record.getMessage()) + + logger = setup_logging("q-slow", to_console=False, use_queue=True) + listener = log_setup._listeners["q-slow"] + listener.handlers = (Slow(),) + + start = time.perf_counter() + logger.info("one") + logger.info("two") + elapsed = time.perf_counter() - start + release.set() + shutdown_logging() + + assert elapsed < 1 + assert handled == ["one", "two"] + + +def test_reconfiguring_stops_the_previous_listener(): + """Calling ``setup_logging`` again must not leak a listener thread.""" + setup_logging("q-again", to_console=False, use_queue=True) + first = log_setup._listeners["q-again"] + + setup_logging("q-again", to_console=False, use_queue=True) + + assert log_setup._listeners["q-again"] is not first + assert first._thread is None + + +def test_switching_the_queue_off_stops_the_listener(): + """A later call without ``use_queue`` returns to direct handlers.""" + setup_logging("q-off", use_queue=True) + + logger = setup_logging("q-off") + + assert "q-off" not in log_setup._listeners + assert isinstance(logger.handlers[0], logging.StreamHandler) + + +def test_shutdown_is_idempotent_and_closes_files(tmp_path): + """Calling it twice is harmless and the file handler is closed.""" + path = tmp_path / "close.log" + setup_logging("q-close", to_console=False, to_file=True, file_path=str(path), use_queue=True) + handler = log_setup._listeners["q-close"].handlers[0] + + shutdown_logging() + shutdown_logging() + + assert handler.stream is None + assert log_setup._listeners == {} + assert logging.getLogger("q-close").handlers == [] + + +def test_logging_without_a_queue_is_unchanged(capsys): + """The default stays synchronous.""" + logger = setup_logging("q-default") + + logger.info("direct") + + assert "direct" in capsys.readouterr().out