This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new 9f5e63d61e6 [v3-4-test] Report language SDK runtime launch failures in
the task log (#75561) (#75741)
9f5e63d61e6 is described below
commit 9f5e63d61e6d53f3ab8268ad2fc3704a169ed5dd
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Tue Oct 27 16:14:27 2026 +0530
[v3-4-test] Report language SDK runtime launch failures in the task log
(#75561) (#75741)
* Add tests for language SDK tasks that cannot launch
* Report language SDK runtime launch failures in the task log
* Resolve artifact bundles the same way as native Dag bundles
* Fail coordinator bugs before launch without a retry
* Say why execute_task uses an ExitStack
The lock and the scan roots are entered inside the launch-error guard but
released only after wait() returns, which is the whole reason the stack is
there rather than a plain with.
* Fail an unsupported supervisor schema version without a retry
A worker whose Task SDK does not know the schema version an artifact speaks
cannot learn it by trying again, so the task should not burn its retries on
it. The check also moves out of the native Dag branch and now runs once for
every task, so an artifact found by scanning the bundle is rejected before
the runtime starts rather than by the attrs validator inside start().
* Write a failed launch's runtime output to the task log
The output a runtime wrote before it failed to connect went into the error
message, so the task log rendered it as one event whose continuation lines
carry no timestamp or level. It now goes to the log the way a connected
runtime's output does, one event per line under task.stdout and task.stderr,
and the message stays on one line.
The kept tail grows to 16 KiB, since a JVM or Node stack trace does not fit
in 4 KiB, and says so when it drops earlier output. Draining a runtime that
keeps printing while it never connects is capped, so it cannot hold up the
failure.
* Assert why a runtime that cannot be executed fails
The test matched the path with a wildcard, which hid whether the reason the
OS gave survives the wrap. It now asserts the whole message, and covers a
runtime that exists but cannot be executed as well as one that is missing.
_connect_without_output moves above the class that uses it, where every
other helper in the file sits.
* Keep the runtime's sockets tracked until the launch has finished
The pipes and the accepted connections were untracked as the argument to
_register_pipe_readers, so they were disowned before it ran. A failure in
it,
or in _on_child_started after it, left four sockets that nothing closed: the
tracker no longer owned them and the half-built subprocess was discarded.
They now stay with the tracker until the task instance has started, which is
the point 'self' takes them over and closes them in wait().
* Forward a failed launch's output through the supervisor's forwarder
_report_startup_output decoded and logged the lines itself, which repeated
forward_to_log with small differences. It now feeds forward_to_log instead,
so a failed launch and a connected runtime split lines and decode them the
same way.
The loggers to forward to come from WatchedSubprocess._target_loggers, split
out of _get_target_loggers so both callers share it: 'self' does not exist
yet when the runtime fails to connect, and the module logger the tuple ends
with belongs to supervisor.py.
* State the launch-error contract in terms of the task instance
TaskLaunchError said a coordinator raises it "before it starts the runtime",
but a runtime that exits or times out before connecting has already been
spawned. The invariant custom coordinators implement against is that the
task instance has not been started, which is what the supervisor does itself
when it handles the error.
The command builders now also say to raise TaskLaunchError with an explicit
retryable when the implementation knows the answer, rather than leaving it
to be guessed from the exception type.
* Drop the comment on execute_task's ExitStack
* Say what a cleared task does to a failed launch's output
"the log keeps output for a try that never ran" reads as if the output shows
up against the next try. It does not: the log file is rendered per try, so
the lines stay in the file of the try whose launch failed. What happens is
narrower, and worth naming: the supervisor stops without reporting a state,
so the file is never uploaded and is left on the worker.
* Keep a Dag file command's launch error and its retryability
* Expect a launch error when a runtime dies before connecting
The process group test landed on main while this branch was open, against
the RuntimeError _accept_connections raised there. This branch raises
TaskLaunchError instead, which is not a RuntimeError.
---------
(cherry picked from commit 21a8a98ba96804270b9535e7c2c3ff3a69281591)
Co-authored-by: ChickenBenny <[email protected]>
Co-authored-by: ZHE YOU LIU <[email protected]>
---
.../src/airflow/sdk/coordinators/_subprocess.py | 312 +++++++++++++-------
.../src/airflow/sdk/execution_time/coordinator.py | 14 +-
.../src/airflow/sdk/execution_time/supervisor.py | 17 +-
.../tests/task_sdk/coordinators/test_subprocess.py | 317 +++++++++++++++++++--
4 files changed, 520 insertions(+), 140 deletions(-)
diff --git a/task-sdk/src/airflow/sdk/coordinators/_subprocess.py
b/task-sdk/src/airflow/sdk/coordinators/_subprocess.py
index 5d5223072fe..1f792cd58a9 100644
--- a/task-sdk/src/airflow/sdk/coordinators/_subprocess.py
+++ b/task-sdk/src/airflow/sdk/coordinators/_subprocess.py
@@ -31,6 +31,7 @@ import contextlib
import ipaddress
import itertools
import json
+import logging
import os
import selectors
import signal
@@ -59,12 +60,13 @@ from airflow.sdk.execution_time.supervisor import (
ActivitySubprocess,
NeverRaised,
ProcessTracker,
+ forward_to_log,
signal_process_tree,
)
if TYPE_CHECKING:
import pathlib
- from collections.abc import Callable, Sequence
+ from collections.abc import Callable, Generator, Sequence
from typing import Self
from structlog.typing import FilteringBoundLogger
@@ -80,6 +82,16 @@ if TYPE_CHECKING:
log: FilteringBoundLogger =
structlog.get_logger(logger_name="coordinators.subprocess")
+# The tail of each stream is kept: a runtime that fails to start reports why
at the end.
+_STARTUP_OUTPUT_LIMIT = 16384
+
+# A runtime that keeps printing while it fails to connect is read no further
than this, so it cannot
+# hold up the failure.
+_STARTUP_DRAIN_BUDGET = 1024 * 1024
+
+_STARTUP_OUTPUT_TRUNCATED = f"[earlier output dropped, last
{_STARTUP_OUTPUT_LIMIT} bytes]".encode()
+
+
def _start_server() -> socket.socket:
server = socket.socket()
server.bind(("127.0.0.1", 0))
@@ -100,7 +112,7 @@ def _socket_address(value: tuple | str) -> tuple[str, int]
| None:
# * IPv4-mapped "::ffff:127.0.0.1" -> "127.0.0.1" (Linux, via
/proc/net/tcp6)
# * IPv4-compatible "::127.0.0.1" -> "127.0.0.1" (macOS, via
psutil)
# Otherwise the JVM's connection fails the check and every Java task is
rejected with
- # "process exited with 1 before connecting".
+ # "The runtime exited with 1 before connecting".
try:
parsed = ipaddress.ip_address(host)
except ValueError:
@@ -224,17 +236,89 @@ def _kill_leftover_group(pid: int) -> None:
os.killpg(pid, signal.SIGKILL)
+def _read_available(soc: socket.socket, drain_size: int) -> bytes:
+ """Return the data *soc* holds now, without waiting for more from a
process that may still run."""
+ soc.setblocking(False)
+ chunks = []
+ budget = _STARTUP_DRAIN_BUDGET
+ with contextlib.suppress(BlockingIOError, OSError):
+ while budget > 0 and (chunk := soc.recv(min(drain_size, budget))):
+ budget -= len(chunk)
+ chunks.append(chunk)
+ return b"".join(chunks)
+
+
+def _report_startup_output(
+ target_loggers: tuple[FilteringBoundLogger, ...],
+ drains: dict[str, socket.socket],
+ drained: dict[socket.socket, bytes],
+ drain_size: int,
+) -> None:
+ """
+ Write what a runtime printed before it failed to connect to the task log,
one event per line.
+
+ The output goes through the forwarder a connected runtime's output goes
through, rather than
+ into the error message, so every line keeps its own level and timestamp.
The task instance is
+ not started yet when this runs. If the supervisor then finds the task was
cleared while it was
+ queued, it stops without reporting a state, and these lines stay in this
try's local log file
+ without being uploaded. They never reach another try: the log file is per
try.
+ """
+ for key, soc in drains.items():
+ output = drained[soc] + _read_available(soc, drain_size)
+ level = logging.ERROR if key == "stderr" else logging.INFO
+ forward = forward_to_log(target_loggers, f"task.{key}", level)
+ next(forward)
+ if len(output) > _STARTUP_OUTPUT_LIMIT:
+ output = output[-_STARTUP_OUTPUT_LIMIT:]
+ forward.send(_STARTUP_OUTPUT_TRUNCATED)
+ for line in output.splitlines(keepends=True):
+ forward.send(line)
+
+
[email protected]
+def _raise_as_launch_error(prefix: str) -> Generator[None, None, None]:
+ """
+ Raise a failure in the block as a TaskLaunchError, so the supervisor fails
the task with its reason.
+
+ Use it only before the task instance is started: the supervisor starts the
task instance itself
+ when it fails one with a TaskLaunchError. A TaskLaunchError raised in the
block is kept as it is.
+ An OSError or ValueError, such as a missing runtime or artifact, or an
artifact that cannot run,
+ is retryable. Any other failure, such as a coordinator bug, is not:
retrying cannot fix it.
+ """
+ try:
+ yield
+ except TaskLaunchError:
+ raise
+ except (OSError, ValueError) as e:
+ raise TaskLaunchError(f"{prefix}: {e}") from e
+ except Exception as e:
+ raise TaskLaunchError(f"{prefix}: {e}", retryable=False) from e
+
+
def _accept_connections(
servers: dict[str, socket.socket],
drains: dict[str, socket.socket],
proc: subprocess.Popen,
*,
+ target_loggers: tuple[FilteringBoundLogger, ...],
max_wait: float = 10.0,
drain_size: int = 4096,
) -> tuple[dict[socket.socket, socket.socket], dict[socket.socket, bytes]]:
- """Block until the subprocess connects to servers, draining stdout/stderr
along the way."""
+ """
+ Block until the subprocess connects to servers, draining stdout/stderr
along the way.
+
+ :raises TaskLaunchError: when the subprocess exits or *max_wait* elapses
before it connects. What
+ the subprocess wrote to the *drains* until then goes to
*target_loggers* first.
+ """
accepted: dict[socket.socket, socket.socket] = {}
drained: dict[socket.socket, bytes] = {s: b"" for s in drains.values()}
+
+ def _fail(reason: str) -> NoReturn:
+ for s in accepted.values():
+ s.close()
+ _report_startup_output(target_loggers, drains, drained, drain_size)
+ raise TaskLaunchError(f"The runtime {reason}")
+
with selectors.DefaultSelector() as sel:
for key, soc in itertools.chain(servers.items(), drains.items()):
sel.register(soc, selectors.EVENT_READ, data=key)
@@ -242,13 +326,9 @@ def _accept_connections(
while len(accepted) < len(servers):
remaining = deadline - time.monotonic()
if remaining <= 0:
- for s in accepted.values():
- s.close()
- raise TimeoutError("process did not connect within timeout")
+ _fail(f"did not connect within {max_wait} seconds")
if proc.poll() is not None:
- for s in accepted.values():
- s.close()
- raise RuntimeError(f"process exited with {proc.returncode}
before connecting")
+ _fail(f"exited with {proc.returncode} before connecting")
for event, _ in sel.select(timeout=min(remaining, 1.0)):
soc = cast("socket.socket", event.fileobj)
if soc in drained:
@@ -369,60 +449,67 @@ class _PopenActivitySubprocess(ActivitySubprocess):
sentry_integration: str = "",
command: Sequence[str],
subprocess_schema_version: str | None = None,
+ subprocess_logs_to_stdout: bool = False,
startup_timeout: float = 10.0,
**kwargs,
) -> Self:
+ process_log = logger or structlog.get_logger(logger_name="task").bind()
+ # 'self' only exists once the runtime has connected, so the loggers
its output is forwarded
+ # to are resolved here, for a launch that fails before that.
+ target_loggers = cls._default_target_loggers(process_log,
subprocess_logs_to_stdout)
with _ResourceTracker(timeout=startup_timeout) as tracker:
- comm_server, logs_server = tracker.track(_start_server(),
_start_server())
- stdout_r, stdout_w = tracker.track(*socket.socketpair())
- stderr_r, stderr_w = tracker.track(*socket.socketpair())
-
- proc = subprocess.Popen(
- [
- *command,
- "--comm={0[0]}:{0[1]}".format(comm_server.getsockname()),
- "--logs={0[0]}:{0[1]}".format(logs_server.getsockname()),
- ],
- stdout=stdout_w.fileno(),
- stderr=stderr_w.fileno(),
- env=_build_runtime_env(),
- # Own process group, so the supervisor can signal everything
the runtime
- # spawned with os.killpg, as the Python task runner has done
since #65505.
- # Popen returns after exec, so the group exists before kill().
- # Must agree with new_process_group=True below.
- process_group=0,
- )
- tracker.track(proc)
- for soc in tracker.untrack(stdout_w, stderr_w):
- soc.close()
- log.info("Starting subprocess", pid=proc.pid)
-
- socks, drained = _accept_connections(
- {"comm": comm_server, "logs": logs_server},
- {"stdout": stdout_r, "stderr": stderr_r},
- proc,
- max_wait=startup_timeout,
- )
- tracker.track(*socks.values())
-
- self = cls(
- id=what.id,
- pid=proc.pid,
- process=PopenTracker(proc),
- process_log=logger or
structlog.get_logger(logger_name="task").bind(),
- start_time=time.monotonic(),
- stdin=socks[comm_server],
- subprocess_schema_version=subprocess_schema_version,
- # The supervisor signals the group; must agree with
process_group=0 above.
- new_process_group=True,
- comm_server=comm_server,
- logs_server=logs_server,
- **kwargs,
- )
- self._register_pipe_readers(
- *tracker.untrack(stdout_r, stderr_r, socks[comm_server],
socks[logs_server]),
- data=drained,
- )
+ with _raise_as_launch_error("The runtime cannot start"):
+ comm_server, logs_server = tracker.track(_start_server(),
_start_server())
+ stdout_r, stdout_w = tracker.track(*socket.socketpair())
+ stderr_r, stderr_w = tracker.track(*socket.socketpair())
+
+ proc = subprocess.Popen(
+ [
+ *command,
+
"--comm={0[0]}:{0[1]}".format(comm_server.getsockname()),
+
"--logs={0[0]}:{0[1]}".format(logs_server.getsockname()),
+ ],
+ stdout=stdout_w.fileno(),
+ stderr=stderr_w.fileno(),
+ env=_build_runtime_env(),
+ # Own process group, so the supervisor can signal
everything the runtime
+ # spawned with os.killpg, as the Python task runner has
done since #65505.
+ # Popen returns after exec, so the group exists before
kill().
+ # Must agree with new_process_group=True below.
+ process_group=0,
+ )
+ tracker.track(proc)
+ for soc in tracker.untrack(stdout_w, stderr_w):
+ soc.close()
+ log.info("Starting subprocess", pid=proc.pid)
+
+ socks, drained = _accept_connections(
+ {"comm": comm_server, "logs": logs_server},
+ {"stdout": stdout_r, "stderr": stderr_r},
+ proc,
+ target_loggers=target_loggers,
+ max_wait=startup_timeout,
+ )
+ tracker.track(*socks.values())
+
+ self = cls(
+ id=what.id,
+ pid=proc.pid,
+ process=PopenTracker(proc),
+ process_log=process_log,
+ start_time=time.monotonic(),
+ stdin=socks[comm_server],
+ subprocess_schema_version=subprocess_schema_version,
+ subprocess_logs_to_stdout=subprocess_logs_to_stdout,
+ # The supervisor signals the group; must agree with
process_group=0 above.
+ new_process_group=True,
+ comm_server=comm_server,
+ logs_server=logs_server,
+ **kwargs,
+ )
+ self._register_pipe_readers(
+ stdout_r, stderr_r, socks[comm_server],
socks[logs_server], data=drained
+ )
self._on_child_started(
ti=what,
dag_rel_path=dag_rel_path,
@@ -430,9 +517,18 @@ class _PopenActivitySubprocess(ActivitySubprocess):
sentry_integration=sentry_integration,
)
- # Untrack everything left. 'self' keeps track of these and closes
- # the servers when the subprocess exits in 'wait'.
- tracker.untrack(comm_server, logs_server, proc)
+ # 'self' owns everything from here: it closes the pipes and the
servers when the
+ # subprocess exits in 'wait'. Until this point the tracker still
owns them, so a
+ # failure registering the pipes or starting the task instance
releases them.
+ tracker.untrack(
+ stdout_r,
+ stderr_r,
+ socks[comm_server],
+ socks[logs_server],
+ comm_server,
+ logs_server,
+ proc,
+ )
return self
@@ -515,9 +611,9 @@ def _initialize_pinned_bundle(target: BundleInfo, logger:
FilteringBoundLogger)
return initialize_ti_bundle(BundleInfo(name=target.name, version=version,
version_data=version_data))
-def _resolve_task_bundle(bundle_info: BundleInfo, logger:
FilteringBoundLogger) -> BaseDagBundle:
+def _resolve_bundle(bundle_info: BundleInfo, logger: FilteringBoundLogger) ->
BaseDagBundle:
"""
- Materialize the Dag bundle of a task, pinned to a concrete version.
+ Materialize the Dag bundle of *bundle_info*, pinned to a concrete version.
*logger* is the task logger, so materialization failures surface in the
task log.
@@ -554,12 +650,18 @@ def _is_file_in_bundle(bundle: BaseDagBundle, rel_path:
str) -> bool:
def _require_supported_schema_version(version: str) -> None:
- """Raise ``ValueError`` unless the supervisor of this worker's Task SDK
knows *version*."""
+ """
+ Check that the supervisor of this worker's Task SDK knows *version*.
+
+ :raises TaskLaunchError: when it does not. Retrying cannot fix it, so this
does not retry.
+ """
try:
get_schema_version_migrator().resolve_version(version)
except ValueError as e:
- raise ValueError(
- f"uses supervisor schema version {version!r}, which this worker's
Task SDK does not support"
+ raise TaskLaunchError(
+ f"The runtime uses supervisor schema version {version!r}, which
this worker's Task SDK "
+ "does not support. Upgrade the worker's Airflow, or rebuild the
artifact against it.",
+ retryable=False,
) from e
@@ -599,17 +701,14 @@ class SubprocessCoordinator(BaseCoordinator):
That is the bundle named by ``task_handler_bundle_name``, or the
task's own
bundle when it is unset. *logger* is the task logger, so
materialization
failures surface in the task log.
+
+ :raises TaskLaunchError: when the bundle cannot be read.
"""
if self.task_handler_bundle_name is None:
target = bundle_info
else:
target = BundleInfo(name=self.task_handler_bundle_name)
-
- bundle = _initialize_pinned_bundle(target, logger)
- path = bundle.path
- if not path.exists():
- raise FileNotFoundError(f"Dag bundle {target.name!r} resolved to
{path}, which does not exist.")
- return bundle
+ return _resolve_bundle(target, logger)
def _get_scan_roots(self) -> tuple[pathlib.Path, ...]:
"""Return the artifact roots resolved for the active task or Dag
parse."""
@@ -630,6 +729,10 @@ class SubprocessCoordinator(BaseCoordinator):
:class:`_PopenActivitySubprocess` once the listening sockets have been
bound. A ``None`` schema version disables schema migration; messages
are
then exchanged at the runtime's native wire format.
+
+ Raise ``TaskLaunchError(..., retryable=...)`` when the implementation
knows whether
+ retrying can help. Any other exception is reported as a launch failure
with a
+ retryability guessed from its type.
"""
raise NotImplementedError
@@ -640,8 +743,9 @@ class SubprocessCoordinator(BaseCoordinator):
Build the command that runs *what* from the native Dag file at *path*.
Subclasses can retrieve the root of the Dag bundle holding *path* with
- :meth:`_get_scan_roots`. The contract is that of
:meth:`_build_execute_task_command`.
- Raise an exception when the file cannot run *what*.
+ :meth:`_get_scan_roots`. The contract is that of
:meth:`_build_execute_task_command`,
+ including how to report whether a failure is worth retrying. Raise an
exception when the
+ file cannot run *what*.
"""
raise NotImplementedError(f"{type(self).__name__} cannot run the tasks
of a native Dag")
@@ -748,8 +852,7 @@ class SubprocessCoordinator(BaseCoordinator):
"""
Check that *rel_path* is a file of *bundle* that this coordinator can
run, and return its command.
- :raises TaskLaunchError: when the file is missing or cannot be read,
cannot run *what*, or needs
- a supervisor schema version that this worker's Task SDK does not
support.
+ :raises TaskLaunchError: when the file is missing or cannot be read,
or cannot run *what*.
"""
version = f" at version {bundle.version!r}" if bundle.version is not
None else ""
try:
@@ -762,14 +865,10 @@ class SubprocessCoordinator(BaseCoordinator):
raise TaskLaunchError(
f"Dag file {rel_path!r} is not a file in Dag bundle
{bundle.name!r}{version}"
)
- try:
+ with _raise_as_launch_error(
+ f"Dag file {rel_path!r} in Dag bundle {bundle.name!r}{version}
cannot run"
+ ):
command, schema_version = self._build_dag_file_command(what=what,
path=bundle.path / rel_path)
- if schema_version is not None:
- _require_supported_schema_version(schema_version)
- except Exception as e:
- raise TaskLaunchError(
- f"Dag file {rel_path!r} in Dag bundle {bundle.name!r}{version}
cannot run: {e}"
- ) from e
return command, schema_version
def execute_task(
@@ -792,28 +891,35 @@ class SubprocessCoordinator(BaseCoordinator):
read other files of the bundle, such as the JARs on a Java classpath.
Any other task runs the
artifact found by scanning the bundle that holds the artifacts.
- :raises TaskLaunchError: when a native Dag file cannot run, before the
runtime starts.
+ :raises TaskLaunchError: when the runtime cannot start: a native Dag
file cannot run, the
+ artifact of any other task cannot be found or run, the artifact
needs a supervisor schema
+ version that this worker's Task SDK does not support, or the
runtime exits or times out
+ before it connects.
"""
task_logger = logger or log
rel_path = os.fspath(dag_rel_path)
- is_native = self._is_native_dag_file(rel_path, bundle_info.name)
- if is_native:
- bundle = _resolve_task_bundle(bundle_info, task_logger)
- else:
- bundle = self._resolve_artifact_bundle(bundle_info, task_logger)
- # Hold the version lock across start()/wait() so bundle cleanup cannot
- # rmtree a version this task is still reading from, mirroring
- # task_runner.main() for the Python task path.
- with (
- BundleVersionLock(bundle_name=bundle.name,
bundle_version=bundle.version),
- self._set_scan_roots([bundle.path]),
- ):
- if is_native:
- command, subprocess_schema_version =
self._build_native_task_command(
- what=what, bundle=bundle, rel_path=rel_path
- )
- else:
- command, subprocess_schema_version =
self._build_execute_task_command(what=what)
+ with contextlib.ExitStack() as stack:
+ with _raise_as_launch_error(
+ f"Cannot prepare the runtime of task {what.task_id!r} in Dag
{what.dag_id!r}"
+ ):
+ is_native = self._is_native_dag_file(rel_path,
bundle_info.name)
+ if is_native:
+ bundle = _resolve_bundle(bundle_info, task_logger)
+ else:
+ bundle = self._resolve_artifact_bundle(bundle_info,
task_logger)
+ # Hold the version lock across start()/wait() so bundle
cleanup cannot
+ # rmtree a version this task is still reading from, mirroring
+ # task_runner.main() for the Python task path.
+ stack.enter_context(BundleVersionLock(bundle_name=bundle.name,
bundle_version=bundle.version))
+ stack.enter_context(self._set_scan_roots([bundle.path]))
+ if is_native:
+ command, subprocess_schema_version =
self._build_native_task_command(
+ what=what, bundle=bundle, rel_path=rel_path
+ )
+ else:
+ command, subprocess_schema_version =
self._build_execute_task_command(what=what)
+ if subprocess_schema_version is not None:
+
_require_supported_schema_version(subprocess_schema_version)
process = _PopenActivitySubprocess.start(
what=what,
dag_rel_path=dag_rel_path,
diff --git a/task-sdk/src/airflow/sdk/execution_time/coordinator.py
b/task-sdk/src/airflow/sdk/execution_time/coordinator.py
index 78d13441b5f..c17f031a83a 100644
--- a/task-sdk/src/airflow/sdk/execution_time/coordinator.py
+++ b/task-sdk/src/airflow/sdk/execution_time/coordinator.py
@@ -70,10 +70,11 @@ class TaskLaunchError(Exception):
"""
A coordinator cannot launch the runtime of a task.
- A coordinator raises this from :meth:`BaseCoordinator.execute_task` before
it starts the runtime.
- The supervisor then fails the task instance and writes the message to the
task log. A *retryable*
- error leaves the task to retry while it has retries left. One that is not,
such as a routing
- mistake that retrying cannot fix, fails the task at once.
+ A coordinator raises this from :meth:`BaseCoordinator.execute_task` before
the task instance is
+ started; the runtime itself may already have been spawned. The supervisor
then starts the task
+ instance, fails it, and writes the message to the task log. A *retryable*
error leaves the task
+ to retry while it has retries left. One that is not, such as a routing
mistake that retrying
+ cannot fix, fails the task at once.
"""
def __init__(self, message: str, *, retryable: bool = True) -> None:
@@ -115,8 +116,9 @@ class BaseCoordinator:
This should execute the task and return a result.
- :raises TaskLaunchError: when the runtime cannot be launched. Raise it
before the runtime
- starts, so the supervisor can fail the task instance with the
message in its task log.
+ :raises TaskLaunchError: when the runtime cannot be launched. Raise it
only before the task
+ instance is started, so the supervisor can start it, fail it, and
put the message in its
+ task log. The runtime itself may already have been spawned.
"""
raise NotImplementedError
diff --git a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
index 0c198ab637a..42430e3fb47 100644
--- a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
+++ b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
@@ -1090,12 +1090,21 @@ class WatchedSubprocess:
length_prefixed_frame_reader(self.handle_requests(log),
on_close=self._on_socket_closed),
)
+ @staticmethod
+ def _default_target_loggers(
+ process_log: FilteringBoundLogger, subprocess_logs_to_stdout: bool
+ ) -> tuple[FilteringBoundLogger, ...]:
+ """
+ Return the loggers child process output is forwarded to, before any
subclass has its say.
+
+ Call this only where there is no instance yet. Everything else goes
through
+ :meth:`_get_target_loggers`, which a subclass can override.
+ """
+ return (process_log, log) if subprocess_logs_to_stdout else
(process_log,)
+
def _get_target_loggers(self) -> tuple[FilteringBoundLogger, ...]:
"""Return the loggers that child process output should be forwarded
to."""
- target_loggers: tuple[FilteringBoundLogger, ...] = (self.process_log,)
- if self.subprocess_logs_to_stdout:
- target_loggers += (log,)
- return target_loggers
+ return self._default_target_loggers(self.process_log,
self.subprocess_logs_to_stdout)
def _create_log_forwarder(
self,
diff --git a/task-sdk/tests/task_sdk/coordinators/test_subprocess.py
b/task-sdk/tests/task_sdk/coordinators/test_subprocess.py
index 779eac3711c..2a8a17ad411 100644
--- a/task-sdk/tests/task_sdk/coordinators/test_subprocess.py
+++ b/task-sdk/tests/task_sdk/coordinators/test_subprocess.py
@@ -147,7 +147,7 @@ class TestAcceptConnections:
mock_proc.poll.return_value = None
try:
- accepted, _ = _accept_connections({"comm": server}, {}, mock_proc)
+ accepted, _ = _accept_connections({"comm": server}, {}, mock_proc,
target_loggers=(log,))
assert server in accepted
accepted[server].close()
finally:
@@ -166,7 +166,9 @@ class TestAcceptConnections:
mock_proc.poll.return_value = None
try:
- accepted, drained = _accept_connections({"comm": comm_server,
"logs": logs_server}, {}, mock_proc)
+ accepted, drained = _accept_connections(
+ {"comm": comm_server, "logs": logs_server}, {}, mock_proc,
target_loggers=(log,)
+ )
assert set(accepted) == {comm_server, logs_server}
assert drained == {}
for sock in accepted.values():
@@ -183,7 +185,7 @@ class TestAcceptConnections:
mock_proc = MagicMock(spec=subprocess.Popen)
mock_proc.poll.return_value = None
try:
- _, drained = _accept_connections({"comm": server}, {}, mock_proc)
+ _, drained = _accept_connections({"comm": server}, {}, mock_proc,
target_loggers=(log,))
assert drained == {}
finally:
server.close()
@@ -197,7 +199,9 @@ class TestAcceptConnections:
mock_proc = MagicMock(spec=subprocess.Popen)
mock_proc.poll.return_value = None
try:
- _, drained = _accept_connections({"comm": server}, {"stdout":
drain_r}, mock_proc)
+ _, drained = _accept_connections(
+ {"comm": server}, {"stdout": drain_r}, mock_proc,
target_loggers=(log,)
+ )
assert drain_r in drained
finally:
drain_r.close()
@@ -218,7 +222,9 @@ class TestAcceptConnections:
mock_proc = MagicMock(spec=subprocess.Popen)
mock_proc.poll.return_value = None
try:
- _, drained = _accept_connections({"comm": server}, {"stdout":
drain_r}, mock_proc)
+ _, drained = _accept_connections(
+ {"comm": server}, {"stdout": drain_r}, mock_proc,
target_loggers=(log,)
+ )
assert drained[drain_r] == b"early output\n"
finally:
drain_r.close()
@@ -230,19 +236,19 @@ class TestAcceptConnections:
mock_proc = MagicMock(spec=subprocess.Popen)
mock_proc.poll.return_value = None
try:
- with pytest.raises(TimeoutError, match="did not connect within
timeout"):
- _accept_connections({"comm": server}, {}, mock_proc,
max_wait=0.05)
+ with pytest.raises(TaskLaunchError, match="^The runtime did not
connect within 0.05 seconds$"):
+ _accept_connections({"comm": server}, {}, mock_proc,
target_loggers=(log,), max_wait=0.05)
finally:
server.close()
- def test_raises_runtime_error_if_process_exits_before_connecting(self):
+ def test_raises_launch_error_if_process_exits_before_connecting(self):
server = _start_server()
mock_proc = MagicMock(spec=subprocess.Popen)
mock_proc.poll.return_value = 1
mock_proc.returncode = 1
try:
- with pytest.raises(RuntimeError, match="process exited with 1"):
- _accept_connections({"comm": server}, {}, mock_proc)
+ with pytest.raises(TaskLaunchError, match="^The runtime exited
with 1 before connecting$"):
+ _accept_connections({"comm": server}, {}, mock_proc,
target_loggers=(log,))
finally:
server.close()
@@ -258,7 +264,7 @@ class TestAcceptConnections:
mock_proc.poll.return_value = None
try:
- accepted, _ = _accept_connections({"comm": server}, {}, mock_proc)
+ accepted, _ = _accept_connections({"comm": server}, {}, mock_proc,
target_loggers=(log,))
accepted[server].sendall(b"hello")
assert client.recv(5) == b"hello"
accepted[server].close()
@@ -275,7 +281,7 @@ class TestAcceptConnections:
mock_proc = MagicMock(spec=subprocess.Popen)
mock_proc.poll.return_value = None
try:
- accepted, _ = _accept_connections({"comm": server}, {}, mock_proc)
+ accepted, _ = _accept_connections({"comm": server}, {}, mock_proc,
target_loggers=(log,))
# Key must be the socket object itself, not the string "comm"
assert server in accepted
assert "comm" not in accepted
@@ -295,7 +301,7 @@ class TestAcceptConnections:
mock_proc.poll.return_value = None
try:
- accepted, _ = _accept_connections({"comm": server}, {}, mock_proc)
+ accepted, _ = _accept_connections({"comm": server}, {}, mock_proc,
target_loggers=(log,))
assert mock_child_connection_check.call_count == 2
assert server in accepted
accepted[server].close()
@@ -325,7 +331,7 @@ sock.recv(1)
child_proc = self._start_connector_process(addr, delay=0.05)
try:
- accepted, _ = _accept_connections({"comm": server}, {}, child_proc)
+ accepted, _ = _accept_connections({"comm": server}, {},
child_proc, target_loggers=(log,))
accepted[server].sendall(b"x")
accepted[server].close()
assert child_proc.wait(timeout=5) == 0
@@ -364,7 +370,7 @@ class TestConnectionFromProcess:
``::ffff:127.0.0.1`` in ``/proc/net/tcp6``. The AF_INET supervisor
socket's
``getpeername()`` reports plain ``127.0.0.1``, so the ownership check
must
treat the mapped and plain forms as the same address -- otherwise every
- Java task is rejected with "process exited with 1 before connecting".
+ Java task is rejected with "The runtime exited with 1 before
connecting".
"""
server = _start_server()
_, port = server.getsockname()
@@ -1266,7 +1272,7 @@ class TestPopenActivitySubprocessProcessGroup:
def
test_a_runtime_that_dies_before_connecting_does_not_leave_its_child(self,
tmp_path):
"""The runtime is reaped before start() fails, so its group is swept
by the leader's pid."""
pid_file = tmp_path / "child.pid"
- with pytest.raises(RuntimeError, match="before connecting"):
+ with pytest.raises(TaskLaunchError, match="before connecting"):
_PopenActivitySubprocess.start(
what=_make_ti(),
dag_rel_path="bundle",
@@ -1370,10 +1376,247 @@ class TestResolveArtifactBundle:
mock_initialize.return_value = _make_bundle(tmp_path / "nope",
version="v1")
coordinator = _StubSubprocessCoordinator(command=["x"],
task_handler_bundle_name="artifacts")
- with pytest.raises(FileNotFoundError, match="does not exist"):
+ with pytest.raises(TaskLaunchError, match="does not exist"):
coordinator._resolve_artifact_bundle(MagicMock(spec=BundleInfo),
log)
+class _MissingArtifactCoordinator(_StubSubprocessCoordinator):
+ def _build_execute_task_command(self, *, what):
+ raise FileNotFoundError("cannot find a JAR with Main-Class metadata in
/bundle")
+
+
+class _BrokenCoordinator(_StubSubprocessCoordinator):
+ def _build_execute_task_command(self, *, what):
+ raise RuntimeError("coordinator bug")
+
+
+def _connect_without_output(servers, drains, proc, **kwargs):
+ """Stand in for _accept_connections when the runtime connects without
writing anything."""
+ accepted = {server: MagicMock(spec=socket.socket) for server in
servers.values()}
+ return accepted, {soc: b"" for soc in drains.values()}
+
+
+class TestExecuteTaskLaunchFailure:
+ """A task whose runtime cannot be launched fails with TaskLaunchError, so
the supervisor reports it."""
+
+ @pytest.mark.usefixtures("artifact_bundle")
+ def test_fails_with_a_retry_when_the_artifact_cannot_be_found(self,
mock_client):
+ coordinator = _MissingArtifactCoordinator(command=["/runtime"])
+
+ with pytest.raises(TaskLaunchError, match="cannot find a JAR with
Main-Class metadata") as raised:
+ coordinator.execute_task(
+ what=_make_ti(),
+ dag_rel_path="dag.py",
+ bundle_info=BundleInfo(name="dags"),
+ client=mock_client,
+ subprocess_logs_to_stdout=False,
+ )
+
+ assert raised.value.retryable is True
+ assert isinstance(raised.value.__cause__, FileNotFoundError)
+
+ @pytest.mark.usefixtures("artifact_bundle")
+ def test_fails_without_a_retry_for_a_coordinator_bug(self, mock_client):
+ coordinator = _BrokenCoordinator(command=["/runtime"])
+
+ with pytest.raises(TaskLaunchError, match="coordinator bug") as raised:
+ coordinator.execute_task(
+ what=_make_ti(),
+ dag_rel_path="dag.py",
+ bundle_info=BundleInfo(name="dags"),
+ client=mock_client,
+ subprocess_logs_to_stdout=False,
+ )
+
+ assert raised.value.retryable is False
+ assert isinstance(raised.value.__cause__, RuntimeError)
+
+ @patch(
+ "airflow.sdk.coordinators._subprocess.initialize_ti_bundle",
+ autospec=True,
+ side_effect=RuntimeError("Error cloning repository"),
+ )
+ def test_fails_with_a_retry_when_the_artifact_bundle_cannot_be_read(self,
_, mock_client):
+ coordinator = _StubSubprocessCoordinator(command=["/runtime"],
task_handler_bundle_name="artifacts")
+
+ with pytest.raises(
+ TaskLaunchError, match="Dag bundle 'artifacts' cannot be read:
Error cloning repository"
+ ) as raised:
+ coordinator.execute_task(
+ what=_make_ti(),
+ dag_rel_path="dag.py",
+ bundle_info=BundleInfo(name="dags"),
+ client=mock_client,
+ subprocess_logs_to_stdout=False,
+ )
+
+ assert raised.value.retryable is True
+ assert coordinator.recorded_roots == []
+
+ @pytest.mark.usefixtures("artifact_bundle")
+ @pytest.mark.parametrize(
+ ("script", "startup_timeout", "reason", "output"),
+ [
+ pytest.param(
+ "echo 'cannot load main class' >&2; exit 3",
+ 10.0,
+ "exited with 3",
+ "cannot load main class",
+ id="exits",
+ ),
+ pytest.param(
+ "echo 'still starting' >&2; sleep 30",
+ 0.5,
+ "did not connect",
+ "still starting",
+ id="times-out",
+ ),
+ ],
+ )
+ def
test_fails_with_a_retry_and_logs_the_runtime_output_when_the_runtime_does_not_connect(
+ self, mock_client, captured_logs, script, startup_timeout, reason,
output
+ ):
+ coordinator = _StubSubprocessCoordinator(
+ command=["sh", "-c", script], task_startup_timeout=startup_timeout
+ )
+
+ with pytest.raises(TaskLaunchError, match=reason) as raised:
+ coordinator.execute_task(
+ what=_make_ti(),
+ dag_rel_path="dag.py",
+ bundle_info=BundleInfo(name="dags"),
+ client=mock_client,
+ subprocess_logs_to_stdout=False,
+ )
+
+ assert raised.value.retryable is True
+ # The output belongs in the task log as its own lines, not folded into
the message.
+ assert "\n" not in str(raised.value)
+ assert ("task.stderr", "error", output) in [
+ (entry.get("logger"), entry["level"], entry["event"]) for entry in
captured_logs
+ ]
+ mock_client.task_instances.start.assert_not_called()
+
+ @pytest.mark.usefixtures("artifact_bundle")
+ @pytest.mark.parametrize(
+ ("executable", "expected"),
+ [
+ pytest.param(False, "No such file or directory", id="missing"),
+ pytest.param(True, "Permission denied", id="not-executable"),
+ ],
+ )
+ def test_fails_with_a_retry_when_the_runtime_cannot_be_executed(
+ self, mock_client, tmp_path, executable, expected
+ ):
+ runtime = tmp_path / "runtime"
+ if executable:
+ runtime.write_text("#!/bin/sh\n")
+ runtime.chmod(0o644)
+ coordinator = _StubSubprocessCoordinator(command=[str(runtime)])
+
+ with pytest.raises(TaskLaunchError) as raised:
+ coordinator.execute_task(
+ what=_make_ti(),
+ dag_rel_path="dag.py",
+ bundle_info=BundleInfo(name="dags"),
+ client=mock_client,
+ subprocess_logs_to_stdout=False,
+ )
+
+ message = str(raised.value)
+ assert message == f"The runtime cannot start: {raised.value.__cause__}"
+ assert expected in message
+ assert str(runtime) in message
+ assert raised.value.retryable is True
+ assert isinstance(raised.value.__cause__, OSError)
+ mock_client.task_instances.start.assert_not_called()
+
+ @pytest.mark.usefixtures("artifact_bundle")
+ def
test_fails_without_a_retry_for_a_schema_version_the_worker_does_not_support(self,
mock_client):
+ coordinator = _StubSubprocessCoordinator(command=["/runtime"],
schema_version="1999-01-01")
+
+ with pytest.raises(TaskLaunchError) as raised:
+ coordinator.execute_task(
+ what=_make_ti(),
+ dag_rel_path="dag.py",
+ bundle_info=BundleInfo(name="dags"),
+ client=mock_client,
+ subprocess_logs_to_stdout=False,
+ )
+
+ assert str(raised.value).startswith(
+ "The runtime uses supervisor schema version '1999-01-01', which
this worker's Task SDK "
+ "does not support."
+ )
+ assert raised.value.retryable is False
+ mock_client.task_instances.start.assert_not_called()
+
+ @pytest.mark.usefixtures("artifact_bundle")
+ @pytest.mark.parametrize(
+ ("failing", "expected"),
+ [
+ pytest.param("_register_pipe_readers", TaskLaunchError,
id="registering-the-pipes"),
+ pytest.param("_on_child_started", RuntimeError,
id="starting-the-task-instance"),
+ ],
+ )
+ @patch("airflow.sdk.coordinators._subprocess._accept_connections",
autospec=True)
+ def
test_releases_the_runtime_sockets_when_the_launch_fails_after_they_are_open(
+ self, mock_accept, mock_client, failing, expected
+ ):
+ """The sockets stay with the tracker until 'self' has taken them over,
so nothing leaks."""
+ mock_accept.side_effect = _connect_without_output
+ registered: list = []
+
+ def record(_self, *socks, data):
+ registered.extend(socks)
+ if failing == "_register_pipe_readers":
+ raise RuntimeError("boom")
+
+ coordinator = _StubSubprocessCoordinator(command=["sleep", "30"])
+
+ with (
+ patch.object(ActivitySubprocess, "_register_pipe_readers",
autospec=True, side_effect=record),
+ patch.object(
+ ActivitySubprocess, "_on_child_started", autospec=True,
side_effect=RuntimeError("boom")
+ ),
+ pytest.raises(expected, match="boom"),
+ ):
+ coordinator.execute_task(
+ what=_make_ti(),
+ dag_rel_path="dag.py",
+ bundle_info=BundleInfo(name="dags"),
+ client=mock_client,
+ subprocess_logs_to_stdout=False,
+ )
+
+ # _register_pipe_readers takes stdout and stderr, then the accepted
comm and logs connections.
+ stdout, stderr, *connections = registered
+ assert (stdout.fileno(), stderr.fileno()) == (-1, -1)
+ for connection in connections:
+ connection.close.assert_called()
+
+ @pytest.mark.usefixtures("artifact_bundle")
+ @patch.object(
+ ActivitySubprocess, "_on_child_started", autospec=True,
side_effect=RuntimeError("API unreachable")
+ )
+ @patch.object(ActivitySubprocess, "_register_pipe_readers", autospec=True)
+ @patch("airflow.sdk.coordinators._subprocess._accept_connections",
autospec=True)
+ def test_does_not_turn_a_failure_after_the_launch_into_a_launch_error(
+ self, mock_accept, _, __, mock_client
+ ):
+ mock_accept.side_effect = _connect_without_output
+ coordinator = _StubSubprocessCoordinator(command=["sleep", "30"])
+
+ with pytest.raises(RuntimeError, match="^API unreachable$"):
+ coordinator.execute_task(
+ what=_make_ti(),
+ dag_rel_path="dag.py",
+ bundle_info=BundleInfo(name="dags"),
+ client=mock_client,
+ subprocess_logs_to_stdout=False,
+ )
+
+
class TestExecuteTaskBundleWiring:
"""execute_task passes the task bundle, forwards resolved roots, and holds
the version lock."""
@@ -1452,8 +1695,10 @@ class _NativeStubCoordinator(_StubSubprocessCoordinator):
@attrs.define(kw_only=True)
class _BrokenNativeCoordinator(_NativeStubCoordinator):
+ error: Exception
+
def _build_dag_file_command(self, *, what, path):
- raise ValueError("no Main-Class")
+ raise self.error
@attrs.define(kw_only=True)
@@ -1704,29 +1949,46 @@ class TestExecuteTaskNativeDagFile:
assert coordinator.recorded_dag_files == []
@pytest.mark.usefixtures("mock_start", "mock_lock", "mock_initialize")
- def test_fails_with_a_retry_when_the_dag_file_cannot_run(self,
mock_client):
- coordinator = _BrokenNativeCoordinator(command=["/runtime"])
+ @pytest.mark.parametrize(
+ ("error", "retryable"),
+ [
+ pytest.param(ValueError("no Main-Class"), True, id="cannot-run"),
+ pytest.param(RuntimeError("no Main-Class"), False,
id="coordinator-bug"),
+ ],
+ )
+ def test_fails_when_the_dag_file_cannot_run(self, mock_client, error,
retryable):
+ coordinator = _BrokenNativeCoordinator(command=["/runtime"],
error=error)
with _native_dag_files("native"), pytest.raises(TaskLaunchError) as
raised:
self._execute(coordinator, mock_client)
- assert raised.value.retryable is True
+ assert raised.value.retryable is retryable
assert str(raised.value) == (
"Dag file 'sub/dag.native' in Dag bundle 'dags' at version 'v9'
cannot run: no Main-Class"
)
- assert isinstance(raised.value.__cause__, ValueError)
+ assert raised.value.__cause__ is error
+
+ @pytest.mark.usefixtures("mock_start", "mock_lock", "mock_initialize")
+ def test_keeps_the_launch_error_of_the_dag_file_command(self, mock_client):
+ error = TaskLaunchError("wrong runtime", retryable=False)
+ coordinator = _BrokenNativeCoordinator(command=["/runtime"],
error=error)
+
+ with _native_dag_files("native"), pytest.raises(TaskLaunchError) as
raised:
+ self._execute(coordinator, mock_client)
+
+ assert raised.value is error
@pytest.mark.usefixtures("mock_start", "mock_lock", "mock_initialize")
- def
test_fails_with_a_retry_for_a_schema_version_the_worker_does_not_support(self,
mock_client):
+ def
test_fails_without_a_retry_for_a_schema_version_the_worker_does_not_support(self,
mock_client):
coordinator = _NativeStubCoordinator(command=["/runtime"],
schema_version="1999-01-01")
with _native_dag_files("native"), pytest.raises(TaskLaunchError) as
raised:
self._execute(coordinator, mock_client)
- assert raised.value.retryable is True
+ assert raised.value.retryable is False
assert str(raised.value).startswith(
- "Dag file 'sub/dag.native' in Dag bundle 'dags' at version 'v9'
cannot run: "
- "uses supervisor schema version '1999-01-01', which this worker's
Task SDK does not support"
+ "The runtime uses supervisor schema version '1999-01-01', which
this worker's Task SDK "
+ "does not support."
)
@pytest.mark.usefixtures("mock_start", "mock_lock")
@@ -1828,7 +2090,7 @@ class TestGetScanRoots:
def test_execute_task_is_not_reentrant(self, mock_client, tmp_path):
coordinator = _StubSubprocessCoordinator(command=["/bin/true"])
with coordinator._set_scan_roots([tmp_path]):
- with pytest.raises(RuntimeError, match="not re-entrant"):
+ with pytest.raises(TaskLaunchError, match="not re-entrant") as
raised:
coordinator.execute_task(
what=_make_ti(),
dag_rel_path="dag.py",
@@ -1836,6 +2098,7 @@ class TestGetScanRoots:
client=mock_client,
subprocess_logs_to_stdout=False,
)
+ assert isinstance(raised.value.__cause__, RuntimeError)
@attrs.define(kw_only=True)
