This is an automated email from the ASF dual-hosted git repository.
potiuk 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 4c32caa92ba Trace the dag-processor job instead of its parent process
with memray (#72367)
4c32caa92ba is described below
commit 4c32caa92bae27c854f32df2779ece82374f9fbb
Author: Y-C <[email protected]>
AuthorDate: Tue Sep 8 02:22:40 2026 +0800
Trace the dag-processor job instead of its parent process with memray
(#72367)
The tracer was attached to the CLI entry point, which forks before the job
runs and does not follow the child, so anyone who opted into profiling got a
capture of interpreter startup rather than of the work they enabled it to
investigate. Dag-processor memory growth is exactly what this feature exists
to diagnose, and it was the one component of the four that reported on the
wrong process.
Co-authored-by: Eason09053360
<[email protected]>
---
.../src/airflow/cli/commands/dag_processor_command.py | 8 ++++++--
.../unit/cli/commands/test_dag_processor_command.py | 17 +++++++++++++++++
2 files changed, 23 insertions(+), 2 deletions(-)
diff --git a/airflow-core/src/airflow/cli/commands/dag_processor_command.py
b/airflow-core/src/airflow/cli/commands/dag_processor_command.py
index 7e7df45a8d6..104065d6d78 100644
--- a/airflow-core/src/airflow/cli/commands/dag_processor_command.py
+++ b/airflow-core/src/airflow/cli/commands/dag_processor_command.py
@@ -47,6 +47,10 @@ def _create_dag_processor_job_runner(args: Any) ->
DagProcessorJobRunner:
@enable_memray_trace(component=MemrayTraceComponents.dag_processor)
+def _run_dag_processor_job(job_runner: DagProcessorJobRunner) -> None:
+ run_job(job=job_runner.job, execute_callable=job_runner._execute)
+
+
@cli_utils.action_cli
@providers_configuration_loaded
def dag_processor(args):
@@ -58,7 +62,7 @@ def dag_processor(args):
from airflow.cli.hot_reload import run_with_reloader
run_with_reloader(
- lambda: run_job(job=job_runner.job,
execute_callable=job_runner._execute),
+ lambda: _run_dag_processor_job(job_runner),
process_name="dag-processor",
)
return
@@ -66,6 +70,6 @@ def dag_processor(args):
run_command_with_daemon_option(
args=args,
process_name="dag-processor",
- callback=lambda: run_job(job=job_runner.job,
execute_callable=job_runner._execute),
+ callback=lambda: _run_dag_processor_job(job_runner),
should_setup_logging=True,
)
diff --git a/airflow-core/tests/unit/cli/commands/test_dag_processor_command.py
b/airflow-core/tests/unit/cli/commands/test_dag_processor_command.py
index 6b6e429b600..1631001b594 100644
--- a/airflow-core/tests/unit/cli/commands/test_dag_processor_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_dag_processor_command.py
@@ -91,3 +91,20 @@ class TestDagProcessorCommand:
mock_reloader.assert_called_once()
# The callback function should be callable
assert callable(mock_reloader.call_args[0][0])
+
+ @conf_vars(
+ {("core", "load_examples"): "False", ("profiling",
"memray_trace_components"): "dag_processor"}
+ )
+
@mock.patch("airflow.cli.commands.dag_processor_command.run_command_with_daemon_option")
+
@mock.patch("airflow.cli.commands.dag_processor_command.DagProcessorJobRunner")
+ def test_memray_traces_the_job_and_not_the_parent_process(self,
mock_runner, mock_daemon_option):
+ """The callback runs on the far side of the daemon fork, so the tracer
has to start there."""
+ mock_runner.return_value.job_type = "DagProcessorJob"
+ memray = mock.MagicMock()
+
+ with mock.patch.dict("sys.modules", {"memray": memray}):
+
dag_processor_command.dag_processor(self.parser.parse_args(["dag-processor"]))
+ memray.Tracker.assert_not_called()
+
+ mock_daemon_option.call_args.kwargs["callback"]()
+ memray.Tracker.assert_called_once()
