This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new 97841c6035a [v3-3-test] Trace the dag-processor job instead of its
parent process with memray (#72367) (#72661)
97841c6035a is described below
commit 97841c6035a65be6941702768a86ce1400712dcc
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Tue Sep 8 14:46:10 2026 +0530
[v3-3-test] Trace the dag-processor job instead of its parent process with
memray (#72367) (#72661)
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.
(cherry picked from commit 4c32caa92bae27c854f32df2779ece82374f9fbb)
Co-authored-by: Y-C <[email protected]>
Co-authored-by: Eason09053360
<[email protected]>
Co-authored-by: Rahul Vats <[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 04e410a5d66..6058e78febc 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 6224be2c5d5..84fd2e25bd7 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
@@ -67,3 +67,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()
