Emit task_finished in handlers and use Django's logging

tests/test_executor.py

@@ -1,6 +1,5 @@
 import dataclasses
 import datetime
-import io
 import json
 import logging
 import multiprocessing
@@ -21,6 +20,7 @@
     task,
 )
 from django.tasks.exceptions import TaskResultDoesNotExist
+from django.tasks.signals import task_finished
 from django.utils import timezone
 
 from tests.testapp.tasks import (
@@ -917,6 +917,23 @@ def test_execute_task_result__failed_execution(self):
         assert len(result.errors) == 1
         assert "ValueError" in result.errors[0].exception_class_path
 
+    def test_execute_task_result__sends_task_finished_inside_exception_handler(self):
+        """Send task_finished while receivers can still read the exception."""
+        exception_types = []
+
+        def record_exception_type(sender, task_result, **kwargs):
+            exception_types.append(sys.exc_info()[0])
+
+        task_finished.connect(record_exception_type)
+        try:
+            WorkerThread(
+                worker=_make_worker(), index=0, backend=default_task_backend
+            ).execute_task_result(_task_result(boom))
+        finally:
+            task_finished.disconnect(record_exception_type)
+
+        assert exception_types == [ValueError]
+
     def test_execute_task_result__preserves_worker_ids(self):
         """execute_task_result preserves worker_ids set by acquire."""
         thread = WorkerThread(
@@ -927,33 +944,43 @@ def test_execute_task_result__preserves_worker_ids(self):
         result = thread.execute_task_result(task_result)
         assert result.worker_ids == ["pre-set-worker"]
 
-    def test_execute_task_result__logs_info_records(self, monkeypatch):
-        """Log task start and success records at INFO level."""
-        stream = io.StringIO()
-        monkeypatch.setattr(handler, "stream", stream)
-        configure_logging(JsonFormatter())
-        logging.getLogger().setLevel(logging.INFO)
+    def test_execute_task_result__logs_info_records(self, caplog):
+        """Log task start and finish records via Django's task signals."""
+        caplog.set_level(logging.INFO, logger="django.tasks")
         task_result = _task_result(log_message, "hello")
         WorkerThread(
             worker=_make_worker(), index=0, backend=default_task_backend
         ).execute_task_result(task_result)
 
-        progress_levels = {
-            record["message"]: record["level"]
-            for record in map(json.loads, stream.getvalue().splitlines())
-            if record["logger"] == "multiprocessing"
-        }
-        assert (
-            progress_levels[
-                f"Executing task '{task_result.id}@{log_message.module_path}'"
-            ]
-            == "INFO"
+        django_task_records = [
+            record for record in caplog.records if record.name == "django.tasks"
+        ]
+        assert any(
+            record.levelname == "INFO" and record.getMessage().endswith("state=RUNNING")
+            for record in django_task_records
         )
-        assert (
-            progress_levels[
-                f"Task '{task_result.id}@{log_message.module_path}' succeeded"
-            ]
-            == "INFO"
+        assert any(
+            record.levelname == "INFO"
+            and record.getMessage().endswith("state=SUCCESSFUL")
+            for record in django_task_records
+        )
+
+    def test_execute_task_result__logs_error_record_on_failure(self, caplog):
+        """Log task failure records via Django's task signals."""
+        caplog.set_level(logging.INFO, logger="django.tasks")
+        task_result = _task_result(boom)
+        WorkerThread(
+            worker=_make_worker(), index=0, backend=default_task_backend
+        ).execute_task_result(task_result)
+
+        django_task_records = [
+            record for record in caplog.records if record.name == "django.tasks"
+        ]
+        assert any(
+            record.levelname == "ERROR"
+            and record.getMessage().endswith("state=FAILED")
+            and record.exc_info
+            for record in django_task_records
         )
 
     def test_call_task__calls_function_with_args(self):

threadmill/executor.py

@@ -426,11 +426,6 @@ def run(self) -> None:
 
     def execute_task_result(self, task_result: TaskResult) -> TaskResult:
         """Execute task from task result and update result lifecycle state."""
-        logger.info(
-            "Executing task '%s@%s'",
-            task_result.id,
-            task_result.task.module_path,
-        )
         started_at = timezone.now()
         task_result = dataclasses.replace(
             task_result,
@@ -442,7 +437,7 @@ def execute_task_result(self, task_result: TaskResult) -> TaskResult:
 
         try:
             return_value = WorkerThread.call_task(task_result)
-        except Exception as exception:
+        except Exception as exception:  # noqa: BLE001
             task_result = dataclasses.replace(
                 task_result,
                 status=TaskResultStatus.FAILED,
@@ -452,11 +447,7 @@ def execute_task_result(self, task_result: TaskResult) -> TaskResult:
                 ],
                 finished_at=timezone.now(),
             )
-            logger.exception(
-                "Task '%s@%s' failed",
-                task_result.id,
-                task_result.task.module_path,
-            )
+            task_finished.send(TaskExecutor, task_result=task_result)
         else:
             task_result = dataclasses.replace(
                 task_result,
@@ -466,12 +457,6 @@ def execute_task_result(self, task_result: TaskResult) -> TaskResult:
             object.__setattr__(
                 task_result, "_return_value", normalize_json(return_value)
             )
-            logger.info(
-                "Task '%s@%s' succeeded",
-                task_result.id,
-                task_result.task.module_path,
-            )
-        finally:
             task_finished.send(TaskExecutor, task_result=task_result)
 
         return task_result