From eff7ba4bb276272518d950eb88405413459f48b1 Mon Sep 17 00:00:00 2001 From: Alexander Tarasov Date: Tue, 13 Oct 2026 19:25:36 +0200 Subject: [PATCH 1/3] feat(healthcheck): Add a healthcheck command for consumer liveness probes Add `python3 -m arroyo.healthcheck `, which removes the file the Healthcheck strategy touches, or exits 1 with a one-line message if it is missing. Consumers run this check today as `rm `, which needs a shell image, or as an inline `python3 -c` snippet in each deployment. Arroyo writes the file, so it should own the check too. The module imports only the standard library and `arroyo`, so it stays cheap enough to run every few seconds. Co-Authored-By: Claude --- arroyo/healthcheck.py | 42 +++++++++++++++++++++++++++++++++++++++ tests/test_healthcheck.py | 38 +++++++++++++++++++++++++++++++++++ 2 files changed, 80 insertions(+) create mode 100644 arroyo/healthcheck.py create mode 100644 tests/test_healthcheck.py diff --git a/arroyo/healthcheck.py b/arroyo/healthcheck.py new file mode 100644 index 00000000..aa86f30f --- /dev/null +++ b/arroyo/healthcheck.py @@ -0,0 +1,42 @@ +""" +Liveness check for consumers that use the ``Healthcheck`` strategy: + + python3 -m arroyo.healthcheck /tmp/health.txt + +The strategy touches the file on every poll, at most once per second. This +check removes it, so the next check fails unless the consumer touches it again +in between. Imports only the standard library and ``arroyo`` itself, so it +starts fast and needs no shell in the image. +""" + +import argparse +import os +import sys +from typing import Optional + + +def healthcheck(path: str) -> None: + """ + Remove the health file, or exit 1 if the consumer has not touched it since + the last check. + """ + try: + os.remove(path) + except FileNotFoundError: + sys.exit( + f"{path} is missing: the consumer has not touched it since the last probe" + ) + + +def main(argv: Optional[list[str]] = None) -> None: + parser = argparse.ArgumentParser( + prog="python3 -m arroyo.healthcheck", + description="Remove the consumer health file, or exit 1 if it is missing.", + ) + parser.add_argument("path", help="the file the Healthcheck strategy touches") + args = parser.parse_args(argv) + healthcheck(args.path) + + +if __name__ == "__main__": + main() diff --git a/tests/test_healthcheck.py b/tests/test_healthcheck.py new file mode 100644 index 00000000..25e73ee7 --- /dev/null +++ b/tests/test_healthcheck.py @@ -0,0 +1,38 @@ +import subprocess +import sys +from pathlib import Path +from unittest import mock + +import pytest + +from arroyo.healthcheck import main +from arroyo.processing.strategies.healthcheck import Healthcheck + + +def test_fails_when_consumer_has_not_touched_file_since_last_check( + tmp_path: Path, +) -> None: + path = tmp_path / "health.txt" + Healthcheck(str(path), mock.Mock()).poll() + + main([str(path)]) + assert not path.exists() + + with pytest.raises(SystemExit) as exc: + main([str(path)]) + assert exc.value.code == ( + f"{path} is missing: the consumer has not touched it since the last probe" + ) + + +def test_import_stays_light() -> None: + # The check runs as a liveness probe every few seconds, so it must not pull + # in the Kafka client or the processing machinery. + code = ( + "import sys, arroyo.healthcheck; " + "print(sorted(m for m in ('confluent_kafka', 'arroyo.processing') if m in sys.modules))" + ) + out = subprocess.run( + [sys.executable, "-c", code], capture_output=True, text=True, check=True + ) + assert out.stdout.strip() == "[]" From 1b5aa2ced90caec883fd0a622c1e298a34948d0b Mon Sep 17 00:00:00 2001 From: Alexander Tarasov Date: Tue, 13 Oct 2026 19:35:00 +0200 Subject: [PATCH 2/3] test(healthcheck): Drop the import-weight test Match the taskbroker client commands, which test behavior only. Co-Authored-By: Claude --- tests/test_healthcheck.py | 15 --------------- 1 file changed, 15 deletions(-) diff --git a/tests/test_healthcheck.py b/tests/test_healthcheck.py index 25e73ee7..402a0d4f 100644 --- a/tests/test_healthcheck.py +++ b/tests/test_healthcheck.py @@ -1,5 +1,3 @@ -import subprocess -import sys from pathlib import Path from unittest import mock @@ -23,16 +21,3 @@ def test_fails_when_consumer_has_not_touched_file_since_last_check( assert exc.value.code == ( f"{path} is missing: the consumer has not touched it since the last probe" ) - - -def test_import_stays_light() -> None: - # The check runs as a liveness probe every few seconds, so it must not pull - # in the Kafka client or the processing machinery. - code = ( - "import sys, arroyo.healthcheck; " - "print(sorted(m for m in ('confluent_kafka', 'arroyo.processing') if m in sys.modules))" - ) - out = subprocess.run( - [sys.executable, "-c", code], capture_output=True, text=True, check=True - ) - assert out.stdout.strip() == "[]" From 2772132cf47658a4ef09c7b55bfa195357c5f699 Mon Sep 17 00:00:00 2001 From: Alexander Tarasov Date: Tue, 13 Oct 2026 19:36:53 +0200 Subject: [PATCH 3/3] ref(healthcheck): Drop the module docstring Keep the explanation on the function, like the taskbroker client commands. Co-Authored-By: Claude --- arroyo/healthcheck.py | 15 ++------------- 1 file changed, 2 insertions(+), 13 deletions(-) diff --git a/arroyo/healthcheck.py b/arroyo/healthcheck.py index aa86f30f..a2cb56e2 100644 --- a/arroyo/healthcheck.py +++ b/arroyo/healthcheck.py @@ -1,14 +1,3 @@ -""" -Liveness check for consumers that use the ``Healthcheck`` strategy: - - python3 -m arroyo.healthcheck /tmp/health.txt - -The strategy touches the file on every poll, at most once per second. This -check removes it, so the next check fails unless the consumer touches it again -in between. Imports only the standard library and ``arroyo`` itself, so it -starts fast and needs no shell in the image. -""" - import argparse import os import sys @@ -17,8 +6,8 @@ def healthcheck(path: str) -> None: """ - Remove the health file, or exit 1 if the consumer has not touched it since - the last check. + Remove the file the Healthcheck strategy touches, so the next check fails + unless the consumer touches it again. Runs as the consumer's liveness probe. """ try: os.remove(path)