You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by sh...@apache.org on 2026/07/20 21:00:46 UTC
(airflow) branch main updated: TriggerRunner: Add batch trigger creation duration metric (#68521)
This is an automated email from the ASF dual-hosted git repository.
shahar1 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 c46a4be2a3c TriggerRunner: Add batch trigger creation duration metric (#68521)
c46a4be2a3c is described below
commit c46a4be2a3cbb0db02324da925bb7f47ce70e0d8
Author: SameerMesiah97 <75...@users.noreply.github.com>
AuthorDate: Mon Jul 20 22:00:39 2026 +0100
TriggerRunner: Add batch trigger creation duration metric (#68521)
Add a timing metric that records the time spent creating all pending triggers during a single create_triggers() invocation. Also, add a unit test verifying that the metric is emitted with the expected tags when triggers are successfully created.
---
airflow-core/newsfragments/68521.feature.rst | 1 +
.../src/airflow/jobs/triggerer_job_runner.py | 13 +++++
airflow-core/tests/unit/jobs/test_triggerer_job.py | 55 +++++++++++++++++++++-
.../observability/metrics/metrics_template.yaml | 6 +++
4 files changed, 74 insertions(+), 1 deletion(-)
diff --git a/airflow-core/newsfragments/68521.feature.rst b/airflow-core/newsfragments/68521.feature.rst
new file mode 100644
index 00000000000..10c76c8b1c4
--- /dev/null
+++ b/airflow-core/newsfragments/68521.feature.rst
@@ -0,0 +1 @@
+Added the ``triggerer.batch_trigger_creation_duration`` metric, which measures the total time spent creating pending triggers during a single trigger runner cycle.
diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
index 50a899c2cf2..759735242ab 100644
--- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
+++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
@@ -1345,6 +1345,12 @@ class TriggerRunner:
async def create_triggers(self):
"""Drain the to_create queue and create all new triggers that have been requested in the DB."""
+ # Emit batch creation duration only when triggers were processed.
+ has_work = bool(self.to_create)
+
+ if has_work:
+ creation_start = time.monotonic()
+
while self.to_create:
await asyncio.sleep(0)
context: Context | None = None
@@ -1419,6 +1425,13 @@ class TriggerRunner:
"events": 0,
}
+ if has_work:
+ stats.timing(
+ "triggerer.batch_trigger_creation_duration",
+ (time.monotonic() - creation_start) * 1000,
+ tags=prune_dict({"team_name": self.team_name}),
+ )
+
async def cancel_triggers(self):
"""
Drain the to_cancel queue and ensure all triggers that are not in the DB are cancelled.
diff --git a/airflow-core/tests/unit/jobs/test_triggerer_job.py b/airflow-core/tests/unit/jobs/test_triggerer_job.py
index a67b6af5254..fa4d68f5fff 100644
--- a/airflow-core/tests/unit/jobs/test_triggerer_job.py
+++ b/airflow-core/tests/unit/jobs/test_triggerer_job.py
@@ -1588,12 +1588,65 @@ class TestTriggerRunner:
):
await runner.create_triggers()
- mock_timing.assert_called_once_with(
+ mock_timing.assert_any_call(
"triggerer.trigger_queue_delay",
1500,
tags=expected_tags,
)
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize(
+ ("team_name", "expected_tags"),
+ [
+ pytest.param("team_a", {"team_name": "team_a"}, id="with_team"),
+ pytest.param(None, {}, id="without_team"),
+ ],
+ )
+ @patch("airflow.jobs.triggerer_job_runner.stats.timing")
+ @patch("airflow.jobs.triggerer_job_runner.Trigger._decrypt_kwargs")
+ @patch(
+ "airflow.jobs.triggerer_job_runner.TriggerRunner.get_trigger_by_classpath",
+ return_value=DateTimeTrigger,
+ )
+ async def test_create_triggers_emits_creation_duration_metric(
+ self,
+ mock_get_trigger_by_classpath,
+ mock_decrypt_kwargs,
+ mock_timing,
+ team_name,
+ expected_tags,
+ ):
+ mock_decrypt_kwargs.return_value = {"moment": timezone.utcnow() + datetime.timedelta(hours=1)}
+
+ workload = workloads.RunTrigger.model_construct(
+ id=1,
+ classpath="abc",
+ encrypted_kwargs="fake",
+ )
+
+ runner = TriggerRunner()
+ runner.team_name = team_name
+ runner.to_create.append(workload)
+
+ await runner.create_triggers()
+
+ mock_timing.assert_called_once()
+
+ metric_name, metric_value = mock_timing.call_args.args
+
+ assert metric_name == "triggerer.batch_trigger_creation_duration"
+
+ # Specific metric_value is not being asserted here as time.monotonic is difficult
+ # to mock deterministically.
+ assert metric_value >= 0
+ assert mock_timing.call_args.kwargs == {"tags": expected_tags}
+
+ task = runner.triggers[workload.id]["task"]
+ task.cancel()
+ await asyncio.gather(task, return_exceptions=True)
+
+ await runner.cleanup_finished_triggers()
+
@pytest.mark.asyncio
@patch("airflow.sdk.execution_time.task_runner.SUPERVISOR_COMMS", create=True)
async def test_sync_state_to_supervisor(self, supervisor_builder):
diff --git a/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml b/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml
index a3e5b6bdd75..c0d43641559 100644
--- a/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml
+++ b/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml
@@ -698,6 +698,12 @@ metrics:
legacy_name: "dagrun.schedule_delay.{dag_id}"
name_variables: ["dag_id"]
+ - name: "triggerer.batch_trigger_creation_duration"
+ description: "Time in milliseconds spent creating a batch of pending triggers."
+ type: "timer"
+ legacy_name: "-"
+ name_variables: []
+
- name: "scheduler.critical_section_duration"
description: "Milliseconds spent in the critical section of scheduler loop"
type: "timer"