Optionally emit Dag tags as metric tags

airflow-core/src/airflow/config_templates/config.yml

@@ -1280,6 +1280,17 @@ metrics:
       type: boolean
       example: ~
       default: "False"
+    dag_tags_in_metrics:
+      description: |
+        Set to ``True`` to include Dag tags as metric tags on all Dag-run and task-instance metrics.
+        Tags that contain a colon (e.g. ``env:prod``) are split into a key/value pair. Plain tags
+        (e.g. ``production``) are emitted as standalone DogStatsd tags, or as ``tag=true`` in InfluxDB
+        line-protocol format. Disabled by default to avoid unexpected cardinality increases, since Dag
+        tags are free-form, user-defined strings.
+      version_added: 3.4.0
+      type: boolean
+      example: ~
+      default: "False"
     otel_on:
       description: |
         Enables sending metrics to OpenTelemetry.

airflow-core/src/airflow/jobs/scheduler_job_runner.py

@@ -334,6 +334,7 @@ def __init__(
         self._scheduler_use_job_schedule = conf.getboolean("scheduler", "use_job_schedule", fallback=True)
         self._parallelism = conf.getint("core", "parallelism")
         self._multi_team = conf.getboolean("core", "multi_team")
+        self._dag_tags_in_metrics = conf.getboolean("metrics", "dag_tags_in_metrics", fallback=False)
         self._max_partition_dag_runs_per_loop = MAX_PARTITION_DAG_RUNS_PER_LOOP
         self._dag_id_to_team_name: dict[str, str | None] = {}
 
@@ -736,10 +737,10 @@ def _executable_task_instances_to_queued(self, max_tis: int, session: Session) -
                     list(unique_dag_ids),
                 )
                 for ti in task_instances_to_examine:
-                    # Set team as a transient attribute; team lives on the Bundle, not
-                    # on the TI/DagRun schema, so we resolve it at scheduling time.
+                    # Team lives on the Bundle, not the TI/DagRun schema, so resolve it at scheduling
+                    # time and stash it on the dag run, where stats_tags reads it for metric tagging.
                     if team := dag_id_to_team_name.get(ti.dag_id):
-                        ti._team_name = team
+                        ti.dag_run._team_name = team
 
             executor_slots_available: dict[ExecutorName, int] = {}
             # First get a mapping of executor names to slots they have available
@@ -1260,6 +1261,7 @@ def _process_executor_events(self, executor: BaseExecutor, session: Session) ->
                 job_id=self.job.id,
                 scheduler_dag_bag=self.scheduler_dag_bag,
                 session=session,
+                eagerly_load_dag_tags=self._dag_tags_in_metrics,
             )
         except Exception as exc:
             stats.incr("scheduler.executor_events.failed", tags={"exception_class": type(exc).__name__})
@@ -1272,7 +1274,12 @@ def _emit_executor_events_batch_metrics(num_events: int) -> None:
 
     @classmethod
     def process_executor_events(
-        cls, executor: BaseExecutor, job_id: int | None, scheduler_dag_bag: DBDagBag, session: Session
+        cls,
+        executor: BaseExecutor,
+        job_id: int | None,
+        scheduler_dag_bag: DBDagBag,
+        session: Session,
+        eagerly_load_dag_tags: bool = False,
     ) -> int:
         """
         Process task completion events from the executor and update task instance states.
@@ -1295,6 +1302,9 @@ def process_executor_events(
         :param job_id: The scheduler job ID, used to detect task requeuing by other schedulers
         :param scheduler_dag_bag: Serialized DAG bag for retrieving task definitions
         :param session: Database session for task instance updates
+        :param eagerly_load_dag_tags: When True, eager-load dag_model.tags so the per-finished-task
+            metrics carry Dag tags without a per-TI lazy load. The scheduler passes its cached flag so
+            the hot path never reads conf; other callers (e.g. ``dag.test()``) leave it at the default.
 
         :return: Number of events processed from the executor event buffer
 
@@ -1386,6 +1396,12 @@ def process_executor_events(
             .options(joinedload(TI.dag_run).selectinload(DagRun.created_dag_version))
             .options(joinedload(TI.dag_version))
         )
+        # When emitting Dag tags as metric tags, eager-load dag_model.tags so the per-finished-task
+        # ti_failures / operator_failures / task.*_duration metrics carry them without a per-TI lazy load.
+        # TI already joins DagModel by dag_id, so warm tags off that relationship directly rather than
+        # via the dag_run hop; the DagModel is shared in the identity map, so dag_run.dag_model.tags is free.
+        if eagerly_load_dag_tags:
+            query = query.options(selectinload(TI.dag_model).selectinload(DagModel.tags))
         # row lock this entire set of taskinstances to make sure the scheduler doesn't fail when we have
         # multi-schedulers
         locked_query = with_row_locks(query, of=TI, session=session, skip_locked=True)
@@ -1922,7 +1938,11 @@ def _do_scheduling(self, session: Session) -> int:
             # examining, rather than making one query per DagRun.
             # Materialize into a list because the multi-team block below iterates
             # the result and ScalarResult is a one-pass iterator.
-            dag_runs = list(DagRun.get_running_dag_runs_to_examine(session=session))
+            dag_runs = list(
+                DagRun.get_running_dag_runs_to_examine(
+                    session=session, eagerly_load_dag_tags=self._dag_tags_in_metrics
+                )
+            )
 
             if self._multi_team and dag_runs:
                 unique_dag_ids = {dr.dag_id for dr in dag_runs}
@@ -3553,9 +3573,6 @@ def _purge_task_instances_without_heartbeats(
         if self._multi_team:
             unique_dag_ids = {ti.dag_id for ti in task_instances_without_heartbeats}
             dag_id_to_team_name = self._get_team_names_for_dag_ids(unique_dag_ids, session)
-            for ti in task_instances_without_heartbeats:
-                if team := dag_id_to_team_name.get(ti.dag_id):
-                    ti._team_name = team
         else:
             dag_id_to_team_name = {}
 

airflow-core/src/airflow/models/dagrun.py

@@ -53,16 +53,25 @@
     update,
 )
 from sqlalchemy.dialects import postgresql
-from sqlalchemy.exc import IntegrityError
+from sqlalchemy.exc import IntegrityError, SQLAlchemyError
 from sqlalchemy.ext.associationproxy import association_proxy
 from sqlalchemy.ext.hybrid import hybrid_property
 from sqlalchemy.ext.mutable import MutableDict
-from sqlalchemy.orm import Mapped, declared_attr, joinedload, mapped_column, relationship, synonym, validates
+from sqlalchemy.orm import (
+    Mapped,
+    declared_attr,
+    joinedload,
+    mapped_column,
+    relationship,
+    synonym,
+    validates,
+)
 from sqlalchemy.orm.exc import StaleDataError
 from sqlalchemy.sql.expression import false, select
 from sqlalchemy.sql.functions import coalesce
 
 from airflow._shared.observability.metrics import stats
+from airflow._shared.observability.metrics.stats import build_dag_metric_tags
 from airflow._shared.observability.traces import (
     DAGRUN_PARENT_TRACE_CONTEXT_KEY,
     TASK_SPAN_DETAIL_LEVEL_KEY,
@@ -576,11 +585,36 @@ def check_version_id_exists_in_dr(self, dag_version_id: UUID, *, session: Sessio
         )
         return session.scalar(select_stmt)
 
+    def dag_tags_for_stats(self) -> dict[str, str]:
+        """Convert dag tags to metric tags. Tags with ':' become key:value; others are standalone (empty value)."""
+        if not airflow_conf.getboolean("metrics", "dag_tags_in_metrics", fallback=False):
+            return {}
+        try:
+            # Lazy-loads dag_model.tags if not already loaded. The scheduler hot loop
+            # (get_running_dag_runs_to_examine) eager-loads these to avoid a per-DagRun query; other,
+            # low-frequency emission paths fall back to this lazy load. On a detached/expired DagRun
+            # the load raises — swallow it so metric tagging never breaks the caller.
+            if not self.dag_model or not self.dag_model.tags:
+                return {}
+            return build_dag_metric_tags(tag.name for tag in self.dag_model.tags)
+        except SQLAlchemyError:
+            return {}
+
     @property
     def stats_tags(self) -> dict[str, str]:
-        return prune_dict(
-            {"dag_id": self.dag_id, "run_type": self.run_type, "team_name": getattr(self, "_team_name", None)}
+        # prune_dict strips falsy values, so merge dag tags after it runs so standalone
+        # tags (empty value) are preserved for DogStatsD emission.
+        base = prune_dict(
+            {
+                "dag_id": self.dag_id,
+                # bare value so it serializes as e.g. "scheduled", not "dagruntype.scheduled"
+                "run_type": getattr(self.run_type, "value", self.run_type),
+                "team_name": getattr(self, "_team_name", None),
+            }
         )
+        dag_tags = self.dag_tags_for_stats()
+        # Built-in keys win on collision; dag tags fill in everything else.
+        return {**dag_tags, **base}
 
     def get_state(self):
         return self._state
@@ -702,7 +736,9 @@ def active_runs_of_dags(
 
     @classmethod
     @retry_db_transaction
-    def get_running_dag_runs_to_examine(cls, session: Session) -> ScalarResult[DagRun]:
+    def get_running_dag_runs_to_examine(
+        cls, *, session: Session, eagerly_load_dag_tags: bool
+    ) -> ScalarResult[DagRun]:
         """
         Return the next DagRuns that the scheduler should attempt to schedule.
 
@@ -733,6 +769,12 @@ def get_running_dag_runs_to_examine(cls, session: Session) -> ScalarResult[DagRu
             .limit(cls.DEFAULT_DAGRUNS_TO_EXAMINE)
         )
 
+        # When dag tags are emitted as metric tags, eagerly load dag_model.tags so stats_tags does not
+        # fire a per-DagRun N+1 lazy load in the scheduler loop. The caller owns the feature decision;
+        # the scheduler passes its cached flag so the loop never reads conf.
+        if eagerly_load_dag_tags:
+            query = query.options(joinedload(cls.dag_model).selectinload(DagModel.tags))
+
         query = query.where(DagRun.run_after <= func.now())
 
         result = session.scalars(with_row_locks(query, of=cls, session=session, skip_locked=True)).unique()

airflow-core/src/airflow/models/taskinstance.py

@@ -100,7 +100,6 @@
 from airflow.ti_deps.dep_context import DepContext
 from airflow.ti_deps.dependencies_deps import REQUEUEABLE_DEPS, RUNNING_DEPS
 from airflow.ti_deps.deps.ready_to_reschedule import ReadyToRescheduleDep
-from airflow.utils.helpers import prune_dict
 from airflow.utils.log.logging_mixin import LoggingMixin
 from airflow.utils.net import get_hostname
 from airflow.utils.platform import getuser
@@ -757,9 +756,8 @@ def __hash__(self):
     @property
     def stats_tags(self) -> dict[str, str]:
         """Returns task instance tags."""
-        return prune_dict(
-            {"dag_id": self.dag_id, "task_id": self.task_id, "team_name": getattr(self, "_team_name", None)}
-        )
+        # Reuse the dag run's tags and add the task-level ones.
+        return {**self.dag_run.stats_tags, "task_id": self.task_id}
 
     @staticmethod
     def insert_mapping(

airflow-core/tests/unit/jobs/test_scheduler_job.py

@@ -518,8 +518,13 @@ def test_process_executor_events(self, mock_get_backend, mock_task_callback, dag
                     "scheduler.tasks.killed_externally",
                     tags={"dag_id": dag_id, "task_id": ti1.task_id},
                 ),
-                mock.call("operator_failures_EmptyOperator", tags={"dag_id": dag_id, "task_id": ti1.task_id}),
-                mock.call("ti_failures", tags={"dag_id": dag_id, "task_id": ti1.task_id}),
+                mock.call(
+                    "operator_failures_EmptyOperator",
+                    tags={"dag_id": dag_id, "task_id": ti1.task_id, "run_type": "manual"},
+                ),
+                mock.call(
+                    "ti_failures", tags={"dag_id": dag_id, "task_id": ti1.task_id, "run_type": "manual"}
+                ),
             ],
             any_order=True,
         )
@@ -645,8 +650,11 @@ def test_process_executor_events_with_no_callback(self, mock_get_backend, mock_t
                     "scheduler.tasks.killed_externally",
                     tags={"dag_id": dag_id, "task_id": task_id},
                 ),
-                mock.call("operator_failures_EmptyOperator", tags={"dag_id": dag_id, "task_id": task_id}),
-                mock.call("ti_failures", tags={"dag_id": dag_id, "task_id": task_id}),
+                mock.call(
+                    "operator_failures_EmptyOperator",
+                    tags={"dag_id": dag_id, "task_id": task_id, "run_type": "manual"},
+                ),
+                mock.call("ti_failures", tags={"dag_id": dag_id, "task_id": task_id, "run_type": "manual"}),
             ],
             any_order=True,
         )
@@ -9793,8 +9801,10 @@ def test_multi_team_config_disabled_uses_legacy_behavior(self, dag_maker, mock_e
             assert result2 == mock_executors[1]  # Matched by executor name
 
     @conf_vars({("core", "multi_team"): "true"})
-    def test_multi_team_sets_team_name_on_task_instances(self, dag_maker, mock_executors, session):
-        """Test that _team_name is set on TaskInstance objects during the scheduling loop."""
+    @mock.patch("airflow._shared.observability.metrics.stats.timing")
+    def test_multi_team_sets_team_name_on_task_instances(self, mock_timing, dag_maker, session):
+        """The scheduling loop resolves the bundle's team onto the dag run, so the QUEUED
+        state-change metric (emitted via TaskInstance.stats_tags) carries team_name."""
         clear_db_teams()
         clear_db_dag_bundles()
 
@@ -9813,19 +9823,21 @@ def test_multi_team_sets_team_name_on_task_instances(self, dag_maker, mock_execu
         dr = dag_maker.create_dagrun()
         ti = dr.get_task_instance("task_a", session=session)
         ti.state = State.SCHEDULED
+        ti.scheduled_dttm = timezone.utcnow()
+        session.merge(ti)
         session.flush()
 
         scheduler_job = Job()
         self.job_runner = SchedulerJobRunner(job=scheduler_job)
-        self.job_runner._multi_team = True
 
-        # Simulate what _executable_task_instances_to_queued does
-        dag_id_to_team_name = self.job_runner._get_team_names_for_dag_ids(["dag_a"], session)
-        if team_name := dag_id_to_team_name.get(ti.dag_id):
-            ti._team_name = team_name
+        queued_tis = self.job_runner._executable_task_instances_to_queued(max_tis=32, session=session)
 
-        assert ti._team_name == "team_a"
-        assert ti.stats_tags == {"dag_id": "dag_a", "task_id": "task_a", "team_name": "team_a"}
+        assert {t.key for t in queued_tis} == {ti.key}
+        scheduled_calls = [
+            c for c in mock_timing.call_args_list if c.args and c.args[0] == "task.scheduled_duration"
+        ]
+        assert scheduled_calls, "expected a task.scheduled_duration metric on QUEUED transition"
+        assert scheduled_calls[0].kwargs["tags"]["team_name"] == "team_a"
 
     @conf_vars({("core", "multi_team"): "true"})
     def test_do_scheduling_multi_team_schedules_task_instances(self, dag_maker, session):