uranusjr commented on code in PR #75561:
URL: https://github.com/apache/airflow/pull/75561#discussion_r4385390915
##########
task-sdk/src/airflow/sdk/coordinators/_subprocess.py:
##########
@@ -206,31 +216,96 @@ def _build_runtime_env() -> dict[str, str]:
}
+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(
+ logger: 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 to the log the way a connected runtime's output does,
rather than into the error
+ message, so every line keeps its own level and timestamp.
+ """
+ for key, soc in drains.items():
+ output = drained[soc] + _read_available(soc, drain_size)
+ # A connected runtime's stderr is logged as an error and its stdout as
info.
+ write = logger.error if key == "stderr" else logger.info
+ name = f"task.{key}"
+ if len(output) > _STARTUP_OUTPUT_LIMIT:
+ output = output[-_STARTUP_OUTPUT_LIMIT:]
+ write(_STARTUP_OUTPUT_TRUNCATED, logger=name)
+ for line in output.decode(errors="replace").splitlines():
+ write(line, logger=name)
Review Comment:
This reimplements `forward_to_log` and `_create_log_forwarder` from
supervisor.py but is slightly different. Reuse `forward_to_log(target_loggers,
f"task.{key}", level)` and pass the target loggers in instead. The comment on
L245 then goes away too.
--
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]
