This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-2-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 465e8562ae40f674320a664e9fe72a0d26c62e38
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Mon Apr 20 18:05:33 2026 +0100
[v3-2-test] Respect dag processor config option to show parsing logs on
stdout (#65528) (#65541)
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 <[email protected]>
---
airflow-core/src/airflow/configuration.py | 1 +
airflow-core/src/airflow/dag_processing/manager.py | 2 +
.../src/airflow/dag_processing/processor.py | 28 +++++++++-
.../tests/unit/dag_processing/test_manager.py | 32 ++++++++++++
.../tests/unit/dag_processing/test_processor.py | 59 ++++++++++++++++++++++
.../src/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 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 2e40230131b..57ebd56c0e9 100644
--- a/airflow-core/src/airflow/dag_processing/manager.py
+++ b/airflow-core/src/airflow/dag_processing/manager.py
@@ -1034,10 +1034,12 @@ class DagFileProcessorManager(LoggingMixin):
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 758472598d2..6cbef2fef67 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
@@ -74,6 +75,8 @@ from airflow.utils.file import iter_airflow_imports
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 @@ class DagFileProcessorProcess(WatchedSubprocess):
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 @@ 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=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 @@ class DagFileProcessorProcess(WatchedSubprocess):
)
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 3f866dee14d..15899a3c7ad 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 @@ 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
@@ -796,6 +798,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, 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 @@ class TestDagFileProcessorManager:
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 @@ class TestDagFileProcessorManager:
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 c46aceda794..667037f9a1e 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 @@ class TestDagFileProcessor:
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 @@ class TestDagFileProcessor:
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 @@ class TestDagFileProcessor:
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 @@ class TestDagFileProcessor:
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 @@ class TestDagFileProcessor:
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 @@ class TestDagFileProcessor:
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 @@ class TestDagFileProcessor:
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 @@ class TestDagProcessingMessageTypes:
+ "\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 0eee4889c25..a4484199f62 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(
