diff --git a/CHANGELOG.md b/CHANGELOG.md index 050e888..91ea5a6 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 1a0db46..4746b50 100644 --- a/README.md +++ b/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. diff --git a/django_database_task/supervisor.py b/django_database_task/supervisor.py index a9c02df..ff9bd45 100644 --- a/django_database_task/supervisor.py +++ b/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." diff --git a/tests/test_supervisor.py b/tests/test_supervisor.py index b017bb9..3cfded5 100644 --- a/tests/test_supervisor.py +++ b/tests/test_supervisor.py @@ -102,6 +102,54 @@ 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) +""" +) + +# 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 = ( + 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 +434,187 @@ 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 + 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( + 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)