(airflow) branch main updated: Add team_name tag to Edge executor sync metric (#69005)

[el...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

eladkal 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 4ee7bd9fcb4 Add team_name tag to Edge executor sync metric (#69005)
4ee7bd9fcb4 is described below

commit 4ee7bd9fcb49c422fa47634d922a415a387749b5
Author: Shivam Rastogi <64...@users.noreply.github.com>
AuthorDate: Tue Jun 30 00:00:52 2026 -0700

    Add team_name tag to Edge executor sync metric (#69005)
    
    * Add team name tag to Edge executor sync metric
    
    * Address Edge executor metric test review
---
 .../providers/edge3/executors/edge_executor.py     |  3 ++-
 .../unit/edge3/executors/test_edge_executor.py     | 24 ++++++++++++++++++++++
 2 files changed, 26 insertions(+), 1 deletion(-)

diff --git a/providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py b/providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py
index 9ac25874c3c..5c582781a45 100644
--- a/providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py
+++ b/providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py
@@ -35,6 +35,7 @@ from airflow.providers.edge3.models.edge_logs import EdgeLogsModel
 from airflow.providers.edge3.models.edge_worker import EdgeWorkerModel, EdgeWorkerState, reset_metrics
 from airflow.providers.edge3.models.types import is_callback_execute
 from airflow.utils.db import DBLocks, create_global_lock
+from airflow.utils.helpers import prune_dict
 from airflow.utils.session import NEW_SESSION, provide_session
 from airflow.utils.state import TaskInstanceState
 
@@ -340,7 +341,7 @@ class EdgeExecutor(BaseExecutor):
     @provide_session
     def sync(self, *, session: Session = NEW_SESSION) -> None:
         """Sync will get called periodically by the heartbeat method."""
-        with Stats.timer("edge_executor.sync.duration"):
+        with Stats.timer("edge_executor.sync.duration", tags=prune_dict({"team_name": self.team_name})):
             orphaned = self._update_orphaned_jobs(session)
             purged = self._purge_jobs(session)
             liveness = self._check_worker_liveness(session)
diff --git a/providers/edge3/tests/unit/edge3/executors/test_edge_executor.py b/providers/edge3/tests/unit/edge3/executors/test_edge_executor.py
index 3554689b794..8975eb7ad94 100644
--- a/providers/edge3/tests/unit/edge3/executors/test_edge_executor.py
+++ b/providers/edge3/tests/unit/edge3/executors/test_edge_executor.py
@@ -120,6 +120,30 @@ class TestEdgeExecutor:
             assert jobs[0].task_id == "started_running_orphaned"
             assert jobs[0].state == TaskInstanceState.REMOVED
 
+    @pytest.mark.skipif(not AIRFLOW_V_3_2_PLUS, reason="team_name is only available in Airflow 3.2+")
+    @pytest.mark.parametrize(
+        ("team_name", "expected_tags"),
+        [
+            ("team_a", {"team_name": "team_a"}),
+            (None, {}),
+        ],
+    )
+    @patch(f"{Stats.__module__}.Stats.timer")
+    def test_sync_duration_metric_tags_team_name(self, mock_stats_timer, team_name, expected_tags):
+        executor = EdgeExecutor(team_name=team_name)
+
+        with (
+            mock.patch.object(executor, "_update_orphaned_jobs", return_value=False),
+            mock.patch.object(executor, "_purge_jobs", return_value=False),
+            mock.patch.object(executor, "_check_worker_liveness", return_value=False),
+        ):
+            executor.sync(session=MagicMock())
+
+        mock_stats_timer.assert_called_once_with(
+            "edge_executor.sync.duration",
+            tags=expected_tags,
+        )
+
     @patch("airflow.providers.edge3.executors.edge_executor.EdgeExecutor.running_state")
     @patch("airflow.providers.edge3.executors.edge_executor.EdgeExecutor.success")
     @patch("airflow.providers.edge3.executors.edge_executor.EdgeExecutor.fail")