diff --git a/docs/operations/scanner-benchmark-runbook.md b/docs/operations/scanner-benchmark-runbook.md index cedfc4966..04f06aee4 100644 --- a/docs/operations/scanner-benchmark-runbook.md +++ b/docs/operations/scanner-benchmark-runbook.md @@ -85,6 +85,13 @@ at most 1 MiB. Logs are kept separately and require an operator-managed disk quota. Missing output, timeout, nonzero exit, unknown/missing metrics, zero samples, and request errors fail the run. Adapters must terminate their own children on failure and `stop` must be idempotent even after partial preparation. +The runner keeps its session leader unreaped while stopping a failed command +or collector: it sends TERM, allows the existing ten-second grace period, then +kills the remaining process group before reaping. This prevents a parent exit +from hiding live descendants or allowing the group ID to be reused before its +last signal. A successful `prepare` preserves adapter-owned services until +`stop`; services that leave the command's process group remain the adapter's +cleanup responsibility. The request contains the fixed manifest fields, selected build, scenario, round, leg, comparison (`build` or `background`), background mode (`on` or `off`), diff --git a/scripts/scanner_abba.py b/scripts/scanner_abba.py index 9c461dc7b..ec3c578be 100644 --- a/scripts/scanner_abba.py +++ b/scripts/scanner_abba.py @@ -7,6 +7,7 @@ import json import math import os from pathlib import Path +import select import shutil import signal import subprocess @@ -91,28 +92,110 @@ def validate_manifest(manifest): require(manifest["expected_healed_objects"][scenario] > 0, f"{scenario} requires repairs") +class OwnedCommand: + """Keep the session leader unreaped until its group's last signal is sent.""" + + def __init__(self, args, log): + require(sys.platform == "darwin" or hasattr(os, "waitid"), "non-reaping child observation is unavailable") + self.args, self.status = args, None + self.queue = select.kqueue() if sys.platform == "darwin" else None + self.process = None + read_gate, write_gate = os.pipe() + try: + # The shell has already exec'd when Popen returns. Gate the target + # until kqueue is registered; preexec_fn would deadlock Popen here. + gate = f'read -r _scanner_gate <&{read_gate} || exit 125; exec {read_gate}<&-; exec "$@"' + self.process = subprocess.Popen(["bash", "-c", gate, "scanner-abba", *args], pass_fds=(read_gate,), + stdout=log, stderr=subprocess.STDOUT, start_new_session=True) + if self.queue is not None: + # Darwin NOTE_EXITSTATUS is not exposed by Python's select constants. + event = select.kevent(self.process.pid, filter=select.KQ_FILTER_PROC, + flags=select.KQ_EV_ADD | select.KQ_EV_ONESHOT, + fflags=select.KQ_NOTE_EXIT | 0x04000000) + self.queue.control([event], 0, 0) + os.write(write_gate, b"\n") + except BaseException: + try: + if self.process is not None: + try: + self._signal_group(signal.SIGKILL) + finally: + self.process.wait(timeout=10) + finally: + if self.queue is not None: + self.queue.close() + raise + finally: + os.close(read_gate) + os.close(write_gate) + + def wait(self, timeout): + if self.status is not None: + return self.status + deadline = time.monotonic() + timeout + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise subprocess.TimeoutExpired(self.args, timeout) + if self.queue is not None: + events = self.queue.control(None, 1, remaining) + if events: + self.status = os.waitstatus_to_exitcode(events[0].data) + return self.status + else: + result = os.waitid(os.P_PID, self.process.pid, os.WEXITED | os.WNOWAIT | os.WNOHANG) + if result is not None: + self.status = result.si_status if result.si_code == os.CLD_EXITED else -result.si_status + return self.status + time.sleep(min(0.05, remaining)) + + def _signal_group(self, sig): + try: + os.killpg(self.process.pid, sig) + return True + except ProcessLookupError: + return False + + def finish(self, terminate=False): + if self.process.returncode is not None: + return self.process.returncode + try: + if terminate: + try: + self._signal_group(signal.SIGTERM) + deadline = time.monotonic() + 10 + while time.monotonic() < deadline and self._signal_group(0): + time.sleep(0.05) + finally: + # Keep the PID reserved through the last group signal, even + # when the cleanup grace period itself is interrupted. + self._signal_group(signal.SIGKILL) + finally: + try: + returncode = self.process.wait(timeout=10) + finally: + if self.queue is not None: + self.queue.close() + return returncode + + def invoke(adapter, action, request, timeout): """The adapter writes bounded JSON separately; stderr/stdout remain raw evidence.""" output = request.parent / f"{action}.json" with (request.parent / f"{action}.log").open("wb") as log: - process = subprocess.Popen([str(adapter), action, str(request), str(output)], - stdout=log, stderr=subprocess.STDOUT, start_new_session=True) + process = OwnedCommand([str(adapter), action, str(request), str(output)], log) try: - returncode = process.wait(timeout=timeout) + returncode = process.wait(timeout) if returncode: raise subprocess.CalledProcessError(returncode, [str(adapter), action]) - finally: - if process.poll() != 0: - try: - os.killpg(process.pid, signal.SIGTERM) - except ProcessLookupError: - pass - try: - process.wait(timeout=10) - except subprocess.TimeoutExpired: - os.killpg(process.pid, signal.SIGKILL) - process.wait() - return read_json(output) + result = read_json(output) + except BaseException: + process.finish(terminate=True) + raise + else: + # Successful prepare may intentionally leave adapter-owned services. + process.finish() + return result def validate_result(result, request, expected): @@ -212,12 +295,12 @@ def collect_live(prepared, request, request_path, adapter): "--samples", str(request["duration_seconds"] // 60 + 1), "--interval-secs", "60", "--out-dir", str(output)] with (request_path.parent / "collector.log").open("wb") as log: - process = subprocess.Popen(args, stdout=log, stderr=subprocess.STDOUT, start_new_session=True) + process = OwnedCommand(args, log) try: started = time.monotonic() result = invoke(adapter, "measure", request_path, request["duration_seconds"] + 300) require(time.monotonic() - started >= request["duration_seconds"], "measurement ended before required window") - require(process.wait(timeout=120) == 0, "scanner collector failed") + require(process.wait(120) == 0, "scanner collector failed") require(output.joinpath("scanner-summary.csv").stat().st_size > 0, "missing collector samples") samples = list((output / "status").glob("scanner-status.*.json")) require(len(samples) == request["duration_seconds"] // 60 + 1, "missing scanner samples") @@ -231,16 +314,7 @@ def collect_live(prepared, request, request_path, adapter): require(isinstance(status.get("healOperations"), dict) and status["healOperations"], "invalid heal status response") return result finally: - # Stop telemetry children as well when measurement fails or times out. - try: - os.killpg(process.pid, signal.SIGTERM) - except ProcessLookupError: - pass - try: - process.wait(timeout=10) - except subprocess.TimeoutExpired: - os.killpg(process.pid, signal.SIGKILL) - process.wait() + process.finish(terminate=True) def run(manifest, adapter, output, data_root): diff --git a/scripts/test_scanner_abba.py b/scripts/test_scanner_abba.py index 30fc00839..9ca3a56bf 100755 --- a/scripts/test_scanner_abba.py +++ b/scripts/test_scanner_abba.py @@ -3,13 +3,17 @@ import contextlib import copy +import fcntl import io import json import os from pathlib import Path +import shlex +import signal import subprocess import sys import tempfile +import time import unittest from unittest.mock import patch @@ -20,9 +24,18 @@ def fake_adapter(): action, request_path, output_path = sys.argv[1:] request = harness.read_json(Path(request_path)) fault = os.environ.get("SCANNER_ABBA_TEST_FAULT", "") + if fault == "stubborn-child" and action in ("prepare", "measure"): + marker = Path(request_path).parent / "stubborn.pid" + if os.fork() == 0: + os.execv(sys.executable, [sys.executable, str(Path(__file__).resolve()), "--stubborn-worker", str(marker)]) + wait_for_marker(marker) + if action == "measure": + time.sleep(60) if action == "prepare": result = {"ready": True} elif action == "stop": + if fault == "stubborn-child": + reap_fixture(Path(request_path).parent / "stubborn.pid") result = {"stopped": True} elif action == "oracle": if fault == "oracle-exit": @@ -73,6 +86,36 @@ def fake_adapter(): return 0 +def wait_for_marker(marker): + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + if marker.exists() and marker.stat().st_size: + return + time.sleep(0.01) + raise AssertionError("fixture child did not become ready") + + +def child_released(marker, timeout=1): + deadline = time.monotonic() + timeout + with marker.open("r+") as stream: + while True: + try: + fcntl.flock(stream, fcntl.LOCK_EX | fcntl.LOCK_NB) + return True + except BlockingIOError: + if time.monotonic() >= deadline: + return False + time.sleep(0.01) + + +def reap_fixture(marker): + if marker.exists() and marker.stat().st_size and not child_released(marker, timeout=0): + # The unique file lock proves the original fixture process still owns this PID. + os.kill(int(marker.read_text()), signal.SIGKILL) + if not child_released(marker, timeout=5): + raise AssertionError("fixture child did not release its process-owned lock") + + class ScannerAbbaTest(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() @@ -96,6 +139,112 @@ class ScannerAbbaTest(unittest.TestCase): with patch.dict(os.environ, {"SCANNER_ABBA_TEST_FAULT": fault}), contextlib.redirect_stdout(io.StringIO()): return harness.run(copy.deepcopy(self.manifest), self.adapter, self.root / "out", self.root / "data") + def test_adapter_timeout_reaps_group_after_parent_exits_on_term(self): + request = self.root / "request.json" + harness.write_json(request, {}) + marker = self.root / "stubborn.pid" + try: + with patch.dict(os.environ, {"SCANNER_ABBA_TEST_FAULT": "stubborn-child"}): + with self.assertRaises(subprocess.TimeoutExpired): + harness.invoke(self.adapter, "measure", request, 3) + wait_for_marker(marker) + self.assertTrue(child_released(marker), "TERM-exited parent left its TERM-ignoring child alive") + finally: + reap_fixture(marker) + + def test_collector_failure_reaps_group_after_parent_exits_on_term(self): + request = self.root / "request.json" + harness.write_json(request, {}) + marker = self.root / "stubborn.pid" + collector = self.root / "run_scanner_validation_harness.sh" + command = [sys.executable, str(self.adapter), "measure", str(request), str(self.root / "unused.json")] + collector.write_text("#!/usr/bin/env bash\nexec " + shlex.join(command) + "\n") + + def failed_measure(*_): + wait_for_marker(marker) + raise ValueError("injected measurement failure") + + try: + with patch.dict(os.environ, {"SCANNER_ABBA_TEST_FAULT": "stubborn-child"}), \ + patch.object(harness, "__file__", str(self.root / "scanner_abba.py")), \ + patch.object(harness, "invoke", side_effect=failed_measure): + with self.assertRaisesRegex(ValueError, "injected measurement failure"): + harness.collect_live({"collector": {"alias": "fixture", "endpoint": "fixture", "metrics_endpoints": "fixture"}}, + {"duration_seconds": 900}, request, self.adapter) + self.assertTrue(child_released(marker), "collector parent exit did not end its telemetry child") + finally: + reap_fixture(marker) + + def test_successful_prepare_keeps_service_alive(self): + request = self.root / "request.json" + harness.write_json(request, {}) + marker = self.root / "stubborn.pid" + try: + with patch.dict(os.environ, {"SCANNER_ABBA_TEST_FAULT": "stubborn-child"}): + self.assertEqual(harness.invoke(self.adapter, "prepare", request, 5), {"ready": True}) + self.assertFalse(child_released(marker, timeout=0), "successful prepare must preserve its service") + self.assertEqual(harness.invoke(self.adapter, "stop", request, 5), {"stopped": True}) + self.assertTrue(child_released(marker), "adapter stop must release its service") + finally: + reap_fixture(marker) + + def test_reaped_owner_never_signals_a_reused_process_group(self): + with (self.root / "owner.log").open("wb") as log: + owner = harness.OwnedCommand([sys.executable, "-c", "pass"], log) + self.assertEqual(owner.wait(5), 0) + self.assertEqual(owner.finish(), 0) + with patch.object(harness.os, "killpg", side_effect=AssertionError("released PGID must not be signalled")): + self.assertEqual(owner.finish(terminate=True), 0) + + def test_cleanup_interruption_still_kills_group_and_reaps_leader(self): + request = self.root / "request.json" + harness.write_json(request, {}) + marker = self.root / "stubborn.pid" + original_sleep = time.sleep + interrupted = False + + def interrupt_once(delay): + nonlocal interrupted + if not interrupted: + interrupted = True + raise KeyboardInterrupt + original_sleep(delay) + + with (self.root / "interrupted.log").open("wb") as log: + with patch.dict(os.environ, {"SCANNER_ABBA_TEST_FAULT": "stubborn-child"}): + owner = harness.OwnedCommand([str(self.adapter), "measure", str(request), str(self.root / "unused.json")], log) + try: + wait_for_marker(marker) + with patch.object(harness.time, "sleep", side_effect=interrupt_once): + with self.assertRaises(KeyboardInterrupt): + owner.finish(terminate=True) + self.assertTrue(child_released(marker), "cleanup cancellation left its child alive") + self.assertIsNotNone(owner.process.returncode, "cleanup cancellation must reap its leader") + finally: + reap_fixture(marker) + owner.process.wait(timeout=5) + + def test_constructor_failure_after_gate_release_kills_group(self): + request = self.root / "request.json" + harness.write_json(request, {}) + marker = self.root / "stubborn.pid" + original_write = os.write + + def release_then_fail(fd, data): + original_write(fd, data) + wait_for_marker(marker) + raise OSError("injected failure after gate release") + + try: + with (self.root / "construction.log").open("wb") as log, \ + patch.dict(os.environ, {"SCANNER_ABBA_TEST_FAULT": "stubborn-child"}), \ + patch.object(harness.os, "write", side_effect=release_then_fail): + with self.assertRaisesRegex(OSError, "injected failure after gate release"): + harness.OwnedCommand([str(self.adapter), "measure", str(request), str(self.root / "unused.json")], log) + self.assertTrue(child_released(marker), "initialization failure left its child alive") + finally: + reap_fixture(marker) + def test_complete_synthetic_matrix_is_not_performance_evidence(self): self.assertEqual(self.run_harness(), 0) report = harness.read_json(self.root / "out/report.json") @@ -176,6 +325,14 @@ class ScannerAbbaTest(unittest.TestCase): if __name__ == "__main__": + if len(sys.argv) == 3 and sys.argv[1] == "--stubborn-worker": + signal.signal(signal.SIGTERM, signal.SIG_IGN) + with Path(sys.argv[2]).open("w+") as marker: + fcntl.flock(marker, fcntl.LOCK_EX) + marker.write(str(os.getpid())) + marker.flush() + while True: + time.sleep(1) if len(sys.argv) == 4 and sys.argv[1] in ("prepare", "measure", "oracle", "stop"): sys.exit(fake_adapter()) unittest.main()