Fix orphaned subprocesses and supervisor crash on heartbeat 409 - #65738
Fix orphaned subprocesses and supervisor crash on heartbeat 409#65738cmettler wants to merge 2 commits into
Conversation
|
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
|
potiuk
added
the
ready for maintainer review
|
CI failures were unrelated — caused by azure-storage-blob 12.30.0 breaking the WASB SAS-token tests (issue #68482), fixed upstream in #68490. After rebasing onto current main, that fix is now in our sources and CI should pass. PR is ready for review when you have time. Drafted-by: Claude Code (Opus 4.7); reviewed by @cmettler before posting |
| # have been placed in its own session via os.setsid() at fork | ||
| # time (see start()). See issue #65505. | ||
| try: | ||
| os.killpg(os.getpgid(self._process.pid), sig) |
There was a problem hiding this comment.
os.killpg(os.getpgid(self._process.pid), sig) trusts that the child has already run os.setsid(). Two paths break that assumption and make this signal the supervisor's own process group:
setsid()is wrapped inwith suppress(OSError)instart(), so if it ever fails the child stays in the supervisor's group._on_child_startedcallsself.kill(signal.SIGKILL)on any exception fromtask_instances.start()(line 1385).setsid()runs in the forked child and the parent doesn't synchronize on it, so a synchronous failure there can reachkill()before the child has runsetsid(). (A 409 over the network is fine, since the round-trip gives the child time to run it; a local/synchronous failure isn't.)
In both cases os.getpgid(child) returns the supervisor's PGID and os.killpg(..., SIGKILL) hits the supervisor and every sibling in its group. The except (ProcessLookupError, PermissionError) fallback doesn't catch this because nothing is raised. test_child_is_session_leader asserts this exact invariant ("so os.killpg() does not signal the supervisor itself"), but the production path has no guard.
Either set the group race-free from the parent too (with suppress(OSError): os.setpgid(pid, pid) right after the fork) or guard the kill site:
pgid = os.getpgid(self._process.pid)
if pgid == os.getpgid(0):
self._process.send_signal(sig)
else:
os.killpg(pgid, sig)Separately: this group-signal only runs on the kill() path. The graceful path in wait() (_forward_signal -> os.kill(self.pid, signum)) still signals the task-runner alone, so the orphan leak this PR targets persists on graceful SIGTERM (e.g. K8s pod termination). Now that the child is a session leader, that path could use killpg too.
There was a problem hiding this comment.
Thinking about this more: airflow.utils.process_utils.reap_process_group() already implements this whole teardown, and it has the exact guard that's missing here:
if not IS_WINDOWS and process_group_id == os.getpgid(0):
raise RuntimeError("I refuse to kill myself")It also covers what this loop doesn't: SIGTERM -> wait -> SIGKILL escalation via psutil.wait_procs, EPERM -> sudo -n kill for the run_as_user case, and ESRCH (the "child hasn't changed its group yet" race) by falling back to signalling the PID directly.
It lives in airflow-core, and the supervisor keeps its airflow.* imports lazy for worker isolation, so it's not a drop-in import. But rather than hand-rolling a second, less complete version here, should we port/copy reap_process_group (+ its self-group guard) into task-sdk and use that instead? One tested teardown path beats two that can drift.
There was a problem hiding this comment.
Good catch — both failure paths were real. Fixed by doing both of your suggestions: the process group is now set from both sides of the fork (os.setpgid(0, 0) in the child, os.setpgid(pid, pid) in the parent right after fork), so the group exists before start() returns and the _on_child_started race is closed deterministically; and kill() now goes through a _signal_subprocess() helper that refuses to killpg when the resolved PGID equals our own (the same invariant as reap_process_group's "I refuse to kill myself" guard).
On the graceful path: that resolved itself upstream — #69034 removed the SIGTERM forwarding entirely in favour of warm shutdown, so kill() is now the only signal-delivery path and it's covered.
On porting reap_process_group/set_new_process_group into task-sdk: agreed that one tested teardown beats two that can drift. I kept this PR minimal but aligned the semantics with set_new_process_group (plain setpgid, no new session), so a port becomes a drop-in. One design question before attempting it as a follow-up: reap_process_group blocks in psutil.wait_procs during escalation, while the supervisor must keep servicing the child's sockets while it dies (last log lines, terminal state — and a full pipe would deadlock otherwise). So I'd either share only the primitives (group setup + guarded group-signal) and keep the supervisor's escalation loop, or port reap_process_group with an injectable wait-hook. If the first, simpler option sounds right to you I'm happy to pick that up once this PR lands.
Drafted-by: Claude Code (Opus 4.7); reviewed by @cmettler before posting
There was a problem hiding this comment.
Verified the new revision: the parent-side setpgid(pid, pid) mirror closes the _on_child_started race deterministically, and the own-group guard in _signal_subprocess() covers the case where both setpgid calls failed. On the follow-up question: sharing only the primitives (group setup + guarded group-signal) and keeping the supervisor's escalation loop sounds right to me. The supervisor has to keep draining the child's sockets while it dies, so reap_process_group's blocking psutil.wait_procs is the wrong shape here, and an injectable wait-hook would complicate its other call sites for no gain. Happy to review that follow-up when you get to it.
One small ask: the PR description still describes the earlier setsid()/session-leader design and cites the old test name (test_child_is_session_leader) -- mind refreshing it to match the current revision before merge?
|
Following up on @ashb’s internal note about the session-leader TODO: Confirmed this PR does make the task-runner a session leader — That makes the pre-existing Could you drop that TODO comment as part of this change? Thanks! |
…nst self-signalling Review feedback on apache#65738: os.killpg(os.getpgid(child)) trusted that the child had already run setsid() -- if setpgid failed or kill() ran before the child was first scheduled (task_instances.start() raising synchronously), getpgid resolved to the supervisor's own group and killpg would have signalled the supervisor and all its siblings, with no exception for the fallback to catch. Use a plain process group (setpgid, matching airflow.utils.process_utils.set_new_process_group) instead of a new session, set it from both sides of the fork so the group exists as soon as start() returns, refuse to killpg our own group, and make the whole behaviour opt-in per subclass (like use_exec) so the DAG processor, triggerer and callback subprocesses keep direct signalling. The graceful SIGTERM-forwarding path now also signals the group, closing the same orphan leak on e.g. K8s pod termination.
eladkal
added
type:bug-fix
When a running TaskInstance is forcibly transitioned out of `running` (e.g. the scheduler resets a stale heartbeat, or an operator PATCHes the state to `failed`), the task-runner's next heartbeat returns HTTP 409 and the supervisor kills the task. Before this change two things went wrong on Linux: 1. Subprocesses the task-runner had spawned (`@task.virtualenv` / `PythonVirtualenvOperator` children, `DockerOperator` exec, Bash shells) were reparented to PID 1 and kept running as orphans until they finished on their own - wasting CPU, RAM and third-party API quota. 2. About 60s later, `_cleanup_open_sockets()` closed the selector while `_service_subprocess()` was still using it, so the supervisor crashed with `ValueError: I/O operation on closed epoll object` (regression from PR apache#51180). The task-runner is now placed in its own session via `os.setsid()` immediately after fork, so its process group ID equals its PID. The supervisor's `kill()` signals the whole group via `os.killpg(os.getpgid(pid), sig)`, which reaches every subprocess the task-runner spawned. Grandchildren without a SIGTERM handler exit promptly, close their inherited pipes, and the supervisor drains `_open_sockets` normally - so `_cleanup_open_sockets()` is never triggered and the selector is never closed mid-loop. `os.killpg`/`os.getpgid` fall back to `self._process.send_signal(sig)` on `ProcessLookupError` or `PermissionError`, preserving prior behaviour when the group has vanished (e.g. the task was already reaped) or permissions are lacking. closes: apache#65505
…nst self-signalling Review feedback on apache#65738: os.killpg(os.getpgid(child)) trusted that the child had already run setsid() -- if setpgid failed or kill() ran before the child was first scheduled (task_instances.start() raising synchronously), getpgid resolved to the supervisor's own group and killpg would have signalled the supervisor and all its siblings, with no exception for the fallback to catch. Use a plain process group (setpgid, matching airflow.utils.process_utils.set_new_process_group) instead of a new session, set it from both sides of the fork so the group exists as soon as start() returns, refuse to killpg our own group, and make the whole behaviour opt-in per subclass (like use_exec) so the DAG processor, triggerer and callback subprocesses keep direct signalling. The graceful SIGTERM-forwarding path now also signals the group, closing the same orphan leak on e.g. K8s pod termination.
6 participants
When a running TaskInstance is forcibly transitioned out of
running(scheduler reset, REST PATCH, etc.), the next heartbeat from the still-running task-runner returns HTTP 409 and the supervisor kills the task. On Linux this produced two bugs:@task.virtualenv,DockerOperator,BashOperator, Cosmos dbt, etc.) was reparented to PID 1 and kept running until it finished on its own, wasting CPU/RAM/API quota._cleanup_open_sockets()closed the selector while_service_subprocess()was still polling it, raisingValueError: I/O operation on closed epoll object(regression from Fix lingering task supervisors whenEOFis missed #51180).Fix
Place the task-runner in its own session via
os.setsid()immediately after fork, then havekill()signal the whole process group viaos.killpg(os.getpgid(pid), sig). This reaches every subprocess the task-runner spawned. Grandchildren without a SIGTERM handler exit promptly and close their inherited pipes, so the supervisor drains_open_socketsnormally and never enters the cleanup-the-selector-mid-loop path.killpg/getpgidfall back toself._process.send_signal(sig)onProcessLookupErrororPermissionError, preserving behaviour when the group has vanished or permissions are lacking.Tests
test_kill_signals_process_group— primary path useskillpg.test_kill_falls_back_to_send_signal_when_group_signal_fails(4 params:{ProcessLookupError, PermissionError} × {getpgid, killpg}).test_child_is_session_leader— real-fork regression: asserts child's PGID == child's PID afterActivitySubprocess.start().os.getpgid/os.killpgexplicitly.uv run --project task-sdk pytest task-sdk/tests/task_sdk/execution_time/test_supervisor.py).closes: #65505
Was generative AI tooling used to co-author this PR?
Generated-by: Claude Opus 4.7 (1M context) following the guidelines
Important
🛠️ Maintainer triage note for @cmettler · by
@potiuk· 2026-06-22 06:31 UTCHelpful heads-up from the maintainers — please address before this PR can be reviewed (see the Pull Request quality criteria):
The ball is in your court — you've been assigned to this PR. Fix the above, then mark it Ready for review.
Automated triage — may be imperfect; a maintainer takes the next look.