refactor: remove redundant checks, lift OTel length restriction, and …

…apply minor refinements

airflow/metrics/datadog_logger.py

@@ -55,28 +55,26 @@
     )
 
 
-def prepare_stat_with_tags(fn: T) -> T:
+def prepare_metric_name_with_tags(fn: T) -> T:
     """Prepare tags and stat."""
 
     @wraps(fn)
-    def wrapper(self, stat: str | None = None, *args, tags: dict[str, str] | None = None, **kwargs):
-        stat = stat or ""
-
+    def wrapper(self, metric_name: str | None = None, *args, tags: dict[str, str] | None = None, **kwargs):
         if tags and self.metrics_tags:
             valid_tags: dict[str, str] = {}
             for k, v in tags.items():
                 if self.metric_tags_validator.test(k):
-                    if all(c not in [",", "="] for c in f"{v}{k}"):
+                    if ":" not in f"{v}{k}":
                         valid_tags[k] = v
                     else:
-                        log.error("Dropping invalid tag: %s=%s.", k, v)
+                        log.error("Dropping invalid tag: %s:%s.", k, v)
             tags_list = [f"{key}:{value}" for key, value in valid_tags.items()]
         else:
             tags_list = []
 
         kwargs["tags"] = tags_list
 
-        return fn(self, stat, *args, **kwargs)
+        return fn(self, metric_name, *args, **kwargs)
 
     return cast(T, wrapper)
 
@@ -96,7 +94,7 @@ def __init__(
         self.metrics_tags = metrics_tags
         self.metric_tags_validator = metric_tags_validator
 
-    @prepare_stat_with_tags
+    @prepare_metric_name_with_tags
     @validate_stat
     def incr(
         self,
@@ -112,7 +110,7 @@ def incr(
             return self.dogstatsd.increment(metric=full_metric_name, value=count, tags=tags, sample_rate=rate)
         return None
 
-    @prepare_stat_with_tags
+    @prepare_metric_name_with_tags
     @validate_stat
     def decr(
         self,
@@ -128,7 +126,7 @@ def decr(
             return self.dogstatsd.decrement(metric=full_metric_name, value=count, tags=tags, sample_rate=rate)
         return None
 
-    @prepare_stat_with_tags
+    @prepare_metric_name_with_tags
     @validate_stat
     def gauge(
         self,
@@ -145,7 +143,7 @@ def gauge(
             return self.dogstatsd.gauge(metric=full_metric_name, value=value, tags=tags, sample_rate=rate)
         return None
 
-    @prepare_stat_with_tags
+    @prepare_metric_name_with_tags
     @validate_stat
     def timing(
         self,
@@ -166,7 +164,7 @@ def timing(
             return self.dogstatsd.timing(metric=full_metric_name, value=dt, tags=tags)
         return None
 
-    @prepare_stat_with_tags
+    @prepare_metric_name_with_tags
     @validate_stat
     def timer(
         self,

airflow/metrics/statsd_logger.py

@@ -53,8 +53,6 @@ def prepare_metric_name_with_tags(fn: T) -> T:
     def wrapper(
         self, metric_name: str | None = None, tags: dict[str, str] | None = None
     ) -> Callable[[str], str]:
-        metric_name = metric_name or ""
-
         if self.influxdb_tags_enabled and tags:
             valid_tags: dict[str, str] = {}
 

airflow/metrics/validators.py

@@ -155,26 +155,11 @@ def stat_name_otel_handler(
     if not (isinstance(stat_name, str) and isinstance(stat_prefix, str)):
         raise InvalidStatsNameException("Stat name and prefix must both be strings.")
 
-    if len(proposed_stat_name) > OTEL_NAME_MAX_LENGTH:
-        # If the name is in the exceptions list, do not fail it for being too long.
-        # It may still be deemed invalid for other reasons below.
-        for exemption in BACK_COMPAT_METRIC_NAMES:
-            if re2.match(exemption, stat_name):
-                # There is a back-compat exception for this name; proceed
-                name_length_exemption = True
-                matched_exemption = exemption.pattern
-                break
-        else:
-            raise InvalidStatsNameException(
-                f"Invalid stat name: {proposed_stat_name}.  Please see "
-                f"https://opentelemetry.io/docs/reference/specification/metrics/api/#instrument-name-syntax"
-            )
-
     # `stat_name_default_handler` throws InvalidStatsNameException if the
     # provided value is not valid or returns the value if it is.  We don't
     # need the return value but will make use of the validation checks. If
     # no exception is thrown, then the proposed name meets OTel requirements.
-    stat_name_default_handler(proposed_stat_name, max_length=999 if name_length_exemption else max_length)
+    stat_name_default_handler(proposed_stat_name, max_length=999)
 
     # This warning is down here instead of up above because the exemption only
     # applies to the length and a name may still be invalid for other reasons.

tests/core/test_otel_logger.py

@@ -35,12 +35,11 @@
     _is_up_down_counter,
     full_name,
 )
-from airflow.metrics.validators import BACK_COMPAT_METRIC_NAMES, MetricNameLengthExemptionWarning
+from airflow.metrics.validators import BACK_COMPAT_METRIC_NAMES
 
 INVALID_STAT_NAME_CASES = [
     (None, "can not be None"),
     (42, "is not a string"),
-    ("X" * OTEL_NAME_MAX_LENGTH, "too long"),
     ("test/$tats", "contains invalid characters"),
 ]
 
@@ -100,8 +99,7 @@ def test_old_name_exception_works(self, caplog):
         name = "task_instance_created_OperatorNameWhichIsSuperLongAndExceedsTheOpenTelemetryCharacterLimit"
         assert len(name) > OTEL_NAME_MAX_LENGTH
 
-        with pytest.warns(MetricNameLengthExemptionWarning):
-            self.stats.incr(name)
+        self.stats.incr(name)
 
         self.meter.get_meter().create_counter.assert_called_once_with(
             name=(full_name(name)[:OTEL_NAME_MAX_LENGTH])
@@ -388,12 +386,6 @@ def test_get_name_invalid_cases(self):
         with pytest.raises(InvalidStatsNameException):
             self.stats.get_name(invalid_name)
 
-    def test_get_name_too_long(self):
-        # Edge case: Name exceeds max length
-        long_name = "a" * (OTEL_NAME_MAX_LENGTH + 1)
-        with pytest.raises(InvalidStatsNameException, match="Invalid stat name.*Please see"):
-            self.stats.get_name(long_name)
-
     def test_get_name_special_characters(self):
         # Edge case: Name contains invalid special characters
         invalid_name = "invalid@name!"

tests/core/test_stats.py

@@ -485,7 +485,7 @@ def test_increment_counter(self):
     def test_increment_counter_with_tags(self):
         self.stats.incr(
             "test_stats_run.delay",
-            tags={"key0": "0", "key1": "val1", "key2": "val2"},
+            tags={"key0": 0, "key1": "val1", "key2": "val2"},
         )
         self.statsd_client.incr.assert_called_once_with("test_stats_run.delay,key0=0,key1=val1", 1, 1)