This is an automated email from the ASF dual-hosted git repository.
ashb pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 1193073f908 Respect dag processor config option to show parsing logs
on stdout (#65528)
1193073f908 is described below
commit 1193073f9083ad8e38facbc07eff0476928008a7
Author: Ash Berlin-Taylor <[email protected]>
AuthorDate: Mon Apr 20 15:33:53 2026 +0100
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.
---
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 5fc34140524..995dd353165 100644
--- a/airflow-core/src/airflow/dag_processing/manager.py
+++ b/airflow-core/src/airflow/dag_processing/manager.py
@@ -1168,10 +1168,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 6f8c293fe5e..148d978084d 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
@@ -531,6 +534,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,
@@ -538,6 +544,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,
@@ -547,7 +554,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
@@ -567,6 +580,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 8f2cb9b8cb4..17845e958d3 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, 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
@@ -1489,10 +1517,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(
@@ -1500,10 +1530,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 15842c1ce9e..bf07764a83a 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),
@@ -2000,3 +2008,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 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(
