diff --git a/steward/ansible/executor.py b/steward/ansible/executor.py index 4199ff3..2786b0c 100644 --- a/steward/ansible/executor.py +++ b/steward/ansible/executor.py @@ -6,6 +6,7 @@ import logging import os import re import shutil +import signal import tempfile import time from dataclasses import dataclass @@ -26,6 +27,8 @@ _FLUSH_SECS = 5.0 _MEM_LINES_CAP = 5000 # bounded per-run in-memory replay buffer (SSE late-joiners) _MEM_TTL_SECS = 300 # drop in-memory state this long after a run completes _FAILURE_CAP = 50 # max failed-task lines captured into structured results +_READ_CHUNK = 65536 # output is read in chunks, never a whole line at once (#5156) +_MAX_LINE_BYTES = 65536 # a longer line is cut here and marked, not allowed to stall # Full run logs persist here (survive restart + the 1 MB DB cap). Override via env. ARTIFACT_DIR = os.environ.get("STEWARD_ANSIBLE_LOG_DIR", "/data/ansible/runs") @@ -261,20 +264,63 @@ async def _set_status(app: "Quart", run_id: str, status) -> None: ) -async def _terminate_then_kill(proc: asyncio.subprocess.Process, grace: float = 10.0) -> None: +def _signal_group(proc: asyncio.subprocess.Process, sig: int) -> None: + """Send sig to the run's whole process group (ansible-playbook and the + ssh and worker processes it forks), or to the process alone when it has + no group of its own.""" try: - proc.terminate() + os.killpg(proc.pid, sig) except ProcessLookupError: - return - try: - await asyncio.wait_for(proc.wait(), timeout=grace) - except asyncio.TimeoutError: + pass + except OSError: try: - proc.kill() + proc.send_signal(sig) except ProcessLookupError: pass +async def _terminate_then_kill(proc: asyncio.subprocess.Process, grace: float = 10.0) -> None: + """SIGTERM the run's process group, then SIGKILL it if it outlives grace. + The group is killed even once the leader has exited, so no worker it + forked is left behind (#5156).""" + if proc.returncode is None: + _signal_group(proc, signal.SIGTERM) + try: + await asyncio.wait_for(proc.wait(), timeout=grace) + except asyncio.TimeoutError: + pass + _signal_group(proc, signal.SIGKILL) + + +async def _lines(stream: asyncio.StreamReader): + """The stream's lines, read in chunks so no line length can stall or break + the reader: a line past _MAX_LINE_BYTES is cut there and marked, and the + rest of it is dropped up to its newline (#5156).""" + pending = b"" + skipping = False + while True: + chunk = await stream.read(_READ_CHUNK) + if not chunk: + break + pending += chunk + while True: + nl = pending.find(b"\n") + if nl < 0: + break + line, pending = pending[:nl], pending[nl + 1:] + if skipping: + skipping = False + continue + yield line + if len(pending) > _MAX_LINE_BYTES and not skipping: + yield pending[:_MAX_LINE_BYTES] + b" [line cut at %d bytes]" % _MAX_LINE_BYTES + pending, skipping = b"", True + elif skipping: + pending = b"" + if pending and not skipping: + yield pending + + def cancel_run(run_id: str) -> bool: """Request cancellation of a run. Terminates the process if it's running; a queued run is flagged so start_run aborts before launch. Returns False @@ -404,6 +450,8 @@ async def start_run( final_status = "failed" logf = None + proc = None + interrupted = False try: if run_id in _cancelled: # cancelled while queued — abort before launching raise _Aborted @@ -463,17 +511,19 @@ async def start_run( if ssh_user: user_args += ["--user", ssh_user] + # Its own session, so the whole group (ssh, forks) can be signalled. proc = await asyncio.create_subprocess_exec( *cmd, *cred_args, *boot_args, *user_args, *ev_args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, cwd=cwd, env=ansible_env(creds, os.environ), + start_new_session=True, ) _run_procs[run_id] = proc assert proc.stdout is not None - async for raw_line in proc.stdout: + async for raw_line in _lines(proc.stdout): line = scrub(raw_line.decode(errors="replace").rstrip("\n\r"), secrets) _broadcast(run_id, line) if logf is not None: @@ -496,7 +546,8 @@ async def start_run( await proc.wait() recap = parse_recap(_run_lines.get(run_id, [])) - if run_id in _cancelled: + # A cancel that lands after the run finished doesn't undo its result. + if run_id in _cancelled and proc.returncode != 0: final_status = "cancelled" elif proc.returncode != 0: final_status = "failed" @@ -511,10 +562,21 @@ async def start_run( else: final_status = "success" finally: + # Whatever ended the read (the run finishing, an error, the task + # being cancelled), nothing the run started outlives it. + if proc is not None: + await asyncio.shield(_terminate_then_kill(proc, grace=5.0)) shutil.rmtree(tmpdir, ignore_errors=True) except _Aborted: final_status = "cancelled" + except asyncio.CancelledError: + # The task itself was cancelled (shutdown): record that the run did + # not finish rather than leave it marked running (#5156). + final_status, interrupted = "cancelled", True + msg = "[run error] Steward stopped the run before it finished." + _broadcast(run_id, msg) + db_output = (db_output + "\n" + msg) if db_output else msg except Exception as exc: logger.exception("Ansible run %s failed with exception", run_id) final_status = "cancelled" if run_id in _cancelled else "failed" @@ -533,5 +595,11 @@ async def start_run( sem.release() results = {"hosts": parse_recap(_run_lines.get(run_id, [])), "failures": failures} - await _flush(final_status, results=results) + try: + await asyncio.shield(_flush(final_status, results=results)) + except Exception: + logger.exception("Ansible run %s: could not record its final status %s", + run_id, final_status) _broadcast(run_id, _Done(status=final_status)) + if interrupted: + raise asyncio.CancelledError diff --git a/tests/core/test_ansible_executor.py b/tests/core/test_ansible_executor.py index 8fde46c..9b55cd1 100644 --- a/tests/core/test_ansible_executor.py +++ b/tests/core/test_ansible_executor.py @@ -28,6 +28,34 @@ def test_parse_recap_empty_without_recap(): assert parse_recap(["TASK [x] ***", "ok: [h]"]) == {} +class _Stream: + def __init__(self, chunks): + self.chunks = list(chunks) + + async def read(self, _n): + return self.chunks.pop(0) if self.chunks else b"" + + +def _read_lines(chunks, monkeypatch, limit=8): + import asyncio + from steward.ansible import executor + monkeypatch.setattr(executor, "_MAX_LINE_BYTES", limit) + + async def _go(): + return [line async for line in executor._lines(_Stream(chunks))] + return asyncio.run(_go()) + + +def test_lines_split_across_chunks(monkeypatch): + assert _read_lines([b"ab", b"c\nde", b"f\n", b"tail"], monkeypatch) == [b"abc", b"def", b"tail"] + + +def test_a_line_past_the_limit_is_cut_and_the_rest_dropped(monkeypatch): + # #5156: a long line can't stall or break the reader; the next line is intact. + got = _read_lines([b"0123456789", b"abcdef", b"gh\nnext\n"], monkeypatch) + assert got == [b"0123456789"[:8] + b" [line cut at 8 bytes]", b"next"] + + def test_cancel_run_flags_untracked_run(): rid = "nonexistent-run-id" assert cancel_run(rid) is True diff --git a/tests/integration/test_ansible_run_end.py b/tests/integration/test_ansible_run_end.py new file mode 100644 index 0000000..a1b54ae --- /dev/null +++ b/tests/integration/test_ansible_run_end.py @@ -0,0 +1,166 @@ +"""Integration: how an Ansible run ends (#5156). + +ansible-playbook is replaced by a small Python script, run as a real process +through the executor, so reading, the process group and the final status are +the real paths. Needs Postgres for the run row. +""" +from __future__ import annotations + +import asyncio +import os +import sys +import time +import uuid + +import pytest + +pytestmark = pytest.mark.integration + +RECAP = ("PLAY RECAP ****\\n" + "localhost : ok=1 changed=0 unreachable=0 failed=0 skipped=0 rescued=0 ignored=0") + +# Starts a child in the same process group and records its pid, so a test can +# check nothing the run started outlives it. +SPAWN_CHILD = ( + "import subprocess, sys, time\n" + "c = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(60)'])\n" + "open(sys.argv[1], 'w').write(str(c.pid))\n" +) + + +@pytest.fixture +def app(): + if not os.environ.get("STEWARD_DATABASE_URL"): + pytest.skip("needs Postgres") + from steward.app import create_app + return create_app(testing=False) + + +def _use_script(monkeypatch, script: str, *args: str) -> None: + from steward.ansible import executor + monkeypatch.setattr(executor, "build_ansible_command", + lambda *_a, **_k: [sys.executable, "-c", script, *args]) + + +async def _new_run(app) -> str: + from steward.models.ansible import AnsibleRun, AnsibleRunStatus + run_id = str(uuid.uuid4()) + async with app.db_sessionmaker() as s: + async with s.begin(): + s.add(AnsibleRun(id=run_id, playbook_path="t.yml", inventory_path="inv.ini", + source_name="t", triggered_by=None, + status=AnsibleRunStatus.running)) + return run_id + + +async def _row(app, run_id): + from sqlalchemy import select + from steward.models.ansible import AnsibleRun + async with app.db_sessionmaker() as s: + return (await s.execute(select(AnsibleRun).where(AnsibleRun.id == run_id))).scalar_one() + + +def _gone(pid: int, within: float = 5.0) -> bool: + """The process has exited (a zombie left for init to reap counts).""" + deadline = time.monotonic() + within + while time.monotonic() < deadline: + try: + with open(f"/proc/{pid}/status") as f: + if "State:\tZ" in f.read(): + return True + except FileNotFoundError: + return True + time.sleep(0.1) + return False + + +def test_a_line_longer_than_the_reader_limit_is_cut_not_fatal(app, monkeypatch, tmp_path): + from steward.ansible import executor + from steward.models.ansible import AnsibleRunStatus + _use_script(monkeypatch, f"print('x' * 200000); print('after'); print('{RECAP}')") + + async def _go(): + run_id = await _new_run(app) + await executor.start_run(app, run_id, "t.yml", "inv.ini", str(tmp_path)) + return await _row(app, run_id) + + run = asyncio.run(_go()) + assert run.status == AnsibleRunStatus.success, run.output + assert f"[line cut at {executor._MAX_LINE_BYTES} bytes]" in run.output + assert "after" in run.output + + +def test_an_error_mid_run_ends_the_whole_process_group(app, monkeypatch, tmp_path): + from steward.ansible import executor + from steward.models.ansible import AnsibleRunStatus + pidfile = tmp_path / "child.pid" + _use_script(monkeypatch, SPAWN_CHILD + "print('boom', flush=True)\ntime.sleep(60)\n", + str(pidfile)) + real_scrub = executor.scrub + + def scrub(line, secrets): + if line == "boom": + raise RuntimeError("reader failed") + return real_scrub(line, secrets) + + monkeypatch.setattr(executor, "scrub", scrub) + + async def _go(): + run_id = await _new_run(app) + await asyncio.wait_for( + executor.start_run(app, run_id, "t.yml", "inv.ini", str(tmp_path)), 30) + return await _row(app, run_id) + + run = asyncio.run(_go()) + assert run.status == AnsibleRunStatus.failed + assert "reader failed" in run.output + assert _gone(int(pidfile.read_text())) + + +def test_a_run_whose_task_is_cancelled_is_not_left_running(app, monkeypatch, tmp_path): + from steward.ansible import executor + from steward.models.ansible import AnsibleRunStatus + pidfile = tmp_path / "child.pid" + _use_script(monkeypatch, SPAWN_CHILD + "print('started', flush=True)\ntime.sleep(60)\n", + str(pidfile)) + + async def _go(): + run_id = await _new_run(app) + task = asyncio.create_task( + executor.start_run(app, run_id, "t.yml", "inv.ini", str(tmp_path))) + deadline = time.monotonic() + 15 + while "started" not in executor._run_lines.get(run_id, []): + assert time.monotonic() < deadline, "the run never started" + await asyncio.sleep(0.1) + task.cancel() # Steward shutting down mid-run + with pytest.raises(asyncio.CancelledError): + await task + return await _row(app, run_id) + + run = asyncio.run(_go()) + assert run.status == AnsibleRunStatus.cancelled + assert run.finished_at is not None + assert _gone(int(pidfile.read_text())) + + +def test_a_cancel_landing_as_the_run_finishes_keeps_its_result(app, monkeypatch, tmp_path): + from steward.ansible import executor + from steward.models.ansible import AnsibleRunStatus + _use_script(monkeypatch, f"print('{RECAP}')") + real_scrub = executor.scrub + run_ids: list[str] = [] + + def scrub(line, secrets): + if line.startswith("localhost :"): # the last line: the cancel arrives now + executor._cancelled.add(run_ids[0]) + return real_scrub(line, secrets) + + monkeypatch.setattr(executor, "scrub", scrub) + + async def _go(): + run_ids.append(await _new_run(app)) + await executor.start_run(app, run_ids[0], "t.yml", "inv.ini", str(tmp_path)) + return await _row(app, run_ids[0]) + + run = asyncio.run(_go()) + assert run.status == AnsibleRunStatus.success