ChickenBenny commented on code in PR #75561:
URL: https://github.com/apache/airflow/pull/75561#discussion_r4377742560
##########
task-sdk/src/airflow/sdk/coordinators/_subprocess.py:
##########
@@ -206,6 +206,48 @@ def _build_runtime_env() -> dict[str, str]:
}
+# The tail of each stream is kept: a runtime that fails to start reports why
at the end.
+_STARTUP_OUTPUT_LIMIT = 4096
+
+
+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 = []
+ with contextlib.suppress(BlockingIOError, OSError):
+ while chunk := soc.recv(drain_size):
+ chunks.append(chunk)
+ return b"".join(chunks)
+
+
+def _build_launch_error(
+ reason: str, drains: dict[str, socket.socket], drained:
dict[socket.socket, bytes], drain_size: int
+) -> TaskLaunchError:
+ """Build the error for a runtime that did not connect, with the output it
wrote before that."""
+ message = f"The runtime {reason}"
+ for key, soc in drains.items():
+ output = drained[soc] + _read_available(soc, drain_size)
+ if text :=
output[-_STARTUP_OUTPUT_LIMIT:].decode(errors="replace").strip():
+ message += f"\n{key}:\n{text}"
+ return TaskLaunchError(message)
+
+
[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.
+ """
+ try:
+ yield
+ except TaskLaunchError:
+ raise
+ except Exception as e:
Review Comment:
Thanks! I ended up doing both of your suggestions:
- OSError / ValueError are still retryable. That covers the "environment"
stuff.
- Anything else now becomes `TaskLaunchError(retryable=False)`, same as
`_is_native_dag_file` already does.
I left RuntimeError out of the retryable set on purpose, since
NotImplementedError and the scan-root guards are both RuntimeErrors.
--
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]
