**Description**


It would be great if the metrics we publish in Airflow change to include TAGs for "variable" parts of those (such as dag_id, task_id, pool name) rather than being part of the metrics.

See here: https://airflow.apache.org/docs/1.10.12/metrics.html, https://airflow.readthedocs.io/en/latest/logging-monitoring/metrics.html  - those variable parts should be tags.

We might consider to change those metrics by default in 2.0 and possibly introduce a backport in 1.10.13 to allow people to migrate.


**Use case / motivation**

Having the variable parts of the metrics makes it really difficult to aggregate the metrics. Seems that statsd python library does not support tags, but there are possible solutions by installing extensions:

https://stackoverflow.com/questions/49852654/sending-statsd-metrics-with-tags-from-python-client

**Related Issues**

<!-- Is there currently another issue associated with this? -->

Note, the stat will only be emitted if the DagRun is a scheduler triggered one
        (i.e. external_trigger is False).
        """
        if self.state == State.RUNNING:
            return
        if self.external_trigger:
            return
        if not finished_tis:
            return
        try:
            dag = self.get_dag()
            if not dag.timetable.periodic:
                
                return
            ordered_tis_by_start_date = [ti for ti in finished_tis if ti.start_date]
            ordered_tis_by_start_date.sort(key=lambda ti: ti.start_date, reverse=False)
            first_start_date = ordered_tis_by_start_date[0].start_date if ordered_tis_by_start_date else None
            if first_start_date:
                
                
                
                
                
                data_interval_end = dag.get_run_data_interval(self).end
                true_delay = first_start_date - data_interval_end
                if true_delay.total_seconds() > 0:
                    Stats.timing(f"dagrun.{dag.dag_id}.first_task_scheduling_delay", true_delay)
        except Exception:
            self.log.warning("Failed to record first_task_scheduling_delay metric:", exc_info=True)
    def _emit_duration_stats_for_finished_state(self):