kaxil commented on code in PR #65738:
URL: https://github.com/apache/airflow/pull/65738#discussion_r3501605509
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -1007,7 +1017,18 @@ def kill(
for sig in escalation_path:
try:
- self._process.send_signal(sig)
+ # Signal the whole process group so subprocesses the
+ # task-runner spawned (venv children, Docker exec, bash
+ # shells, etc.) are also reached. Requires the task-runner to
+ # 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)
Review 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:
```python
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.
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -690,6 +690,16 @@ def start(
pid = os.fork()
if pid == 0:
+ # Put the task-runner into its own session so its PGID == its own
+ # PID. The supervisor can then deliver signals to the whole tree
+ # via os.killpg() in kill(), reaching every subprocess the
+ # task-runner spawned (e.g. venv children from
+ # PythonVirtualenvOperator). Without this, a SIGTERM from kill()
+ # only hits the task-runner and any Popen children are reparented
+ # to PID 1 and leak as orphans. See issue #65505.
+ with suppress(OSError):
+ os.setsid()
Review Comment:
On the scope point: the existing `set_new_process_group()`
(`airflow.utils.process_utils`) does this with `os.setpgid(0, 0)` rather than
`os.setsid()`. That still gives a killable process group (so `killpg` reaches
the whole tree) but without creating a new session / detaching the controlling
terminal, so it avoids the Ctrl-C / foreground-group change noted above. Its
companion `reap_process_group()` already has the SIGTERM -> SIGKILL escalation
and EPERM/ESRCH handling.
Same question as on the kill() thread: should we port/copy these over into
task-sdk and use them here instead of a fresh `setsid`/`killpg` path? Reusing
the existing `setpgid` group + self-group guard would address both the
over-broad scope and the self-signal risk in one go.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
