Summarise the --workers supervisor's state in its log records

CHANGELOG.md

@@ -4,6 +4,17 @@
 
 ### Added
 
+- **A summary of the `--workers` supervisor's state in its log records.**
+  When every worker is exiting with an error within ten seconds of starting
+  at once, the supervisor logs one `Every worker is failing to start`
+  WARNING, since that is a misconfiguration or a dependency being down
+  rather than one bad worker, and a `Workers recovered` INFO once a worker
+  runs for ten seconds again or exits with code 0. `Supervisor finished`
+  gains `restarts` and `abnormal_restarts` next to the `exit_codes` it
+  already lists. Added records and fields only: no logger name, message,
+  existing field or stdout output changes, so existing projects need no
+  changes.
+  ([#65](https://github.com/tokibito/django-database-task/issues/65))
 - **`pid` and `worker_index` on the worker's log records.** A worker reads
   its index once at startup from the `DJANGO_DATABASE_TASK_WORKER_INDEX`
   environment variable, a whole number of 1 or more, and adds it and its own

README.md

@@ -749,9 +749,21 @@ Shutdown complete: every worker exited.
 - **Restarts.** With `--continuous`, a worker that exits is replaced. One that
   exits with an error within ten seconds of starting is restarted after a
   growing delay, from one second up to thirty, so a misconfiguration does not
-  spin. Without `--continuous` each worker drains the queue once and the
-  command exits when all of them have: the run-once shape of
-  [Running from a job scheduler](#running-from-a-job-scheduler), N at a time.
+  spin. When every worker is failing that way at once, which points at the
+  configuration or a dependency that is down rather than at one bad worker,
+  the supervisor logs one `Every worker is failing to start` WARNING, with
+  `workers` and the last `exit_codes`. It logs one `Workers recovered` INFO
+  once a worker has run for ten seconds again, or has exited with code 0, as
+  a worker that reaches `--max-tasks` sooner does; the record carries
+  `worker_index`, `pid`, `failing_for` in seconds, and `recovered_by`
+  (`uptime` or `clean_exit`). A slot added with `SIGTTIN` does not count
+  until it has done one of the two. `Supervisor finished` carries `restarts`, how
+  many workers were started in place of one that exited, and
+  `abnormal_restarts`, how many of those followed an exit with a code other
+  than 0 or `--empty-exit-code`. Without `--continuous` each worker drains the
+  queue once and the command exits when all of them have: the run-once shape
+  of [Running from a job scheduler](#running-from-a-job-scheduler), N at a
+  time.
 - **`--max-tasks`** goes to each worker. With `--continuous` that is
   gunicorn's `max_requests`: a worker exits after that many tasks and is
   replaced, which bounds a leak in task code.

django_database_task/supervisor.py

@@ -219,6 +219,13 @@ class _Worker:
     retiring: bool = False
     #: The pid of the last process, kept for the exit message.
     pid: int | None = None
+    #: The exit code of the last process.
+    exit_code: int | None = None
+    #: The restart now due follows an abnormal exit; counted when it starts.
+    restart_after_abnormal: bool = False
+    #: When a process last exited with code 0 of its own accord, on the
+    #: monotonic clock.
+    clean_exit_at: float | None = None
 
     @property
     def running(self):
@@ -289,6 +296,11 @@ def __init__(
         self._exit_codes = []
         self._forced = False
         self._pending_scale = 0
+        self._restarts = 0
+        self._abnormal_restarts = 0
+        #: When every worker started failing, on the monotonic clock; None
+        #: while at least one is not.
+        self._failing_since = None
         self._scaling_handlers = {}
         self._shutdown = GracefulShutdown(
             timeout=0,
@@ -366,6 +378,9 @@ def _supervise(self):
         while not self._shutdown.is_set():
             self._apply_scaling()
             self._reap()
+            # Before the restarts, so that a recovery by a clean exit names
+            # the process that exited rather than its replacement.
+            self._check_failing()
             self._start_due()
             if not any(
                 worker.running or worker.restart_at is not None
@@ -432,6 +447,10 @@ def _spawn(self, worker):
             kwargs = {"process_group": 0}
         env = {**os.environ, WORKER_INDEX_ENV: str(worker.index)}
         worker.process = subprocess.Popen(self.args, env=env, **kwargs)
+        if worker.pid is not None:
+            self._restarts += 1
+            if worker.restart_after_abnormal:
+                self._abnormal_restarts += 1
         worker.pid = worker.process.pid
         worker.started_at = time.monotonic()
         worker.restart_at = None
@@ -457,6 +476,10 @@ def _reap(self):
             asked_to_stop = self._shutdown.is_set() or worker.retiring
             stopped = asked_to_stop and _exited_on_signal(code)
             self._exit_codes.append(0 if stopped else code)
+            worker.exit_code = code
+            worker.restart_after_abnormal = self._abnormal(code, stopped)
+            if code == 0 and not stopped:
+                worker.clean_exit_at = now
 
             if worker.retiring:
                 self._workers.remove(worker)
@@ -468,6 +491,71 @@ def _reap(self):
                 worker.restart_at = now + delay
             self._report_exit(worker, code, uptime, delay, stopped)
 
+    def _abnormal(self, code, stopped):
+        return not stopped and code not in (0, self.empty_exit_code)
+
+    def _is_failing(self, worker, now):
+        """
+        True if the slot's last process failed to start and none has since
+        run past ``stable_after``: the slot is waiting out a back-off, or
+        the process it restarted has not yet shown it will stay up.
+        """
+        return worker.failures > 0 and not self._is_stable(worker, now)
+
+    def _is_stable(self, worker, now):
+        """True if the slot's process has run past ``stable_after``."""
+        return worker.running and now - worker.started_at >= self.stable_after
+
+    def _check_failing(self):
+        """
+        Report when every worker is failing to start, and when one recovers.
+
+        One worker that keeps failing is that worker's problem, and its exit
+        records say so. All of them failing at once is a misconfiguration
+        or a dependency that is down, which is worth one record of its own.
+        It is cleared once a worker has run past ``stable_after`` again, or
+        has exited with code 0 since: a worker that reaches ``--max-tasks``
+        sooner than that is working, not failing.
+        """
+        now = time.monotonic()
+        active = [worker for worker in self._workers if not worker.retiring]
+        if self._failing_since is None:
+            if active and all(self._is_failing(worker, now) for worker in active):
+                self._failing_since = now
+                logger.warning(
+                    "Every worker is failing to start: workers=%d",
+                    len(active),
+                    extra={
+                        "workers": len(active),
+                        "exit_codes": [worker.exit_code for worker in active],
+                    },
+                )
+            return
+        for worker in active:
+            if self._is_stable(worker, now):
+                recovered_by = "uptime"
+            elif (
+                worker.clean_exit_at is not None
+                and worker.clean_exit_at >= self._failing_since
+            ):
+                recovered_by = "clean_exit"
+            else:
+                continue
+            failing_for = now - self._failing_since
+            self._failing_since = None
+            logger.info(
+                "Workers recovered: index=%d pid=%d",
+                worker.index,
+                worker.pid,
+                extra={
+                    "worker_index": worker.index,
+                    "pid": worker.pid,
+                    "failing_for": failing_for,
+                    "recovered_by": recovered_by,
+                },
+            )
+            return
+
     def _restart_delay(self, worker, code, uptime):
         if code == 0 or uptime >= self.stable_after:
             worker.failures = 0
@@ -571,7 +659,12 @@ def _finish(self):
         logger.info(
             "Supervisor finished: exit_code=%d",
             code,
-            extra={"exit_code": code, "exit_codes": list(self._exit_codes)},
+            extra={
+                "exit_code": code,
+                "exit_codes": list(self._exit_codes),
+                "restarts": self._restarts,
+                "abnormal_restarts": self._abnormal_restarts,
+            },
         )
         return code
 
@@ -653,7 +746,7 @@ def _report_start(self):
         )
 
     def _report_exit(self, worker, code, uptime, delay, stopped):
-        abnormal = not stopped and code not in (0, self.empty_exit_code)
+        abnormal = self._abnormal(code, stopped)
         message = f"Worker {worker.index} (pid {worker.pid}) exited"
         if stopped:
             message += " on the shutdown signal."