**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? -->

"""
        Enqueues TaskInstances for execution.
        There are three steps:
        1. Pick TIs by priority with the constraint that they are in the expected states
        and that we do exceed max_active_runs or pool limits.
        2. Change the state for the TIs above atomically.
        3. Enqueue the TIs in the executor.
        HA note: This function is a "critical section" meaning that only a single executor process can execute
        this function at the same time. This is achieved by doing ``SELECT ... from pool FOR UPDATE``. For DBs
        that support NOWAIT, a "blocked" scheduler will skip this and continue on with other tasks (creating
        new DAG runs, progressing TIs from None to SCHEDULED etc.); DBs that don't support this (such as
        MariaDB or MySQL 5.x) the other schedulers will wait for the lock before continuing.
        :param session:
        :return: Number of task instance with state changed.
        """
        if self.max_tis_per_query == 0:
            max_tis = self.executor.slots_available
        else:
            max_tis = min(self.max_tis_per_query, self.executor.slots_available)
        queued_tis = self._executable_task_instances_to_queued(max_tis, session=session)
        self._enqueue_task_instances_with_queued_state(queued_tis, session=session)
        return len(queued_tis)
    def _process_executor_events(self, session: Session) -> int: