This is an automated email from the ASF dual-hosted git repository. ashb pushed a commit to branch worktree-dag-proc-stdout-again in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 42c02e6b41236a5a13fe5d95b7d77a3571773d19 Author: Ash Berlin-Taylor <[email protected]> AuthorDate: Mon Apr 20 11:42:17 2026 +0100 Respect dag processor config option to show parsing logs on stdout This is very useful when running things in a container, where it's much easier to get container logs than files on disk. This config option was added in 2.4.0 but got broken around 3.0 or 3.1. Most of capabilities existed already, we just needed to pass the right value down. Example log line: ``` 2026-04-20T10:44:13.977892Z [info ] Ptint from dag [dag_processor.stdout] bundle_name=main dag_file=dags/example_dag_advanced.py ``` Since we've got structured logging now it includes the bundle name and rel path of the dag file --- airflow-core/src/airflow/configuration.py | 1 + airflow-core/src/airflow/dag_processing/manager.py | 1 + .../src/airflow/dag_processing/processor.py | 23 ++++++++++++++++- .../tests/unit/dag_processing/test_manager.py | 30 ++++++++++++++++++++++ .../src/airflow/sdk/execution_time/supervisor.py | 12 ++++++--- 5 files changed, 62 insertions(+), 5 deletions(-) diff --git a/airflow-core/src/airflow/configuration.py b/airflow-core/src/airflow/configuration.py index be07df7f418..7790a252af3 100644 --- a/airflow-core/src/airflow/configuration.py +++ b/airflow-core/src/airflow/configuration.py @@ -279,6 +279,7 @@ class AirflowConfigParser(_SharedAirflowConfigParser): ("logging", "gunicorn_logging_level"): _available_logging_levels, ("webserver", "analytical_tool"): ["google_analytics", "metarouter", "segment", "matomo", ""], ("api", "grid_view_sorting_order"): ["topological", "hierarchical_alphabetical"], + ("logging", "dag_processor_log_target"): ["file", "stdout"], } upgraded_values: dict[tuple[str, str], str] diff --git a/airflow-core/src/airflow/dag_processing/manager.py b/airflow-core/src/airflow/dag_processing/manager.py index 5fc34140524..e47b01cb9f0 100644 --- a/airflow-core/src/airflow/dag_processing/manager.py +++ b/airflow-core/src/airflow/dag_processing/manager.py @@ -1172,6 +1172,7 @@ class DagFileProcessorManager(LoggingMixin): selector=self.selector, logger=logger, logger_filehandle=logger_filehandle, + subprocess_logs_to_stdout=conf.get("logging", "dag_processor_log_target") == "stdout", client=self.client, ) diff --git a/airflow-core/src/airflow/dag_processing/processor.py b/airflow-core/src/airflow/dag_processing/processor.py index 6f8c293fe5e..4ea2c33c1da 100644 --- a/airflow-core/src/airflow/dag_processing/processor.py +++ b/airflow-core/src/airflow/dag_processing/processor.py @@ -18,6 +18,7 @@ from __future__ import annotations import contextlib import importlib +import logging import os import traceback from collections.abc import Callable, Sequence @@ -531,6 +532,9 @@ class DagFileProcessorProcess(WatchedSubprocess): client: Client """The HTTP client to use for communication with the API server.""" + bundle_name: str + dag_file_rel_path: str + @classmethod def start( # type: ignore[override] cls, @@ -547,7 +551,13 @@ class DagFileProcessorProcess(WatchedSubprocess): _pre_import_airflow_modules(os.fspath(path), logger) - proc: Self = super().start(target=target, client=client, **kwargs) + proc: Self = super().start( + target=target, + client=client, + bundle_name=bundle_name, + dag_file_rel_path=str(Path(path).relative_to(bundle_path)), + **kwargs, + ) proc.had_callbacks = bool(callbacks) # Track if this process had callbacks proc._on_child_started(callbacks, path, bundle_path, bundle_name) return proc @@ -567,6 +577,17 @@ class DagFileProcessorProcess(WatchedSubprocess): ) self.send_msg(msg, request_id=0) + def _get_target_loggers(self): + base = super()._get_target_loggers() + if not self.subprocess_logs_to_stdout: + return base + return tuple( + logger.bind(dag_file=self.dag_file_rel_path, bundle_name=self.bundle_name) for logger in base + ) + + def _create_log_forwarder(self, loggers, name, log_level=logging.INFO): + return super()._create_log_forwarder(loggers, name.replace("task.", "dag_processor.", 1), log_level) + def _handle_request(self, msg: ToManager, log: FilteringBoundLogger, req_id: int) -> None: from airflow.sdk.api.datamodels._generated import ( ConnectionResponse, diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py b/airflow-core/tests/unit/dag_processing/test_manager.py index 8f2cb9b8cb4..8b4aed7d010 100644 --- a/airflow-core/tests/unit/dag_processing/test_manager.py +++ b/airflow-core/tests/unit/dag_processing/test_manager.py @@ -187,6 +187,8 @@ class TestDagFileProcessorManager: stdin=write_end, logger_filehandle=logger_filehandle, client=MagicMock(), + bundle_name="testing", + dag_file_rel_path="test_dag.py", ) if start_time: ret.start_time = start_time @@ -799,6 +801,32 @@ class TestDagFileProcessorManager: manager.cleanup_stale_bundle_versions() mock_bundle_manager.return_value.remove_stale_bundle_versions.assert_called_once_with() + @pytest.mark.parametrize( + ("log_target", "expected_subprocess_logs_to_stdout"), + [ + ("stdout", True), + ("file", False), + ], + ) + @mock.patch.object(DagFileProcessorManager, "_get_logger_for_dag_file") + def test_create_process_subprocess_logs_to_stdout( + self, mock_get_logger, configure_testing_dag_bundle, log_target, expected_subprocess_logs_to_stdout + ): + mock_logger = MagicMock() + mock_filehandle = MagicMock() + mock_get_logger.return_value = [mock_logger, mock_filehandle] + + with conf_vars({("logging", "dag_processor_log_target"): log_target}): + manager = DagFileProcessorManager(max_runs=1, processor_timeout=60) + dag_file = DagFileInfo( + bundle_name="testing", rel_path=Path("my_dag.py"), bundle_path=Path("/tmp") + ) + with mock.patch.object(DagFileProcessorProcess, "start") as mock_start: + manager._create_process(dag_file) + + _, kwargs = mock_start.call_args + assert kwargs["subprocess_logs_to_stdout"] is expected_subprocess_logs_to_stdout + def test_kill_timed_out_processors_kill(self): manager = DagFileProcessorManager(max_runs=1, processor_timeout=5) # Set start_time to ensure timeout occurs: start_time = current_time - (timeout + 1) = always (timeout + 1) seconds @@ -1493,6 +1521,7 @@ class TestDagFileProcessorManager: selector=mock.ANY, logger=mock_logger, logger_filehandle=mock_filehandle, + subprocess_logs_to_stdout=False, client=mock.ANY, ), mock.call( @@ -1504,6 +1533,7 @@ class TestDagFileProcessorManager: selector=mock.ANY, logger=mock_logger, logger_filehandle=mock_filehandle, + subprocess_logs_to_stdout=False, client=mock.ANY, ), ] diff --git a/task-sdk/src/airflow/sdk/execution_time/supervisor.py b/task-sdk/src/airflow/sdk/execution_time/supervisor.py index ba7e2091f25..39f505a0109 100644 --- a/task-sdk/src/airflow/sdk/execution_time/supervisor.py +++ b/task-sdk/src/airflow/sdk/execution_time/supervisor.py @@ -566,10 +566,7 @@ class WatchedSubprocess: ) ) - target_loggers: tuple[FilteringBoundLogger, ...] = (self.process_log,) - - if self.subprocess_logs_to_stdout: - target_loggers += (log,) + target_loggers = self._get_target_loggers() self.selector.register( stdout, selectors.EVENT_READ, self._create_log_forwarder(target_loggers, "task.stdout") @@ -592,6 +589,13 @@ class WatchedSubprocess: length_prefixed_frame_reader(self.handle_requests(log), on_close=self._on_socket_closed), ) + 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 + def _create_log_forwarder(self, loggers, name, log_level=logging.INFO) -> Callable[[socket], bool]: """Create a socket handler that forwards logs to a logger.""" loggers = tuple(
