This is an automated email from the ASF dual-hosted git repository.
kaxil pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 4e52a36e81a Allow opting into fork+exec task processes via
execute_tasks_new_python_interpreter (#72164)
4e52a36e81a is described below
commit 4e52a36e81a430e77948f30ce8b9add3042cf30e
Author: Gang Zhang <[email protected]>
AuthorDate: Mon Sep 14 14:02:38 2026 -0700
Allow opting into fork+exec task processes via
execute_tasks_new_python_interpreter (#72164)
* Allow opting into fork+exec task processes via
execute_tasks_new_python_interpreter
On Linux the supervisor bare-forks the task process. If any supervisor
thread (OTel exporter, google-auth refresh, ...) holds a C-level lock such
as OpenSSL's at the instant of the fork, the child inherits it permanently
locked and deadlocks at its first TLS call -- typically the DAG bundle
download -- stuck in RUNNING forever with no timeout able to fire (#71707).
exec()-ing a fresh interpreter right after fork (what _FORK_EXEC_PLATFORMS
already forces on macOS) makes the inherited-lock state unreachable. Extend
the existing [core] execute_tasks_new_python_interpreter config to opt into
that path on any platform, and document the new effect.
Kept opt-in because the fresh interpreter re-imports the SDK and plugins
(measured ~9s per task on a plugin-heavy production image; it shows up in
task duration, not queued duration).
Supersedes #71723; the config approach and platform gate are unchanged from
it. The test here patches the configuration through the environment variable
so it reaches the task-sdk conf object the supervisor actually reads.
Co-authored-by: Pulak Saha <[email protected]>
Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
* Re-apply PR_SET_DUMPABLE in the exec'd task child
execve resets the dumpable flag that supervise_task() set before the fork,
so an exec'd
child was readable through /proc/<pid>/environ by a same-UID sibling.
Re-apply it as the
first step of _child_exec_main(), update the security docs that described
the flag as
purely inherited, and pin it with a Linux-only fork+exec test.
* Scope execute_tasks_new_python_interpreter to the task process
_should_use_exec() is also consulted by the Dag processor (one child per
file per
parse loop) and the triggerer runner, so reading the option there would
exec a fresh
interpreter per parse. Keep it as the platform gate and read the option in
ActivitySubprocess.start() only, which is also the shape v3-2/v3-3 can
carry (they
check darwin inline at that spot). Document the tasks-only scope, that the
option was
a no-op for tasks on Airflow 3 until now, and the Edge worker's second
interpreter.
* Test the exec opt-in through conf_vars like the rest of the file
conf_vars already patches the task-sdk conf once the supervisor module is
imported, so the
env-var route was only an inconsistency.
* Reword the security docs so the spellchecker accepts them
"exec'd" is neither a dictionary word nor in the wordlist; say "a child
started
through exec" instead.
* Restore PR_SET_DUMPABLE in the exec bootstrap and pin the fork+exec
wiring in tests
The exec'd child was dumpable from execve until _child_exec_main() ran,
i.e. for the
whole Airflow import; a /proc/<pid>/mem descriptor or ptrace attach taken
then survives
the later prctl. The -c bootstrap (now _CHILD_EXEC_BOOTSTRAP) restores the
flag with
ctypes before importing Airflow, leaving interpreter start as the only
window;
_child_exec_main() keeps the logged fallback.
Tests now pin what the review found unpinned: the real prelude string is
exec'd and
asserted to flip the flag 1 -> 0 (Linux), _child_exec_main() is checked to
call
_make_process_nondumpable() before _fork_main(), and
ActivitySubprocess.start() is
checked to pass the task-process decision as use_exec (stub targets stay
bare fork).
The forked test branch exits via os._exit in a finally so a failed execv
cannot run
pytest in the child. Docstrings no longer describe exec as macOS-only.
* Config/docs: fix the runaway rst literal, note callbacks keep bare fork,
describe the bootstrap prctl
``exec``s left the literal open until the next backticks in the
Configuration
Reference; reworded. Task callbacks run from a closure the exec path cannot
name, so
they stay on bare fork -- say so in the config text and the newsfragment.
The security
docs now describe the prctl in the exec bootstrap and name ptrace_scope >=
1 as what
covers the interpreter-start window.
* Guard the bootstrap's ctypes import and run the exec bootstrap end to end
in tests
The prelude imported ctypes outside its try, so an interpreter built
without _ctypes
would have died before _child_exec_main() could reach the logged fallback;
the import now
sits inside the try like _make_process_nondumpable().
test_fork_exec_bootstrap_runs_an_importable_target_end_to_end drives
ActivitySubprocess.start
through the real os.execv: the fresh interpreter runs
_CHILD_EXEC_BOOTSTRAP, rebuilds FDs 0-3,
rehydrates the target by name and its stdout comes back through the
supervisor. It uses an
importable probe standing in for _subprocess_main rather than the task
runner, because the
suite stubs _get_plugins in-process -- inherited by a bare-forked child,
not by a fresh
interpreter, which then fails with 'Plugins folder is not set'. Also
pinned: the bootstrap
starts with the prelude and ends in _child_exec_main(), and the prelude
survives a missing
_ctypes.
* Docs: Yama scope of ptrace_scope, and that the task process reads the
global option
ptrace_scope >= 1 gates PTRACE_MODE_ATTACH only, so it covers
/proc/<pid>/mem and ptrace
attach for the interpreter-start window but not environ/maps (which show
the supervisor's
environment every task under the worker already has). The task process
reads the global
value of the option; Edge workers resolve it per team.
* Tests: hoist the probe-target and subprocess imports to module level
Both were already available at module scope (subprocess is imported at the
top of the
file and task_sdk is on the path for the other imports), so the local
imports in
test_fork_exec_bootstrap_runs_an_importable_target_end_to_end and
test_prelude_survives_an_interpreter_without_ctypes only added noise.
---------
Co-authored-by: Pulak Saha <[email protected]>
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
---
airflow-core/docs/security/security_model.rst | 7 +-
airflow-core/docs/security/workload.rst | 7 +-
airflow-core/newsfragments/72164.significant.rst | 23 +++
.../src/airflow/config_templates/config.yml | 8 ++
.../src/airflow/sdk/execution_time/supervisor.py | 64 +++++++--
.../task_sdk/execution_time/exec_probe_target.py | 30 ++++
.../task_sdk/execution_time/test_supervisor.py | 160 ++++++++++++++++++++-
7 files changed, 280 insertions(+), 19 deletions(-)
diff --git a/airflow-core/docs/security/security_model.rst
b/airflow-core/docs/security/security_model.rst
index 8e561051070..54db6f55603 100644
--- a/airflow-core/docs/security/security_model.rst
+++ b/airflow-core/docs/security/security_model.rst
@@ -591,8 +591,11 @@ model — Airflow does not enforce these natively.
For higher security, pass sensitive configuration values via environment
variables rather than
configuration files. Environment variables are inherently safer than
configuration files in
Airflow's worker processes because of a built-in protection: on Linux, the
supervisor process
- calls ``prctl(PR_SET_DUMPABLE, 0)`` before forking the task process, and
this flag is inherited
- by the forked child. This marks both processes as non-dumpable, which
prevents same-UID sibling
+ calls ``prctl(PR_SET_DUMPABLE, 0)`` before forking the task process; a
bare-forked child inherits
+ the flag, and a child started through ``exec`` restores it in its
bootstrap, before importing
+ Airflow, because ``execve`` resets it (``kernel.yama.ptrace_scope >= 1``
covers ``/proc/<pid>/mem``
+ and ``ptrace`` attach for the remaining interpreter-start window). This
marks both processes as
+ non-dumpable, which prevents same-UID sibling
processes from reading ``/proc/<pid>/environ``, ``/proc/<pid>/mem``, or
attaching via
``ptrace``. In contrast, configuration files on disk are readable by any
process running as
the same Unix user. Environment variables can also be scoped to individual
processes or
diff --git a/airflow-core/docs/security/workload.rst
b/airflow-core/docs/security/workload.rst
index 4b76169924c..c9686833213 100644
--- a/airflow-core/docs/security/workload.rst
+++ b/airflow-core/docs/security/workload.rst
@@ -85,8 +85,11 @@ Worker process memory protection (Linux)
''''''''''''''''''''''''''''''''''''''''
On Linux, the supervisor process calls ``prctl(PR_SET_DUMPABLE, 0)`` at the
start of
-``supervise_task()`` before forking the task process. This flag is inherited
by the forked
-child. Marking processes as non-dumpable prevents same-UID sibling processes
from reading
+``supervise_task()`` before forking the task process. A bare-forked child
inherits the flag;
+a child started through ``exec`` (macOS, or ``[core]
execute_tasks_new_python_interpreter``)
+restores it in its bootstrap, before importing Airflow, because ``execve``
resets it; for the remaining
+interpreter-start window, ``kernel.yama.ptrace_scope >= 1`` covers
``/proc/<pid>/mem`` and ``ptrace``
+attach. Marking processes as non-dumpable prevents same-UID sibling processes
from reading
``/proc/<pid>/mem``, ``/proc/<pid>/environ``, or ``/proc/<pid>/maps``, and
blocks
``ptrace(PTRACE_ATTACH)``. This is critical because each supervisor holds a
distinct JWT
token in memory — without this protection, a malicious task process running as
the same
diff --git a/airflow-core/newsfragments/72164.significant.rst
b/airflow-core/newsfragments/72164.significant.rst
new file mode 100644
index 00000000000..29532d28699
--- /dev/null
+++ b/airflow-core/newsfragments/72164.significant.rst
@@ -0,0 +1,23 @@
+``[core] execute_tasks_new_python_interpreter`` now applies to Airflow 3 task
processes
+
+On Airflow 3 the option had no effect on task execution (only the Edge worker
read it). When set to
+``True``, the task supervisor now ``exec``\ s a fresh interpreter right after
forking the task process,
+which prevents the fork from inheriting a lock held by a supervisor thread (a
permanent hang at the
+task's first TLS call). Deployments that kept the option ``True`` from Airflow
2 get this behaviour,
+and its per-task interpreter start-up cost, on upgrade without further action;
set it to ``False`` to
+keep bare fork. The task process reads the global value, so a team-scoped Edge
setting alone does not
+turn it on. Edge workers with the option ``True`` already start a fresh
interpreter for the supervisor
+and will now start a second one for the task. The Dag processor, the triggerer
and task
+callbacks (which the callback supervisor runs from a closure that cannot be
named for an exec'd
+child) are not affected and keep bare fork.
+
+* Types of change
+
+ * [ ] Dag changes
+ * [x] Config changes
+ * [ ] API changes
+ * [ ] CLI changes
+ * [x] Behaviour changes
+ * [ ] Plugin changes
+ * [ ] Dependency changes
+ * [ ] Code interface changes
diff --git a/airflow-core/src/airflow/config_templates/config.yml
b/airflow-core/src/airflow/config_templates/config.yml
index 3601a94759c..b6774cc8657 100644
--- a/airflow-core/src/airflow/config_templates/config.yml
+++ b/airflow-core/src/airflow/config_templates/config.yml
@@ -223,6 +223,14 @@ core:
* ``False``: Execute via forking of the parent process
* ``True``: Spawning a new python process, slower than fork, but means
plugin changes picked
up by tasks straight away
+
+ On Airflow 3 this applies to the task process only. When ``True`` the
supervisor starts
+ the task in a fresh interpreter with ``exec`` right after forking it,
so the task cannot
+ inherit a lock a supervisor thread held at fork time (which can
otherwise hang it at its
+ first TLS call). The Dag processor, the triggerer and task callbacks
are not affected and
+ keep bare fork. Each task pays an interpreter start plus re-import.
The task process reads
+ the global value; Edge workers, which resolve it per team, already
start a fresh interpreter
+ for the supervisor when it is ``True`` and start a second one for the
task.
default: "False"
example: ~
version_added: 2.0.0
diff --git a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
index 0a480851285..ab4c60464a3 100644
--- a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
+++ b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
@@ -517,6 +517,22 @@ def _should_use_exec() -> bool:
return sys.platform in _FORK_EXEC_PLATFORMS
+def _task_process_uses_exec() -> bool:
+ """
+ Whether the task process should ``exec`` a fresh interpreter after the
fork.
+
+ Forced where bare fork is unsafe (macOS); elsewhere a deployment opts in
with
+ ``[core] execute_tasks_new_python_interpreter``. exec replaces the child's
address
+ space, so it cannot inherit a lock a supervisor thread held at fork time
(e.g.
+ OpenSSL's, which otherwise hangs the task at its first TLS call; #71707).
Only the
+ task process reads the option -- it has always described task execution --
so the Dag
+ processor (one child per file per parse loop) and the triggerer keep the
platform gate.
+ """
+ return _should_use_exec() or conf.getboolean(
+ "core", "execute_tasks_new_python_interpreter", fallback=False
+ )
+
+
def _resolve_child_target(dotted: str) -> Callable[[], None]:
"""
Resolve a ``module:qualname`` string to the callable the exec'd child runs.
@@ -531,15 +547,38 @@ def _resolve_child_target(dotted: str) -> Callable[[],
None]:
return pkgutil.resolve_name(dotted)
+# Runs in the exec'd child before anything else. execve reset PR_SET_DUMPABLE
(4 in
+# <linux/prctl.h>); restore it before the Airflow import so the window in
which a same-UID
+# sibling can open /proc/<pid>/mem or ptrace-attach is interpreter start only
(a descriptor
+# or attach taken in that window survives a later prctl -- the kernel checks
once, at open).
+# _child_exec_main() repeats the call as the logged fallback.
+_CHILD_EXEC_PRELUDE = """\
+import sys
+if sys.platform == "linux":
+ try:
+ import ctypes
+
+ ctypes.CDLL(None, use_errno=True).prctl(4, 0, 0, 0, 0)
+ except Exception:
+ pass
+"""
+_CHILD_EXEC_BOOTSTRAP = _CHILD_EXEC_PRELUDE + (
+ "from airflow.sdk.execution_time.supervisor import
_child_exec_main\n_child_exec_main()\n"
+)
+
+
def _child_exec_main():
"""
- Entry point for the child process when using fork+exec (macOS).
+ Entry point for the child process when using fork+exec.
After exec, FDs 0/1/2/3 are the requests/stdout/stderr/log sockets the
parent
placed there via dup2. The target to run is named in
``_AIRFLOW_CHILD_TARGET``
(``module:qualname``); it is rehydrated and handed to :func:`_fork_main`,
which
sets up the structured log channel from FD 3 exactly as the bare-fork path
does.
"""
+ # The bootstrap already restored PR_SET_DUMPABLE before importing Airflow;
this is the
+ # logged fallback (execve had reset what supervise_task() set before the
fork).
+ _make_process_nondumpable()
# FDs 0, 1, 2 were dup2'd onto the socketpairs before exec.
child_requests = socket(fileno=0)
child_stdout = socket(fileno=1)
@@ -690,9 +729,11 @@ class WatchedSubprocess:
"""
Fork and start a new subprocess with the specified target function.
- :param use_exec: If True, on platforms that need it (currently macOS),
- immediately ``os.execv`` a fresh Python interpreter after
``os.fork``.
- This avoids macOS fork-safety issues with Objective-C frameworks.
+ :param use_exec: If True, immediately ``os.execv`` a fresh Python
interpreter
+ after ``os.fork``: forced on platforms that need it (macOS, whose
Objective-C
+ frameworks are not fork-safe) and opted into for the task process
elsewhere via
+ ``[core] execute_tasks_new_python_interpreter`` (a lock a
supervisor thread
+ held at fork time cannot survive into a fresh address space).
``target`` is rehydrated in the exec'd child from its
``module:qualname``,
so any importable entry point (task execution, DAG processor,
triggerer)
is supported.
@@ -742,8 +783,8 @@ class WatchedSubprocess:
try:
if use_exec:
- # macOS: exec a fresh Python interpreter to drop the
inherited
- # ObjC/CoreFoundation state that is not fork-safe.
Redirect the
+ # exec a fresh Python interpreter to drop inherited state
that is not
+ # fork-safe (ObjC/CoreFoundation on macOS; a held lock
elsewhere). Redirect the
# socketpairs onto the fixed FDs the exec'd child
reconstructs:
# 0 (requests/stdin), 1 (stdout), 2 (stderr), 3
(structured logs).
# The source fds are always >= 3 (0/1/2 stay open in every
launch
@@ -760,12 +801,7 @@ class WatchedSubprocess:
os.set_inheritable(fd, True)
os.execv(
sys.executable,
- [
- sys.executable,
- "-c",
- "from airflow.sdk.execution_time.supervisor import
_child_exec_main;"
- " _child_exec_main()",
- ],
+ [sys.executable, "-c", _CHILD_EXEC_BOOTSTRAP],
)
# execv replaces the process -- unreachable on success
else:
@@ -1452,10 +1488,10 @@ class ActivitySubprocess(WatchedSubprocess):
**kwargs,
) -> Self:
"""Fork and start a new subprocess to execute the given task."""
- # Opt in to fork+exec on platforms that need it (currently macOS).
+ # fork+exec where the platform needs it (macOS) or the deployment
opted in.
# Tests override `target` with a local stub to exercise the base
# infrastructure; keep bare fork for those.
- use_exec = target is _subprocess_main and _should_use_exec()
+ use_exec = target is _subprocess_main and _task_process_uses_exec()
proc: Self = super().start(
id=what.id,
client=client,
diff --git a/task-sdk/tests/task_sdk/execution_time/exec_probe_target.py
b/task-sdk/tests/task_sdk/execution_time/exec_probe_target.py
new file mode 100644
index 00000000000..75affc0cc61
--- /dev/null
+++ b/task-sdk/tests/task_sdk/execution_time/exec_probe_target.py
@@ -0,0 +1,30 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Importable entry point for the fork+exec end-to-end test; it runs inside
the exec'd child."""
+
+from __future__ import annotations
+
+import sys
+
+from airflow.sdk.execution_time.comms import CommsDecoder
+
+
+def exec_probe_main() -> None:
+ """Stand-in for ``_subprocess_main``: consume the startup message, then
report over stdout."""
+ CommsDecoder()._get_response()
+ print("exec-probe-ok")
+ sys.stdout.flush()
diff --git a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
index 93422f4426a..8175e1ff228 100644
--- a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
+++ b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
@@ -51,6 +51,7 @@ from opentelemetry.trace import get_current_span
from pytest_unordered import unordered
from structlog.typing import FilteringBoundLogger
from task_sdk import FAKE_BUNDLE, make_client
+from task_sdk.execution_time import exec_probe_target
from uuid6 import uuid7
from airflow.executors.workloads import BundleInfo
@@ -4479,6 +4480,34 @@ def _drop_root_if_needed():
os.setuid(_NOBODY_UID)
[email protected](sys.platform != "linux", reason="PR_SET_DUMPABLE is
Linux-only")
+def test_exec_bootstrap_restores_nondumpable_before_airflow_import():
+ """The real exec bootstrap prelude sets PR_SET_DUMPABLE back to 0 (execve
resets it to 1)."""
+ probe = (
+ "import ctypes\n"
+ "libc = ctypes.CDLL(None, use_errno=True)\n"
+ "after_exec = libc.prctl(3, 0, 0, 0, 0)\n" # PR_GET_DUMPABLE, before
the prelude runs
+ + supervisor._CHILD_EXEC_PRELUDE
+ + "print(after_exec, libc.prctl(3, 0, 0, 0, 0))\n"
+ )
+ read_fd, write_fd = os.pipe()
+ pid = os.fork()
+ if pid == 0: # pragma: no cover - child
+ try:
+ os.close(read_fd)
+ os.dup2(write_fd, 1)
+ _make_process_nondumpable() # what supervise_task() does before
the fork
+ os.execv(sys.executable, [sys.executable, "-c", probe])
+ finally:
+ os._exit(1) # only reached if execv failed; never run pytest in
the child
+ os.close(write_fd)
+ with os.fdopen(read_fd) as out:
+ report = out.read().split()[-2:] # anything the exec'd interpreter
logs first is noise
+ _, status = os.waitpid(pid, 0)
+ assert status == 0, f"exec'd child exited with {status}"
+ assert report == ["1", "0"], f"dumpable flag after exec, then after the
prelude: {report}"
+
+
@pytest.mark.skipif(sys.platform != "linux", reason="PR_SET_DUMPABLE is
Linux-only")
def test_nondumpable_blocks_sibling_proc_read():
"""A sibling process (same non-root UID) cannot read /proc/<pid>/environ
or /proc/<pid>/mem of a nondumpable process."""
@@ -4595,6 +4624,61 @@ def
test_api_client_clears_dag_bag_override_when_dag_is_none():
in_process_api_server.cache_clear()
[email protected](
+ ("task_uses_exec", "target", "expected_use_exec"),
+ [
+ (True, supervisor._subprocess_main, True),
+ (False, supervisor._subprocess_main, False),
+ (True, lambda: None, False),
+ ],
+)
+def test_activity_start_opts_into_fork_exec(monkeypatch, mocker,
task_uses_exec, target, expected_use_exec):
+ """ActivitySubprocess.start passes the task-process decision as use_exec,
real entry point only."""
+ monkeypatch.setattr(supervisor, "_task_process_uses_exec", lambda:
task_uses_exec)
+ base_start = mocker.patch(
+ "airflow.sdk.execution_time.supervisor.WatchedSubprocess.start",
return_value=MagicMock()
+ )
+
+ ActivitySubprocess.start(
+ dag_rel_path=os.devnull,
+ bundle_info=FAKE_BUNDLE,
+ what=TaskInstance(
+ id="4d828a62-a417-4936-a7a6-2b3fabacecab",
+ task_id="b",
+ dag_id="c",
+ run_id="d",
+ try_number=1,
+ dag_version_id=uuid7(),
+ queue="default",
+ ),
+ client=MagicMock(spec=sdk_client.Client),
+ target=target,
+ logger=MagicMock(),
+ )
+
+ assert base_start.call_args.kwargs["use_exec"] is expected_use_exec
+ assert base_start.call_args.kwargs["target"] is target
+
+
+class TestTaskProcessUsesExec:
+ """The config opt-in for fork+exec of the task process where the platform
does not force it."""
+
+ @pytest.mark.parametrize(
+ ("platform", "config_value", "expected"),
+ [
+ ("darwin", None, True),
+ ("darwin", "False", True),
+ ("linux", None, False),
+ ("linux", "False", False),
+ ("linux", "True", True),
+ ],
+ )
+ def test_task_process_uses_exec(self, monkeypatch, platform, config_value,
expected):
+ monkeypatch.setattr(supervisor.sys, "platform", platform)
+ with conf_vars({("core", "execute_tasks_new_python_interpreter"):
config_value}):
+ assert supervisor._task_process_uses_exec() is expected
+
+
class TestResolveChildTarget:
"""Test rehydrating the exec'd child's entry point from
_AIRFLOW_CHILD_TARGET."""
@@ -4620,8 +4704,82 @@ class TestResolveChildTarget:
@pytest.mark.usefixtures("disable_capturing")
+def test_fork_exec_bootstrap_runs_an_importable_target_end_to_end(
+ captured_logs, time_machine, monkeypatch, client_with_ti_start
+):
+ """
+ Drive the real ``os.execv`` bootstrap: the fresh interpreter runs
``_CHILD_EXEC_BOOTSTRAP``,
+ rebuilds FDs 0-3, rehydrates the target by name and hands it to
``_fork_main``.
+
+ The probe stands in for ``_subprocess_main`` rather than running the task
runner: the
+ suite stubs plugin loading in-process (conftest ``_get_plugins``), which a
bare-forked
+ child inherits and a fresh interpreter cannot.
+ """
+ tests_dir =
os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+ monkeypatch.setenv(
+ "PYTHONPATH", os.pathsep.join(p for p in (tests_dir,
os.environ.get("PYTHONPATH", "")) if p)
+ )
+ monkeypatch.setattr(supervisor, "_task_process_uses_exec", lambda: True)
+ # ActivitySubprocess.start only execs its own entry point; let the probe
be that entry point.
+ monkeypatch.setattr(supervisor, "_subprocess_main",
exec_probe_target.exec_probe_main)
+ time_machine.move_to(timezone.datetime(2024, 11, 7, 12, 34, 56, 78901),
tick=False)
+
+ proc = ActivitySubprocess.start(
+ dag_rel_path=os.devnull,
+ bundle_info=FAKE_BUNDLE,
+ what=TaskInstance(
+ id="4d828a62-a417-4936-a7a6-2b3fabacecab",
+ task_id="b",
+ dag_id="c",
+ run_id="d",
+ try_number=1,
+ dag_version_id=uuid7(),
+ queue="default",
+ ),
+ client=client_with_ti_start,
+ target=exec_probe_target.exec_probe_main,
+ )
+
+ assert proc.wait() == 0, captured_logs
+ assert {
+ "logger": "task.stdout",
+ "event": "exec-probe-ok",
+ "level": "info",
+ "timestamp": "2024-11-07T12:34:56.078901Z",
+ } in captured_logs
+
+
class TestChildExecMain:
- """Test the macOS fork+exec child entry point."""
+ """Test the fork+exec child entry point."""
+
+ def test_bootstrap_is_prelude_then_entry_point(self):
+ assert
supervisor._CHILD_EXEC_BOOTSTRAP.startswith(supervisor._CHILD_EXEC_PRELUDE)
+ assert
supervisor._CHILD_EXEC_BOOTSTRAP.rstrip().endswith("_child_exec_main()")
+ compile(supervisor._CHILD_EXEC_BOOTSTRAP, "<bootstrap>", "exec")
+
+ def test_prelude_survives_an_interpreter_without_ctypes(self):
+ """Without _ctypes the prelude must fall through to the logged
fallback, not kill the task."""
+ probe = (
+ "import sys\nsys.modules['_ctypes'] = None\n" +
supervisor._CHILD_EXEC_PRELUDE + "print('ok')\n"
+ )
+ result = subprocess.run(
+ [sys.executable, "-c", probe], capture_output=True, text=True,
timeout=60, check=False
+ )
+ assert result.returncode == 0, result.stderr
+ assert result.stdout.strip() == "ok"
+
+ def test_reapplies_nondumpable_before_running_the_target(self,
monkeypatch):
+ """The logged fallback for the bootstrap prelude runs before
_fork_main hands off."""
+ calls: list[str] = []
+ monkeypatch.setattr(supervisor, "_make_process_nondumpable", lambda:
calls.append("nondumpable"))
+ monkeypatch.setattr(supervisor, "_fork_main", lambda *a:
calls.append("fork_main"))
+ monkeypatch.setattr(supervisor, "_resolve_child_target", lambda
dotted: supervisor._subprocess_main)
+ monkeypatch.setattr(supervisor, "socket", lambda fileno: MagicMock())
+ monkeypatch.setenv("_AIRFLOW_CHILD_TARGET",
"airflow.sdk.execution_time.supervisor:_subprocess_main")
+
+ supervisor._child_exec_main()
+
+ assert calls == ["nondumpable", "fork_main"]
def test_uses_fds_0123_and_inherits_log_channel(self, monkeypatch):
"""_child_exec_main wraps FDs 0/1/2 as sockets and passes log_fd=3
(inherited log channel)."""
