williamBartos opened a new issue #8743: URL: https://github.com/apache/airflow/issues/8743
**Description** A recent pr enabled dogstatsd support for Airflow metrics: https://github.com/apache/airflow/pull/7376. While this _enables_ the use of dogstatsd, the code sending metrics to [SafeDogStatsdLogger](https://github.com/apache/airflow/blob/master/airflow/stats.py#L147) doesn't utilize tagging and instead sends a unique, monolithic metric that cant be aggregated across unique identifiers such as `<dag_id>`. Note: SafeDogStatsdLogger does take a list of tags as an argument. An example here are the timer metrics sent by a DagRun, such as `dagrun.duration.failed.<dag_id>`. When sent by the DagRun object, `<dag_id>` isn't a tag but part of the entire metric itself: https://github.com/apache/airflow/blob/master/airflow/models/dagrun.py#L412-L420 **What is the problem here?** By sending metrics to DataDog without tags, it becomes impossible to aggregate metrics across `<dag_id>` because each `dagrun.duration.failed.<dag_id>` sent by a DAG is completely _unique_ to that `<dag_id>`. If I have 20 dags in production and want to monitor `dagrun.duration.failed.<dag_id>`, that means I'll need 20 separate monitors!  But if `<dag_id>` is sent as a _tag_, a single monitor could be used and DataDog can group the metric by `<dag_id>`. **Use case / motivation** The current way metrics are sent to DataDog isn't scalable as its preventing a user from aggregating common metrics across unique tags. Following the DagRun example given above, the information needed to send this metric as a tag is available. Given this line of code: https://github.com/apache/airflow/blob/master/airflow/models/dagrun.py#L418 and the accompanying function definition: https://github.com/apache/airflow/blob/master/airflow/stats.py#L172 we can modify the function call to send `<dag_id>` as a tag: toy example: ``` duration = (self.end_date - self.start_date) if self.state is State.SUCCESS: if isinstance(Stats, SafeDogStatsdLogger) Stats.timing('dagrun.duration.success', duration, tags=[self.dag_id]) else: Stats.timing('dagrun.duration.success.{}'.format(self.dag_id), duration) ``` There are also other ways to accomplish this - I'm willing to discuss that here or as part of a PR, and to implement the agreed upon solution. **Related Issues** This is the ticket that created the `SafeDogStatsdLogger` class: https://github.com/apache/airflow/pull/7376 ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected]
