uranusjr commented on code in PR #75561:
URL: https://github.com/apache/airflow/pull/75561#discussion_r4386561093
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -1090,12 +1090,15 @@ def _register_pipe_readers(
length_prefixed_frame_reader(self.handle_requests(log),
on_close=self._on_socket_closed),
)
- def _get_target_loggers(self) -> tuple[FilteringBoundLogger, ...]:
+ @staticmethod
+ def _target_loggers(
+ process_log: FilteringBoundLogger, subprocess_logs_to_stdout: bool
+ ) -> tuple[FilteringBoundLogger, ...]:
Review Comment:
This new method sits in a polymorphic hook (DagFileProcessorProcess
overrides `_get_target_loggers`), and the startup path calls it directly, so a
subclass that overrides `_get_target_loggers` isn't consulted for output from a
runtime that never connected.
There's no `_PopenActivitySubprocess` override today, so behavior is
unchanged. Give the function a better name (e.g. _default_target_loggers) so
the next override doesn't silently miss this path.
##########
task-sdk/src/airflow/sdk/coordinators/_subprocess.py:
##########
@@ -372,67 +452,81 @@ def start( # type: ignore[override]
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._target_loggers(process_log,
kwargs.get("subprocess_logs_to_stdout", False))
Review Comment:
Make `subprocess_logs_to_stdout` an explicit argument instead?
--
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]
