FrankYang0529 commented on code in PR #72458:
URL: https://github.com/apache/airflow/pull/72458#discussion_r3958162705
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -604,7 +605,12 @@ def dag_tags_for_stats(self) -> dict[str, str]:
# 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)
+ tag_names = (tag.name for tag in self.dag_model.tags)
+ if airflow_conf.getboolean("metrics", "statsd_datadog_enabled",
fallback=False) or airflow_conf.getboolean(
+ "metrics", "statsd_on", fallback=False
+ ):
+ return build_dag_metric_tags(tag_names)
+ return expand_dag_tags(tag_names)
Review Comment:
Followup from https://github.com/apache/airflow/pull/72364. How about moving
tag normalization to `prepare_stat_with_tags`? It already check
`statsd_datadog_enabled`, so we don't need to update logic here.
https://github.com/apache/airflow/blob/f28c8ce27370802d9aba6708f56f62379bae8a9b/shared/observability/src/airflow_shared/observability/metrics/statsd_logger.py#L44-L62
```diff
diff --git
a/shared/observability/src/airflow_shared/observability/metrics/statsd_logger.py
b/shared/observability/src/airflow_shared/observability/metrics/statsd_logger.py
index eba2035817..bf07f4277f 100644
---
a/shared/observability/src/airflow_shared/observability/metrics/statsd_logger.py
+++
b/shared/observability/src/airflow_shared/observability/metrics/statsd_logger.py
@@ -18,6 +18,7 @@
from __future__ import annotations
import logging
+import re
from collections.abc import Callable
from functools import wraps
from typing import TYPE_CHECKING, TypeVar, cast
@@ -26,6 +27,7 @@ from statsd import StatsClient, UnixSocketStatsClient
from .protocols import Timer
from .validators import (
+ ALLOWED_CHARACTERS,
PatternAllowListValidator,
PatternBlockListValidator,
get_validator,
@@ -40,6 +42,13 @@ T = TypeVar("T", bound=Callable)
log = logging.getLogger(__name__)
+_INVALID_TAG_CHARS_RE =
re.compile(f"[^{re.escape(''.join(ALLOWED_CHARACTERS))}]")
+
+
+def _normalize_influx_tag(part: str) -> str:
+ """Replace characters the stat name validator would reject with
underscores."""
+ return _INVALID_TAG_CHARS_RE.sub("_", part)
+
def prepare_stat_with_tags(fn: T) -> T:
"""Add tags to stat with influxdb standard format if
influxdb_tags_enabled is True."""
@@ -52,9 +61,11 @@ def prepare_stat_with_tags(fn: T) -> T:
if stat is not None and tags is not None:
for k, v in tags.items():
if self.metric_tags_validator.test(k):
- v_str = "true" if v == "" else v
+ v_str = "true" if v == "" else str(v)
if all(c not in [",", "="] for c in f"{v_str}{k}"):
- stat += f",{k}={v_str}"
+ # Tags become part of the stat name here, so
they must satisfy the same
+ # character rules as the name or
``validate_stat`` drops the whole metric.
+ stat +=
f",{_normalize_influx_tag(k)}={_normalize_influx_tag(v_str)}"
else:
log.error("Dropping invalid tag: %s=%s.", k, v)
return fn(self, stat, *args, tags=tags, **kwargs)
```
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
