From 7ebb6c0b12846aee11956e785318a93bbadc621e Mon Sep 17 00:00:00 2001 From: Shinya Okano Date: Fri, 9 Oct 2026 12:17:32 +0900 Subject: [PATCH 1/2] Summarise the --workers supervisor's state in its log records Each restart was one record, with nothing that said the whole fleet was down. The supervisor now logs one "Every worker is failing to start" WARNING when every slot is failing within the stable window at once, and a "Workers recovered" INFO once a worker runs past the window again. "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. Refs #65 --- CHANGELOG.md | 10 ++ README.md | 15 ++- django_database_task/supervisor.py | 81 +++++++++++++- tests/test_supervisor.py | 163 +++++++++++++++++++++++++++++ 4 files changed, 264 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 050e888..32d6d6c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,16 @@ ### Added +- **A summary of the `--workers` supervisor's state in its log records.** + When every worker is failing within the stable window 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 past the window + again. `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 diff --git a/README.md b/README.md index 1a0db46..6c62c95 100644 --- a/README.md +++ b/README.md @@ -749,9 +749,18 @@ 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`, and a `Workers recovered` INFO, with + `worker_index`, `pid` and `failing_for` in seconds, once a worker has run + past the ten seconds again. `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. diff --git a/django_database_task/supervisor.py b/django_database_task/supervisor.py index a9c02df..ce7186d 100644 --- a/django_database_task/supervisor.py +++ b/django_database_task/supervisor.py @@ -219,6 +219,10 @@ 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 @property def running(self): @@ -289,6 +293,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, @@ -367,6 +376,7 @@ def _supervise(self): self._apply_scaling() self._reap() self._start_due() + self._check_failing() if not any( worker.running or worker.restart_at is not None for worker in self._workers @@ -432,6 +442,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 +471,8 @@ 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 worker.retiring: self._workers.remove(worker) @@ -468,6 +484,62 @@ 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. + """ + 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 + recovered = next( + (worker for worker in active if self._is_stable(worker, now)), None + ) + if recovered is None: + return + failing_for = now - self._failing_since + self._failing_since = None + logger.info( + "Workers recovered: index=%d pid=%d", + recovered.index, + recovered.pid, + extra={ + "worker_index": recovered.index, + "pid": recovered.pid, + "failing_for": failing_for, + }, + ) + def _restart_delay(self, worker, code, uptime): if code == 0 or uptime >= self.stable_after: worker.failures = 0 @@ -571,7 +643,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 +730,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." diff --git a/tests/test_supervisor.py b/tests/test_supervisor.py index b017bb9..c0f9669 100644 --- a/tests/test_supervisor.py +++ b/tests/test_supervisor.py @@ -102,6 +102,45 @@ def handler(signum, frame): ) +# Exits with 1 for its first argv[3] launches, counted by the ready files, +# then stays up until signalled. +FAILS_THEN_STAYS_UP_WORKER = ( + PRELUDE + + """ +mark(ready) +if len(os.listdir(ready)) <= int(sys.argv[3]): + sys.exit(1) +asked = False +def handler(signum, frame): + global asked + asked = True +install(handler) +while not asked: + time.sleep(0.02) +""" +) + +# The first worker to start stays up until signalled; every other one exits +# with 1. +ONE_STAYS_UP_WORKER = ( + PRELUDE + + """ +mark(ready) +try: + os.close(os.open(os.path.join(stopped, "first"), os.O_CREAT | os.O_EXCL)) +except FileExistsError: + sys.exit(1) +asked = False +def handler(signum, frame): + global asked + asked = True +install(handler) +while not asked: + time.sleep(0.02) +""" +) + + def wait_for(condition, timeout=20, what="condition"): deadline = time.monotonic() + timeout while time.monotonic() < deadline: @@ -386,6 +425,130 @@ def test_the_supervisors_own_environment_is_left_alone(self, tmp_path, monkeypat assert WORKER_INDEX_ENV not in os.environ +def messages_starting(caplog, prefix): + return [r for r in caplog.records if r.getMessage().startswith(prefix)] + + +class TestSupervisorState: + def test_every_worker_failing_is_reported_once(self, tmp_path, caplog): + harness = Harness( + tmp_path, + EXITING_WORKER, + 1, + workers=2, + restart=True, + backoff_start=0.05, + backoff_max=0.1, + stable_after=60, + ) + with caplog.at_level(logging.INFO, logger="django_database_task"): + harness.start() + harness.wait_exits(6) + harness.supervisor.request_shutdown() + harness.join() + + (record,) = messages_starting(caplog, "Every worker is failing to start") + assert record.levelno == logging.WARNING + assert record.workers == 2 + assert record.exit_codes == [1, 1] + assert not messages_starting(caplog, "Workers recovered") + + def test_one_worker_failing_is_not_every_worker(self, tmp_path, caplog): + harness = Harness( + tmp_path, + ONE_STAYS_UP_WORKER, + workers=2, + restart=True, + backoff_start=0.05, + backoff_max=0.1, + stable_after=60, + ) + with caplog.at_level(logging.INFO, logger="django_database_task"): + harness.start() + harness.wait_exits(4) + harness.supervisor.request_shutdown() + harness.join() + + assert not messages_starting(caplog, "Every worker is failing to start") + + def test_a_worker_past_the_stable_window_clears_it(self, tmp_path, caplog): + harness = Harness( + tmp_path, + FAILS_THEN_STAYS_UP_WORKER, + 2, + workers=1, + restart=True, + backoff_start=0.05, + backoff_max=0.1, + stable_after=0.5, + ) + with caplog.at_level(logging.INFO, logger="django_database_task"): + harness.start() + wait_for( + lambda: messages_starting(caplog, "Workers recovered"), + what="the recovery record", + ) + (running,) = harness.supervisor.running_pids + harness.supervisor.request_shutdown() + harness.join() + + (failing,) = messages_starting(caplog, "Every worker is failing to start") + (recovered,) = messages_starting(caplog, "Workers recovered") + assert caplog.records.index(failing) < caplog.records.index(recovered) + assert recovered.levelno == logging.INFO + assert recovered.worker_index == 1 + assert recovered.pid == running + assert recovered.failing_for >= 0.5 + + def test_finish_counts_the_restarts(self, tmp_path, caplog): + harness = Harness( + tmp_path, + EXITING_WORKER, + 1, + workers=1, + restart=True, + backoff_start=0.05, + backoff_max=0.05, + stable_after=60, + ) + with caplog.at_level(logging.INFO, logger="django_database_task"): + harness.start() + harness.wait_exits(3) + harness.supervisor.request_shutdown() + harness.join() + + (record,) = messages_starting(caplog, "Supervisor finished") + started = messages_starting(caplog, "Worker process started") + assert record.restarts == len(started) - 1 + assert record.restarts >= 2 + assert record.abnormal_restarts == record.restarts + + def test_a_clean_exit_is_not_an_abnormal_restart(self, tmp_path, caplog): + harness = Harness( + tmp_path, EXITING_WORKER, 0, workers=1, restart=True, stable_after=60 + ) + with caplog.at_level(logging.INFO, logger="django_database_task"): + harness.start() + harness.wait_exits(3) + harness.supervisor.request_shutdown() + harness.join() + + (record,) = messages_starting(caplog, "Supervisor finished") + assert record.restarts >= 2 + assert record.abnormal_restarts == 0 + assert not messages_starting(caplog, "Every worker is failing to start") + + def test_a_run_once_supervisor_restarts_nothing(self, tmp_path, caplog): + harness = Harness(tmp_path, EXITING_WORKER, 3, workers=2, restart=False) + with caplog.at_level(logging.INFO, logger="django_database_task"): + harness.start().join() + + (record,) = messages_starting(caplog, "Supervisor finished") + assert record.restarts == 0 + assert record.abnormal_restarts == 0 + assert not messages_starting(caplog, "Every worker is failing to start") + + class TestSupervisorShutdown: def test_forwards_the_signal_and_waits_for_the_workers(self, tmp_path): harness = Harness(tmp_path, GRACEFUL_WORKER, workers=2, restart=True) From 5423f16dc3a377eda1a52425234834fecd419f1c Mon Sep 17 00:00:00 2001 From: Shinya Okano Date: Fri, 9 Oct 2026 13:13:32 +0900 Subject: [PATCH 2/2] Count a clean exit as a recovery from every worker failing A worker that reaches --max-tasks sooner than the stable window exits with code 0 and is restarted at once, so it never ran past the window and the supervisor kept reporting every worker as failing while that one was working. A clean exit now clears the state too, and the record says which of the two cleared it in recovered_by. The check runs before the restarts so that the record names the process that exited. Refs #65 --- CHANGELOG.md | 17 ++++---- README.md | 9 ++-- django_database_task/supervisor.py | 52 +++++++++++++++-------- tests/test_supervisor.py | 66 ++++++++++++++++++++++++++++++ 4 files changed, 115 insertions(+), 29 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 32d6d6c..91ea5a6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,14 +5,15 @@ ### Added - **A summary of the `--workers` supervisor's state in its log records.** - When every worker is failing within the stable window 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 past the window - again. `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. + 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` diff --git a/README.md b/README.md index 6c62c95..4746b50 100644 --- a/README.md +++ b/README.md @@ -752,9 +752,12 @@ Shutdown complete: every worker exited. 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`, and a `Workers recovered` INFO, with - `worker_index`, `pid` and `failing_for` in seconds, once a worker has run - past the ten seconds again. `Supervisor finished` carries `restarts`, how + `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 diff --git a/django_database_task/supervisor.py b/django_database_task/supervisor.py index ce7186d..ff9bd45 100644 --- a/django_database_task/supervisor.py +++ b/django_database_task/supervisor.py @@ -223,6 +223,9 @@ class _Worker: 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): @@ -375,8 +378,10 @@ def _supervise(self): while not self._shutdown.is_set(): self._apply_scaling() self._reap() - self._start_due() + # 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 for worker in self._workers @@ -473,6 +478,8 @@ def _reap(self): 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) @@ -506,7 +513,9 @@ def _check_failing(self): 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. + 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] @@ -522,23 +531,30 @@ def _check_failing(self): }, ) return - recovered = next( - (worker for worker in active if self._is_stable(worker, now)), None - ) - if recovered is None: + 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 - failing_for = now - self._failing_since - self._failing_since = None - logger.info( - "Workers recovered: index=%d pid=%d", - recovered.index, - recovered.pid, - extra={ - "worker_index": recovered.index, - "pid": recovered.pid, - "failing_for": failing_for, - }, - ) def _restart_delay(self, worker, code, uptime): if code == 0 or uptime >= self.stable_after: diff --git a/tests/test_supervisor.py b/tests/test_supervisor.py index c0f9669..3cfded5 100644 --- a/tests/test_supervisor.py +++ b/tests/test_supervisor.py @@ -120,6 +120,15 @@ def handler(signum, frame): """ ) +# Exits with 1 for its first argv[3] launches, then with 0. +FAILS_THEN_EXITS_CLEANLY_WORKER = ( + PRELUDE + + """ +mark(ready) +sys.exit(1 if len(os.listdir(ready)) <= int(sys.argv[3]) else 0) +""" +) + # The first worker to start stays up until signalled; every other one exits # with 1. ONE_STAYS_UP_WORKER = ( @@ -499,6 +508,63 @@ def test_a_worker_past_the_stable_window_clears_it(self, tmp_path, caplog): assert recovered.worker_index == 1 assert recovered.pid == running assert recovered.failing_for >= 0.5 + assert recovered.recovered_by == "uptime" + + def test_a_clean_exit_clears_it(self, tmp_path, caplog): + """A worker reaching --max-tasks quickly is working, not failing.""" + harness = Harness( + tmp_path, + FAILS_THEN_EXITS_CLEANLY_WORKER, + 2, + workers=1, + restart=True, + backoff_start=0.05, + backoff_max=0.1, + stable_after=60, + ) + with caplog.at_level(logging.INFO, logger="django_database_task"): + harness.start() + wait_for( + lambda: messages_starting(caplog, "Workers recovered"), + what="the recovery record", + ) + harness.supervisor.request_shutdown() + harness.join() + + (failing,) = messages_starting(caplog, "Every worker is failing to start") + (recovered,) = messages_starting(caplog, "Workers recovered") + clean_exits = [ + record + for record in messages_starting(caplog, "Worker process exited") + if record.exit_code == 0 + ] + assert caplog.records.index(failing) < caplog.records.index(recovered) + assert recovered.recovered_by == "clean_exit" + assert recovered.pid == clean_exits[0].pid + + def test_a_new_slot_is_not_a_recovery(self, tmp_path, caplog): + harness = Harness( + tmp_path, + EXITING_WORKER, + 1, + workers=1, + restart=True, + backoff_start=0.05, + backoff_max=0.1, + stable_after=60, + ) + with caplog.at_level(logging.INFO, logger="django_database_task"): + harness.start() + wait_for( + lambda: messages_starting(caplog, "Every worker is failing"), + what="the failing record", + ) + harness.supervisor.add_worker() + harness.wait_exits(6) + harness.supervisor.request_shutdown() + harness.join() + + assert not messages_starting(caplog, "Workers recovered") def test_finish_counts_the_restarts(self, tmp_path, caplog): harness = Harness(