From 39312be1988223141944166f418e24835e596a79 Mon Sep 17 00:00:00 2001
From: Ash Berlin-Taylor <ash@apache.org>
Date: Mon, 20 Apr 2026 15:33:53 +0100
Subject: [PATCH] [v3-2-test] Respect dag processor config option to show
 parsing logs on stdout (#65528)

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.
(cherry picked from commit 1193073f9083ad8e38facbc07eff0476928008a7)

Co-authored-by: Ash Berlin-Taylor <ash@apache.org>
---
 airflow-core/src/airflow/configuration.py     |  1 +
 .../src/airflow/dag_processing/manager.py     |  2 +
 .../src/airflow/dag_processing/processor.py   | 28 ++++++++-
 .../tests/unit/dag_processing/test_manager.py | 32 ++++++++++
 .../unit/dag_processing/test_processor.py     | 59 +++++++++++++++++++
 .../airflow/sdk/execution_time/supervisor.py  | 12 ++--
 6 files changed, 129 insertions(+), 5 deletions(-)

diff --git a/airflow-core/src/airflow/configuration.py b/airflow-core/src/airflow/configuration.py
index be07df7f418c4..7790a252af37a 100644
--- a/airflow-core/src/airflow/configuration.py
+++ b/airflow-core/src/airflow/configuration.py
@@ -279,6 +279,7 @@ def _update_logging_deprecated_template_to_one_from_defaults(self):
         ("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 2e40230131bde..57ebd56c0e9eb 100644
--- a/airflow-core/src/airflow/dag_processing/manager.py
+++ b/airflow-core/src/airflow/dag_processing/manager.py
@@ -1034,10 +1034,12 @@ def _create_process(self, dag_file: DagFileInfo) -> DagFileProcessorProcess:
             path=dag_file.absolute_path,
             bundle_path=cast("Path", dag_file.bundle_path),
             bundle_name=dag_file.bundle_name,
+            dag_file_rel_path=str(dag_file.rel_path),
             callbacks=callback_to_execute_for_file,
             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 758472598d28c..6cbef2fef67bb 100644
--- a/airflow-core/src/airflow/dag_processing/processor.py
+++ b/airflow-core/src/airflow/dag_processing/processor.py
@@ -18,6 +18,7 @@
 
 import contextlib
 import importlib
+import logging
 import os
 import traceback
 from collections.abc import Callable, Sequence
@@ -74,6 +75,8 @@
 from airflow.utils.state import TaskInstanceState
 
 if TYPE_CHECKING:
+    from socket import socket
+
     from structlog.typing import FilteringBoundLogger
 
     from airflow.api_fastapi.execution_api.app import InProcessExecutionAPI
@@ -513,6 +516,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,
@@ -520,6 +526,7 @@ def start(  # type: ignore[override]
         path: str | os.PathLike[str],
         bundle_path: Path,
         bundle_name: str,
+        dag_file_rel_path: str,
         callbacks: list[CallbackRequest],
         target: Callable[[], None] = _parse_file_entrypoint,
         client: Client,
@@ -529,7 +536,13 @@ def start(  # type: ignore[override]
 
         _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=dag_file_rel_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
@@ -549,6 +562,19 @@ def _on_child_started(
         )
         self.send_msg(msg, request_id=0)
 
+    def _get_target_loggers(self) -> tuple[FilteringBoundLogger, ...]:
+        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: tuple[FilteringBoundLogger, ...], name: str, log_level: int = logging.INFO
+    ) -> Callable[[socket], bool]:
+        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 3f866dee14d4e..15899a3c7ad1e 100644
--- a/airflow-core/tests/unit/dag_processing/test_manager.py
+++ b/airflow-core/tests/unit/dag_processing/test_manager.py
@@ -184,6 +184,8 @@ def mock_processor(self, start_time: float | None = None) -> tuple[DagFileProces
             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
@@ -796,6 +798,32 @@ def test_cleanup_stale_bundle_versions(self, mock_bundle_manager):
         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, 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
@@ -1384,10 +1412,12 @@ def test_callback_queue(self, mock_get_logger, configure_testing_dag_bundle):
                     path=Path(dag2_path.bundle_path, dag2_path.rel_path),
                     bundle_path=dag2_path.bundle_path,
                     bundle_name="testing",
+                    dag_file_rel_path=str(dag2_path.rel_path),
                     callbacks=[dag2_req1],
                     selector=mock.ANY,
                     logger=mock_logger,
                     logger_filehandle=mock_filehandle,
+                    subprocess_logs_to_stdout=False,
                     client=mock.ANY,
                 ),
                 mock.call(
@@ -1395,10 +1425,12 @@ def test_callback_queue(self, mock_get_logger, configure_testing_dag_bundle):
                     path=Path(dag1_path.bundle_path, dag1_path.rel_path),
                     bundle_path=dag1_path.bundle_path,
                     bundle_name="testing",
+                    dag_file_rel_path=str(dag1_path.rel_path),
                     callbacks=[dag1_req1, dag1_req2],
                     selector=mock.ANY,
                     logger=mock_logger,
                     logger_filehandle=mock_filehandle,
+                    subprocess_logs_to_stdout=False,
                     client=mock.ANY,
                 ),
             ]
diff --git a/airflow-core/tests/unit/dag_processing/test_processor.py b/airflow-core/tests/unit/dag_processing/test_processor.py
index c46aceda7949b..667037f9a1e0a 100644
--- a/airflow-core/tests/unit/dag_processing/test_processor.py
+++ b/airflow-core/tests/unit/dag_processing/test_processor.py
@@ -18,6 +18,7 @@
 from __future__ import annotations
 
 import inspect
+import logging
 import pathlib
 import sys
 import textwrap
@@ -164,6 +165,7 @@ def dag_in_a_fn():
             path=path,
             bundle_path=tmp_path,
             bundle_name="testing",
+            dag_file_rel_path=str(path.relative_to(tmp_path)),
             callbacks=[],
             logger=logger,
             logger_filehandle=logger_filehandle,
@@ -200,6 +202,7 @@ def dag_in_a_fn():
             path=path,
             bundle_path=tmp_path,
             bundle_name="testing",
+            dag_file_rel_path=str(path.relative_to(tmp_path)),
             callbacks=[],
             logger=logger,
             logger_filehandle=logger_filehandle,
@@ -234,6 +237,7 @@ def dag_in_a_fn():
             path=path,
             bundle_path=tmp_path,
             bundle_name="testing",
+            dag_file_rel_path=str(path.relative_to(tmp_path)),
             callbacks=[],
             logger=logger,
             logger_filehandle=logger_filehandle,
@@ -278,6 +282,7 @@ def dag_in_a_fn():
             path=path,
             bundle_path=tmp_path,
             bundle_name="testing",
+            dag_file_rel_path=str(path.relative_to(tmp_path)),
             callbacks=[],
             logger=logger,
             logger_filehandle=logger_filehandle,
@@ -316,6 +321,7 @@ def dag_in_a_fn():
             path=path,
             bundle_path=tmp_path,
             bundle_name="testing",
+            dag_file_rel_path=str(path.relative_to(tmp_path)),
             callbacks=[],
             logger=logger,
             logger_filehandle=logger_filehandle,
@@ -346,6 +352,7 @@ def dag_in_a_fn():
             path=path,
             bundle_path=tmp_path,
             bundle_name="testing",
+            dag_file_rel_path=str(path.relative_to(tmp_path)),
             callbacks=[],
             logger=logger,
             logger_filehandle=logger_filehandle,
@@ -380,6 +387,7 @@ def test_import_module_in_bundle_root(self, tmp_path: pathlib.Path, inprocess_cl
             path=dag1_path,
             bundle_path=tmp_path,
             bundle_name="testing",
+            dag_file_rel_path=str(dag1_path.relative_to(tmp_path)),
             callbacks=[],
             logger=MagicMock(spec=FilteringBoundLogger),
             logger_filehandle=MagicMock(spec=BinaryIO),
@@ -1967,3 +1975,54 @@ def get_type_names(union_type):
             + "\n".join(f"  - {t}" for t in sorted(task_diff))
             + "\n\nEither handle these types in ToDagProcessor or update in_task_runner_but_not_in_dag_processing_process list."
         )
+
+
+class TestDagFileProcessorProcess:
+    @pytest.fixture
+    def proc(self):
+        from socket import socketpair
+        from unittest.mock import MagicMock
+
+        proc_mock = MagicMock()
+        proc_mock.create_time.return_value = 0.0
+        r, w = socketpair()
+        instance = DagFileProcessorProcess(
+            process_log=structlog.get_logger().bind(),
+            id=uuid.uuid4(),
+            pid=1234,
+            process=proc_mock,
+            stdin=w,
+            logger_filehandle=MagicMock(spec=BinaryIO),
+            client=MagicMock(spec=Client),
+            bundle_name="mybundle",
+            dag_file_rel_path="dags/my_dag.py",
+        )
+        instance._open_sockets.clear()
+        r.close()
+        w.close()
+        return instance
+
+    def test_get_target_loggers_file_mode_no_context_added(self, proc):
+        proc.subprocess_logs_to_stdout = False
+        loggers = proc._get_target_loggers()
+        assert len(loggers) == 1
+        with structlog.testing.capture_logs() as cap:
+            loggers[0].info("test")
+        assert "dag_file" not in cap[0]
+        assert "bundle_name" not in cap[0]
+
+    def test_get_target_loggers_stdout_mode_binds_dag_file_context(self, proc):
+        proc.subprocess_logs_to_stdout = True
+        loggers = proc._get_target_loggers()
+        with structlog.testing.capture_logs() as cap:
+            for bound_logger in loggers:
+                bound_logger.info("test")
+        assert all(e.get("dag_file") == "dags/my_dag.py" for e in cap)
+        assert all(e.get("bundle_name") == "mybundle" for e in cap)
+
+    def test_create_log_forwarder_rewrites_task_prefix_to_dag_processor(self, proc):
+        from airflow.sdk.execution_time.supervisor import WatchedSubprocess
+
+        with patch.object(WatchedSubprocess, "_create_log_forwarder") as mock_base:
+            proc._create_log_forwarder((), "task.stdout")
+        mock_base.assert_called_once_with((), "dag_processor.stdout", logging.INFO)
diff --git a/task-sdk/src/airflow/sdk/execution_time/supervisor.py b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
index 0eee4889c2572..a4484199f6297 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 @@ def _register_pipe_readers(self, stdout: socket, stderr: socket, requests: socke
             )
         )
 
-        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 @@ def _register_pipe_readers(self, stdout: socket, stderr: socket, requests: socke
             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(
