From 640df8430a397a65e12a6ec72ceaf77ccfa10c0c Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Sat, 5 Sep 2026 17:12:47 +0200 Subject: [PATCH 1/6] Log worker records as JSON with a pluggable formatter Replace the hardcoded multiprocessing log formatter with a new JsonFormatter default and a TextFormatter preset. The formatter is swappable via TaskExecutor(log_formatter=...) and the new --log-format json|text worker option; spawned worker processes receive and apply the configured formatter. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- README.md | 24 +++++ tests/test_command.py | 14 +++ tests/test_executor.py | 100 ++++++++++++++++++- threadmill/executor.py | 47 ++++++++- threadmill/management/commands/threadmill.py | 12 ++- 5 files changed, 191 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 87054fe..5e45f04 100644 --- a/README.md +++ b/README.md @@ -97,6 +97,30 @@ All workers will finish the tasks they acquired and acknowledge them. You can use `--exit-empty` to exit immediately after all tasks have been processed, which might be useful for draining a one-off queue. +#### Logging + +Workers log one JSON object per line by default, which is easy to ingest for log collectors: + +```json +{"created_at": "2026-09-05T14:57:25.680105+00:00", "level": "INFO", "logger": "multiprocessing", "message": "Task successful '0198c7bf-8dce-7f7e-9d76-2f3b4a1c5e6d'", "process": 1234, "process_name": "WorkerProcess-1", "thread": "myhost:1234-0"} +``` + +Failed tasks include an `exception` field with the traceback. + +If you prefer human-readable logs, use the text preset: + +```console +uv run manage.py threadmill worker --log-format text +``` + +The formatter remains pluggable. Pass any `logging.Formatter` instance to the executor: + +```python +from threadmill.executor import JsonFormatter, TaskExecutor + +TaskExecutor(backend=backend, log_formatter=JsonFormatter()) +``` + ### Inspector ![Inspector TUI screenshot](https://github.com/codingjoe/threadmill/raw/main/docs/images/TUI-screenshot.svg) diff --git a/tests/test_command.py b/tests/test_command.py index d808520..81585d2 100644 --- a/tests/test_command.py +++ b/tests/test_command.py @@ -7,6 +7,7 @@ from django.tasks import default_task_backend from tests.testapp.tasks import compute_workload, io_workload, memory_workload +from threadmill.executor import TextFormatter, handler from threadmill.management.commands import threadmill @@ -30,6 +31,19 @@ def test_add_arguments__register_all_worker_options(self): assert parsed_arguments.threads == 1 assert parsed_arguments.max_tasks == 0 assert parsed_arguments.max_tasks_jitter == 0 + assert parsed_arguments.log_format == "json" + + def test_call_command__log_format_text(self): + """call_command runs the worker with the text log formatter.""" + call_command( + "threadmill", + "worker", + verbosity=0, + workers=1, + exit_empty=True, + log_format="text", + ) + assert isinstance(handler.formatter, TextFormatter) @pytest.mark.benchmark def test_call_command__benchmark_compute( diff --git a/tests/test_executor.py b/tests/test_executor.py index b7af974..8a82cc9 100644 --- a/tests/test_executor.py +++ b/tests/test_executor.py @@ -1,5 +1,6 @@ import dataclasses import datetime +import json import logging import multiprocessing import sys @@ -27,7 +28,15 @@ echo, ) from threadmill.backends.base import Broker -from threadmill.executor import TaskExecutor, WorkerProcess, WorkerThread +from threadmill.executor import ( + JsonFormatter, + TaskExecutor, + TextFormatter, + WorkerProcess, + WorkerThread, + handler, + set_log_formatter, +) @task(queue_name="default") @@ -73,6 +82,83 @@ def _make_worker(*, max_tasks: int | None = None) -> WorkerProcess: ) +class TestJsonFormatter: + """Tests for the JsonFormatter class.""" + + def test_format__returns_json_payload(self) -> None: + """Return a JSON object with structured record fields.""" + record = logging.LogRecord( + "multiprocessing", + logging.INFO, + __file__, + 1, + "Task successful %r", + ("abc",), + None, + ) + payload = json.loads(JsonFormatter().format(record)) + assert set(payload) == { + "created_at", + "level", + "logger", + "message", + "process", + "process_name", + "thread", + } + assert payload["message"] == "Task successful 'abc'" + assert payload["level"] == "INFO" + assert payload["logger"] == "multiprocessing" + assert payload["thread"] == "MainThread" + assert datetime.datetime.fromisoformat(payload["created_at"]).tzinfo is not None + + def test_format__includes_exception_traceback(self) -> None: + """Include the exception traceback when the record carries exc_info.""" + try: + raise ValueError("boom") + except ValueError: + record = logging.LogRecord( + "multiprocessing", + logging.ERROR, + __file__, + 1, + "Task failed %r", + ("abc",), + sys.exc_info(), + ) + payload = json.loads(JsonFormatter().format(record)) + assert "ValueError: boom" in payload["exception"] + + +class TestTextFormatter: + """Tests for the TextFormatter class.""" + + def test_format__returns_text_record(self) -> None: + """Return a human-readable single-line record.""" + record = logging.LogRecord( + "multiprocessing", + logging.INFO, + __file__, + 1, + "Task successful %r", + ("abc",), + None, + ) + formatted = TextFormatter().format(record) + assert formatted.startswith("INFO: ") + assert formatted.endswith(" - Task successful 'abc'") + + +class TestSetLogFormatter: + """Tests for the set_log_formatter function.""" + + def test_set_log_formatter__sets_handler_formatter(self) -> None: + """set_log_formatter swaps the formatter on the shared log handler.""" + set_log_formatter(TextFormatter()) + assert isinstance(handler.formatter, TextFormatter) + set_log_formatter(JsonFormatter()) + + class TestTaskExecutor: """Tests for the TaskExecutor dataclass and its methods.""" @@ -132,6 +218,7 @@ def test_create_worker_process__starts_worker(self): executor = TaskExecutor(backend=default_task_backend, queues=("default",)) worker = executor.create_worker_process() assert worker.is_alive() + assert worker.log_formatter is executor.log_formatter worker.shutdown() def test_run__processes_enqueued_tasks_end_to_end(self): @@ -148,6 +235,7 @@ def test_run__processes_enqueued_tasks_end_to_end(self): run_thread = threading.Thread(target=executor.run, daemon=True) run_thread.start() time.sleep(2) + assert handler.formatter is executor.log_formatter executor.shutdown() run_thread.join(timeout=5) assert not run_thread.is_alive() @@ -286,6 +374,16 @@ def test_shutdown_requested__is_settable(self): worker.shutdown_requested.set() assert worker.shutdown_requested.is_set() + def test_run__applies_log_formatter_and_stops(self): + """run() applies the log formatter and returns when shutdown is requested.""" + worker = _make_worker() + worker.shutdown_requested.set() + run_thread = threading.Thread(target=worker.run) + run_thread.start() + run_thread.join(timeout=5) + assert not run_thread.is_alive() + assert handler.formatter is worker.log_formatter + class TestWorkerThread: """Tests for the WorkerThread class.""" diff --git a/threadmill/executor.py b/threadmill/executor.py index f9490b3..0d7d75a 100644 --- a/threadmill/executor.py +++ b/threadmill/executor.py @@ -3,6 +3,7 @@ import asyncio import dataclasses import datetime +import json import logging import multiprocessing import random @@ -25,16 +26,48 @@ if typing.TYPE_CHECKING: from .backends.base import Broker, ThreadmillTaskBackend + +class JsonFormatter(logging.Formatter): + """Format log records as single-line JSON objects.""" + + def format(self, record: logging.LogRecord) -> str: + """Return the record as a JSON object with structured fields.""" + payload = { + "created_at": datetime.datetime.fromtimestamp( + record.created, tz=datetime.UTC + ).isoformat(), + "level": record.levelname, + "logger": record.name, + "message": record.getMessage(), + "process": record.process, + "process_name": record.processName, + "thread": record.threadName, + } + if record.exc_info: + payload["exception"] = "".join(format_exception(*record.exc_info)) + return json.dumps(payload) + + +class TextFormatter(logging.Formatter): + """Format log records as single-line human-readable text.""" + + def __init__(self) -> None: + """Initialize the formatter with the default text format.""" + super().__init__("%(levelname)s: %(asctime)s - pid=%(process)s - %(message)s") + + logger = multiprocessing.get_logger() -formatter = logging.Formatter( - "%(levelname)s: %(asctime)s - pid=%(process)s - %(message)s" -) handler = logging.StreamHandler() -handler.setFormatter(formatter) +handler.setFormatter(JsonFormatter()) logger.addHandler(handler) logger.setLevel(logging.INFO) +def set_log_formatter(formatter: logging.Formatter) -> None: + """Set the formatter for records logged by the multiprocessing logger.""" + handler.setFormatter(formatter) + + @dataclasses.dataclass(kw_only=True, slots=True) class TaskExecutor: """Tasks consumed from shared joinable queues via process and thread pools.""" @@ -53,6 +86,7 @@ class TaskExecutor: queues: tuple[str] broker: Broker | None = dataclasses.field(default=None, init=False) exit_empty: bool = False + log_formatter: logging.Formatter = dataclasses.field(default_factory=JsonFormatter) def __post_init__(self) -> None: """Initialize derived orchestration fields and queues.""" @@ -74,12 +108,14 @@ def create_worker_process(self) -> WorkerProcess: self.backend.alias, self.queues, self.exit_empty, + self.log_formatter, ) worker.start() return worker def run(self) -> None: """Start consuming tasks until shutdown is requested.""" + set_log_formatter(self.log_formatter) self.worker_processes = [ self.create_worker_process() for _ in range(self.process_count) ] @@ -132,6 +168,7 @@ def __init__( backend_alias: str = "", queues: tuple[str, ...] = (), exit_empty: bool = False, + log_formatter: logging.Formatter | None = None, ) -> None: """Create process with dedicated thread pool for task execution.""" self.shutdown_requested = multiprocessing.Event() @@ -141,12 +178,14 @@ def __init__( self.backend_alias = backend_alias self.queues = queues self.exit_empty = exit_empty + self.log_formatter = log_formatter or JsonFormatter() self.task_count = 0 self.lock: threading.Lock | None = None self.expired: threading.Event | None = None def run(self) -> None: """Start consumer execution inside this process.""" + set_log_formatter(self.log_formatter) django.setup() logger.info("Starting worker process %s", self.name) self.lock = threading.Lock() diff --git a/threadmill/management/commands/threadmill.py b/threadmill/management/commands/threadmill.py index c1001d6..d93436a 100644 --- a/threadmill/management/commands/threadmill.py +++ b/threadmill/management/commands/threadmill.py @@ -10,7 +10,9 @@ task_backends, ) -from ...executor import TaskExecutor +from ...executor import JsonFormatter, TaskExecutor, TextFormatter + +log_formatters = {"json": JsonFormatter, "text": TextFormatter} def kill_softly(signum, frame): @@ -71,6 +73,12 @@ def add_arguments(self, parser): action="store_true", help="Drain the task queue and exit with 0.", ) + parser.add_argument( + "--log-format", + choices=tuple(log_formatters), + default="json", + help="Format for worker log records. Defaults to JSON.", + ) def handle( self, @@ -83,6 +91,7 @@ def handle( max_tasks, max_tasks_jitter, exit_empty, + log_format, **options, ): match sys.platform: @@ -109,6 +118,7 @@ def handle( max_tasks_jitter=max_tasks_jitter, exit_empty=exit_empty, queues=queues, + log_formatter=log_formatters[log_format](), ) try: exe.run() From fe5c8df6fdfc871745dc6b8173616d93ddee41e6 Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Mon, 7 Sep 2026 10:43:25 +0200 Subject: [PATCH 2/6] Accept a logging format string for --log-format Replace the json|text presets with a free-form logging format string: --log-format '%(levelname)s %(message)s' wraps the string in logging.Formatter, while omitting the option keeps the JSON default. Invalid format strings fail fast with CommandError via a smoke-test emit on a real LogRecord, before any handler is touched. TaskExecutor.log_formatter remains the single owner of the JSON default; WorkerProcess now requires the formatter as a keyword-only argument and applies it directly on the shared handler. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- README.md | 24 ------- tests/test_command.py | 67 ++++++++++++++++++-- tests/test_executor.py | 32 +--------- tests/test_signals.py | 12 +++- threadmill/executor.py | 35 ++++------ threadmill/management/commands/threadmill.py | 25 ++++++-- 6 files changed, 101 insertions(+), 94 deletions(-) diff --git a/README.md b/README.md index 5e45f04..87054fe 100644 --- a/README.md +++ b/README.md @@ -97,30 +97,6 @@ All workers will finish the tasks they acquired and acknowledge them. You can use `--exit-empty` to exit immediately after all tasks have been processed, which might be useful for draining a one-off queue. -#### Logging - -Workers log one JSON object per line by default, which is easy to ingest for log collectors: - -```json -{"created_at": "2026-09-05T14:57:25.680105+00:00", "level": "INFO", "logger": "multiprocessing", "message": "Task successful '0198c7bf-8dce-7f7e-9d76-2f3b4a1c5e6d'", "process": 1234, "process_name": "WorkerProcess-1", "thread": "myhost:1234-0"} -``` - -Failed tasks include an `exception` field with the traceback. - -If you prefer human-readable logs, use the text preset: - -```console -uv run manage.py threadmill worker --log-format text -``` - -The formatter remains pluggable. Pass any `logging.Formatter` instance to the executor: - -```python -from threadmill.executor import JsonFormatter, TaskExecutor - -TaskExecutor(backend=backend, log_formatter=JsonFormatter()) -``` - ### Inspector ![Inspector TUI screenshot](https://github.com/codingjoe/threadmill/raw/main/docs/images/TUI-screenshot.svg) diff --git a/tests/test_command.py b/tests/test_command.py index 81585d2..08fbed1 100644 --- a/tests/test_command.py +++ b/tests/test_command.py @@ -1,4 +1,6 @@ import argparse +import logging +import re import signal from unittest.mock import patch @@ -7,10 +9,17 @@ from django.tasks import default_task_backend from tests.testapp.tasks import compute_workload, io_workload, memory_workload -from threadmill.executor import TextFormatter, handler +from threadmill.executor import JsonFormatter, handler from threadmill.management.commands import threadmill +@pytest.fixture(autouse=True) +def restore_log_formatter(): + """Restore the default JSON log formatter after each test.""" + yield + handler.setFormatter(JsonFormatter()) + + class TestKillSoftly: def test_kill_softly__raise_keyboard_interrupt_with_signal_name(self): """Raise KeyboardInterrupt with signal metadata in message.""" @@ -31,19 +40,65 @@ def test_add_arguments__register_all_worker_options(self): assert parsed_arguments.threads == 1 assert parsed_arguments.max_tasks == 0 assert parsed_arguments.max_tasks_jitter == 0 - assert parsed_arguments.log_format == "json" + assert parsed_arguments.log_format is None + + def test_call_command__log_format(self): + """Run the worker with the given log format string.""" + call_command( + "threadmill", + "worker", + verbosity=0, + workers=1, + exit_empty=True, + log_format="%(levelname)s %(message)s", + ) + record = logging.LogRecord( + "threadmill", logging.INFO, __file__, 1, "Hello %s", ("world",), None + ) + assert handler.formatter.format(record) == "INFO Hello world" + + @pytest.mark.parametrize("log_format", ["100% done", "%(message)s 100% done", 123]) + def test_call_command__log_format__raise_command_error(self, log_format): + """Raise CommandError for a log format that fails to format a record.""" + with pytest.raises( + CommandError, match=re.escape(f"Invalid log format: {log_format!r}") + ): + call_command( + "threadmill", + "worker", + verbosity=0, + workers=1, + exit_empty=True, + log_format=log_format, + ) - def test_call_command__log_format_text(self): - """call_command runs the worker with the text log formatter.""" + def test_call_command__log_format__defaults_to_json(self): + """Default to the JSON log formatter.""" + handler.setFormatter(logging.Formatter("%(message)s")) call_command( "threadmill", "worker", verbosity=0, workers=1, exit_empty=True, - log_format="text", ) - assert isinstance(handler.formatter, TextFormatter) + assert isinstance(handler.formatter, JsonFormatter) + + def test_call_command__log_format__empty_string(self): + """Honor an explicitly empty log format string.""" + handler.setFormatter(JsonFormatter()) + call_command( + "threadmill", + "worker", + verbosity=0, + workers=1, + exit_empty=True, + log_format="", + ) + record = logging.LogRecord( + "threadmill", logging.INFO, __file__, 1, "Hello %s", ("world",), None + ) + assert handler.formatter.format(record) == "Hello world" @pytest.mark.benchmark def test_call_command__benchmark_compute( diff --git a/tests/test_executor.py b/tests/test_executor.py index 8a82cc9..74e1061 100644 --- a/tests/test_executor.py +++ b/tests/test_executor.py @@ -31,11 +31,9 @@ from threadmill.executor import ( JsonFormatter, TaskExecutor, - TextFormatter, WorkerProcess, WorkerThread, handler, - set_log_formatter, ) @@ -79,6 +77,7 @@ def _make_worker(*, max_tasks: int | None = None) -> WorkerProcess: max_tasks=max_tasks, backend_alias="default", queues=("default",), + log_formatter=JsonFormatter(), ) @@ -130,35 +129,6 @@ def test_format__includes_exception_traceback(self) -> None: assert "ValueError: boom" in payload["exception"] -class TestTextFormatter: - """Tests for the TextFormatter class.""" - - def test_format__returns_text_record(self) -> None: - """Return a human-readable single-line record.""" - record = logging.LogRecord( - "multiprocessing", - logging.INFO, - __file__, - 1, - "Task successful %r", - ("abc",), - None, - ) - formatted = TextFormatter().format(record) - assert formatted.startswith("INFO: ") - assert formatted.endswith(" - Task successful 'abc'") - - -class TestSetLogFormatter: - """Tests for the set_log_formatter function.""" - - def test_set_log_formatter__sets_handler_formatter(self) -> None: - """set_log_formatter swaps the formatter on the shared log handler.""" - set_log_formatter(TextFormatter()) - assert isinstance(handler.formatter, TextFormatter) - set_log_formatter(JsonFormatter()) - - class TestTaskExecutor: """Tests for the TaskExecutor dataclass and its methods.""" diff --git a/tests/test_signals.py b/tests/test_signals.py index e94b289..2a3ed9d 100644 --- a/tests/test_signals.py +++ b/tests/test_signals.py @@ -10,7 +10,12 @@ from django.utils import timezone from tests.testapp.tasks import echo -from threadmill.executor import TaskExecutor, WorkerProcess, WorkerThread +from threadmill.executor import ( + JsonFormatter, + TaskExecutor, + WorkerProcess, + WorkerThread, +) def _task_result(task, *args) -> TaskResult: @@ -75,7 +80,10 @@ def test_execute_task_result__closes_stale_connection(self, stale_connection): stale_connection, "close", wraps=stale_connection.close ) as spy: worker = WorkerProcess( - thread_count=1, backend_alias="default", queues=("default",) + thread_count=1, + backend_alias="default", + queues=("default",), + log_formatter=JsonFormatter(), ) thread = WorkerThread(worker=worker, index=0, backend=None) diff --git a/threadmill/executor.py b/threadmill/executor.py index 0d7d75a..80474b1 100644 --- a/threadmill/executor.py +++ b/threadmill/executor.py @@ -31,7 +31,6 @@ class JsonFormatter(logging.Formatter): """Format log records as single-line JSON objects.""" def format(self, record: logging.LogRecord) -> str: - """Return the record as a JSON object with structured fields.""" payload = { "created_at": datetime.datetime.fromtimestamp( record.created, tz=datetime.UTC @@ -48,14 +47,6 @@ def format(self, record: logging.LogRecord) -> str: return json.dumps(payload) -class TextFormatter(logging.Formatter): - """Format log records as single-line human-readable text.""" - - def __init__(self) -> None: - """Initialize the formatter with the default text format.""" - super().__init__("%(levelname)s: %(asctime)s - pid=%(process)s - %(message)s") - - logger = multiprocessing.get_logger() handler = logging.StreamHandler() handler.setFormatter(JsonFormatter()) @@ -63,11 +54,6 @@ def __init__(self) -> None: logger.setLevel(logging.INFO) -def set_log_formatter(formatter: logging.Formatter) -> None: - """Set the formatter for records logged by the multiprocessing logger.""" - handler.setFormatter(formatter) - - @dataclasses.dataclass(kw_only=True, slots=True) class TaskExecutor: """Tasks consumed from shared joinable queues via process and thread pools.""" @@ -103,19 +89,19 @@ def get_maximum_tasks_per_child(self) -> int | None: def create_worker_process(self) -> WorkerProcess: """Create and start a new worker process.""" worker = WorkerProcess( - self.thread_count, - self.get_maximum_tasks_per_child(), - self.backend.alias, - self.queues, - self.exit_empty, - self.log_formatter, + thread_count=self.thread_count, + max_tasks=self.get_maximum_tasks_per_child(), + backend_alias=self.backend.alias, + queues=self.queues, + exit_empty=self.exit_empty, + log_formatter=self.log_formatter, ) worker.start() return worker def run(self) -> None: """Start consuming tasks until shutdown is requested.""" - set_log_formatter(self.log_formatter) + handler.setFormatter(self.log_formatter) self.worker_processes = [ self.create_worker_process() for _ in range(self.process_count) ] @@ -163,12 +149,13 @@ class WorkerProcess(multiprocessing.Process): def __init__( self, + *, thread_count: int, max_tasks: int | None = None, backend_alias: str = "", queues: tuple[str, ...] = (), exit_empty: bool = False, - log_formatter: logging.Formatter | None = None, + log_formatter: logging.Formatter, ) -> None: """Create process with dedicated thread pool for task execution.""" self.shutdown_requested = multiprocessing.Event() @@ -178,14 +165,14 @@ def __init__( self.backend_alias = backend_alias self.queues = queues self.exit_empty = exit_empty - self.log_formatter = log_formatter or JsonFormatter() + self.log_formatter = log_formatter self.task_count = 0 self.lock: threading.Lock | None = None self.expired: threading.Event | None = None def run(self) -> None: """Start consumer execution inside this process.""" - set_log_formatter(self.log_formatter) + handler.setFormatter(self.log_formatter) django.setup() logger.info("Starting worker process %s", self.name) self.lock = threading.Lock() diff --git a/threadmill/management/commands/threadmill.py b/threadmill/management/commands/threadmill.py index d93436a..7c40219 100644 --- a/threadmill/management/commands/threadmill.py +++ b/threadmill/management/commands/threadmill.py @@ -1,3 +1,4 @@ +import logging import signal import sys @@ -10,9 +11,7 @@ task_backends, ) -from ...executor import JsonFormatter, TaskExecutor, TextFormatter - -log_formatters = {"json": JsonFormatter, "text": TextFormatter} +from ...executor import JsonFormatter, TaskExecutor def kill_softly(signum, frame): @@ -75,9 +74,10 @@ def add_arguments(self, parser): ) parser.add_argument( "--log-format", - choices=tuple(log_formatters), - default="json", - help="Format for worker log records. Defaults to JSON.", + help=( + "Logging format string for worker log records, e.g." + " '%%(levelname)s %%(message)s'. Defaults to JSON." + ), ) def handle( @@ -110,6 +110,17 @@ def handle( raise CommandError( f"Backend does not support all specified queues: {_non_queues!r}" ) + try: + log_formatter = ( + JsonFormatter() if log_format is None else logging.Formatter(log_format) + ) + log_formatter.format( + logging.LogRecord( + "threadmill", logging.INFO, __file__, 1, "Ready", (), None + ) + ) + except (TypeError, ValueError) as e: + raise CommandError(f"Invalid log format: {log_format!r}") from e exe = TaskExecutor( backend=backend, workers=workers, @@ -118,7 +129,7 @@ def handle( max_tasks_jitter=max_tasks_jitter, exit_empty=exit_empty, queues=queues, - log_formatter=log_formatters[log_format](), + log_formatter=log_formatter, ) try: exe.run() From 91fa36a0bee6b3a1b4aa2e1b362d8593089155c0 Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Mon, 7 Sep 2026 11:59:03 +0200 Subject: [PATCH 3/6] Use DjangoJSONEncoder and the Django default timezone Serialize log records with DjangoJSONEncoder instead of casting the datetime to an ISO string, and stamp records with Django's current timezone rather than hardcoded UTC. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- tests/test_executor.py | 7 ++++++- threadmill/executor.py | 7 ++++--- 2 files changed, 10 insertions(+), 4 deletions(-) diff --git a/tests/test_executor.py b/tests/test_executor.py index 74e1061..909dcbd 100644 --- a/tests/test_executor.py +++ b/tests/test_executor.py @@ -109,7 +109,12 @@ def test_format__returns_json_payload(self) -> None: assert payload["level"] == "INFO" assert payload["logger"] == "multiprocessing" assert payload["thread"] == "MainThread" - assert datetime.datetime.fromisoformat(payload["created_at"]).tzinfo is not None + assert ( + datetime.datetime.fromisoformat(payload["created_at"]).utcoffset() + == datetime.datetime.fromtimestamp( + record.created, tz=timezone.get_current_timezone() + ).utcoffset() + ) def test_format__includes_exception_traceback(self) -> None: """Include the exception traceback when the record carries exc_info.""" diff --git a/threadmill/executor.py b/threadmill/executor.py index 80474b1..81cd811 100644 --- a/threadmill/executor.py +++ b/threadmill/executor.py @@ -17,6 +17,7 @@ from traceback import format_exception import django +from django.core.serializers.json import DjangoJSONEncoder from django.tasks import TaskResult, task_backends from django.tasks.base import TaskContext, TaskError, TaskResultStatus from django.tasks.signals import task_finished, task_started @@ -33,8 +34,8 @@ class JsonFormatter(logging.Formatter): def format(self, record: logging.LogRecord) -> str: payload = { "created_at": datetime.datetime.fromtimestamp( - record.created, tz=datetime.UTC - ).isoformat(), + record.created, tz=timezone.get_current_timezone() + ), "level": record.levelname, "logger": record.name, "message": record.getMessage(), @@ -44,7 +45,7 @@ def format(self, record: logging.LogRecord) -> str: } if record.exc_info: payload["exception"] = "".join(format_exception(*record.exc_info)) - return json.dumps(payload) + return json.dumps(payload, cls=DjangoJSONEncoder) logger = multiprocessing.get_logger() From bfa386b0dcfa962bc2874337aa6909dd522e76c6 Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Mon, 7 Sep 2026 12:03:28 +0200 Subject: [PATCH 4/6] Include extra record attributes in JSON log records Merge non-standard LogRecord attributes (set via logging's extra={}) into the JSON payload as top-level fields. Values must be JSON-serializable, matching how normalize_json treats task return values. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- tests/test_executor.py | 17 +++++++++++++++++ threadmill/executor.py | 35 +++++++++++++++++++++++++++++++++++ 2 files changed, 52 insertions(+) diff --git a/tests/test_executor.py b/tests/test_executor.py index 909dcbd..298a4df 100644 --- a/tests/test_executor.py +++ b/tests/test_executor.py @@ -116,6 +116,23 @@ def test_format__returns_json_payload(self) -> None: ).utcoffset() ) + def test_format__includes_extra_attributes(self) -> None: + """Include extra record attributes in the JSON payload.""" + record = logging.LogRecord( + "multiprocessing", + logging.INFO, + __file__, + 1, + "Task successful %r", + ("abc",), + None, + ) + record.request_id = "abc" + record.duration_ms = 42 + payload = json.loads(JsonFormatter().format(record)) + assert payload["request_id"] == "abc" + assert payload["duration_ms"] == 42 + def test_format__includes_exception_traceback(self) -> None: """Include the exception traceback when the record carries exc_info.""" try: diff --git a/threadmill/executor.py b/threadmill/executor.py index 81cd811..bbc1194 100644 --- a/threadmill/executor.py +++ b/threadmill/executor.py @@ -31,6 +31,34 @@ class JsonFormatter(logging.Formatter): """Format log records as single-line JSON objects.""" + standard_attributes = frozenset( + { + "args", + "asctime", + "created", + "exc_info", + "exc_text", + "filename", + "funcName", + "levelname", + "levelno", + "lineno", + "message", + "module", + "msecs", + "msg", + "name", + "pathname", + "process", + "processName", + "relativeCreated", + "stack_info", + "taskName", + "thread", + "threadName", + } + ) + def format(self, record: logging.LogRecord) -> str: payload = { "created_at": datetime.datetime.fromtimestamp( @@ -43,6 +71,13 @@ def format(self, record: logging.LogRecord) -> str: "process_name": record.processName, "thread": record.threadName, } + payload.update( + { + key: value + for key, value in record.__dict__.items() + if key not in self.standard_attributes + } + ) if record.exc_info: payload["exception"] = "".join(format_exception(*record.exc_info)) return json.dumps(payload, cls=DjangoJSONEncoder) From 7e935b1f5ab643158d0938f1d61f3cf3a6328535 Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Mon, 7 Sep 2026 12:14:15 +0200 Subject: [PATCH 5/6] Merge extra record attributes via dict unpacking Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- threadmill/executor.py | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/threadmill/executor.py b/threadmill/executor.py index bbc1194..446349c 100644 --- a/threadmill/executor.py +++ b/threadmill/executor.py @@ -70,14 +70,12 @@ def format(self, record: logging.LogRecord) -> str: "process": record.process, "process_name": record.processName, "thread": record.threadName, - } - payload.update( - { + **{ key: value for key, value in record.__dict__.items() if key not in self.standard_attributes - } - ) + }, + } if record.exc_info: payload["exception"] = "".join(format_exception(*record.exc_info)) return json.dumps(payload, cls=DjangoJSONEncoder) From 47855e03e8c783a5c4fffc220f7f87e750d37722 Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Mon, 7 Sep 2026 12:15:41 +0200 Subject: [PATCH 6/6] Merge extra record attributes with the dict merge operator Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- threadmill/executor.py | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/threadmill/executor.py b/threadmill/executor.py index 446349c..0f0bbb1 100644 --- a/threadmill/executor.py +++ b/threadmill/executor.py @@ -70,11 +70,10 @@ def format(self, record: logging.LogRecord) -> str: "process": record.process, "process_name": record.processName, "thread": record.threadName, - **{ - key: value - for key, value in record.__dict__.items() - if key not in self.standard_attributes - }, + } | { + key: value + for key, value in record.__dict__.items() + if key not in self.standard_attributes } if record.exc_info: payload["exception"] = "".join(format_exception(*record.exc_info))