(airflow) branch main updated: Fix Dag metrics dropped when a tag contains invalid characters (#72364)

[potiuk]

This is an automated email from the ASF dual-hosted git repository.

potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/main by this push:
     new baa949fd4b0 Fix Dag metrics dropped when a tag contains invalid 
characters (#72364)
baa949fd4b0 is described below

commit baa949fd4b030ebdfe9c9aa1c27543ac8e64a76a
Author: PoAn Yang <pay...@apache.org>
AuthorDate: Tue Sep 8 03:24:23 2026 +0900

    Fix Dag metrics dropped when a tag contains invalid characters (#72364)
    
    Signed-off-by: PoAn Yang <pay...@apache.org>
---
 .../src/airflow_shared/observability/metrics/stats.py   |  4 +++-
 .../tests/observability/metrics/test_stats.py           | 17 ++++++++++++++++-
 2 files changed, 19 insertions(+), 2 deletions(-)

diff --git 
a/shared/observability/src/airflow_shared/observability/metrics/stats.py 
b/shared/observability/src/airflow_shared/observability/metrics/stats.py
index 7b51e580cd0..4904a18d0b6 100644
--- a/shared/observability/src/airflow_shared/observability/metrics/stats.py
+++ b/shared/observability/src/airflow_shared/observability/metrics/stats.py
@@ -48,7 +48,9 @@ def build_dag_metric_tags(tag_names: Iterable[str]) -> 
dict[str, str]:
     result: dict[str, str] = {}
     for name in tag_names:
         key, _, value = name.partition(":")
-        result[key] = value
+        result[normalize_name_for_stats(key, log_warning=False)] = 
normalize_name_for_stats(
+            value, log_warning=False
+        )
     return result
 
 
diff --git a/shared/observability/tests/observability/metrics/test_stats.py 
b/shared/observability/tests/observability/metrics/test_stats.py
index d9f11796bf8..ee3e9dc8a81 100644
--- a/shared/observability/tests/observability/metrics/test_stats.py
+++ b/shared/observability/tests/observability/metrics/test_stats.py
@@ -528,6 +528,10 @@ class TestStatsWithInfluxDBEnabled:
         self.stats.incr(stat, tags=tags)
         self.statsd_client.incr.assert_called_once_with(expected, 1, 1)
 
+    def test_increment_counter_with_dag_tags(self):
+        self.stats.incr("test_stats_run.delay", 
tags=build_dag_metric_tags(["my tag", "env:pro d"]))
+        
self.statsd_client.incr.assert_called_once_with("test_stats_run.delay,my_tag=true,env=pro_d",
 1, 1)
+
 
 def always_invalid(stat_name):
     raise InvalidStatsNameException(f"Invalid name: {stat_name}")
@@ -853,13 +857,24 @@ class TestCustomStatsName:
             {"production": "", "env": "prod", "team": "data"},
             id="mixed",
         ),
-        pytest.param(["a:b:c"], {"a": "b:c"}, id="value-with-colon"),
+        pytest.param(["a:b:c"], {"a": "b_c"}, id="value-with-colon"),
         pytest.param(["env:"], {"env": ""}, id="trailing-colon-is-standalone"),
+        pytest.param(["my tag"], {"my_tag": ""}, id="space-in-key"),
+        pytest.param(["env:pro d"], {"env": "pro_d"}, id="space-in-value"),
+        pytest.param(["a,b"], {"a_b": ""}, id="comma-in-key"),
+        pytest.param(["env:a=b"], {"env": "a_b"}, id="equals-in-value"),
+        pytest.param(["déjà"], {"d_j_": ""}, id="non-ascii"),
     ],
 )
 def test_build_dag_metric_tags(tag_names: list[str], expected: dict[str, str]) 
-> None:
     assert build_dag_metric_tags(tag_names) == expected
 
 
+@mock.patch.object(airflow_shared.observability.metrics.stats, "log", 
autospec=True)
+def test_build_dag_metric_tags_does_not_warn(mock_log) -> None:
+    build_dag_metric_tags(["my tag"])
+    mock_log.warning.assert_not_called()
+
+
 def test_build_dag_metric_tags_accepts_generator() -> None:
     assert build_dag_metric_tags(name for name in ["env:prod"]) == {"env": 
"prod"}