fix(metrics/otel): bracket IPv6 host literals in exporter endpoint URL

shared/observability/src/airflow_shared/observability/common.py

@@ -30,6 +30,21 @@
 log = structlog.getLogger(__name__)
 
 
+def _format_url_host(host: str | None) -> str | None:
+    """
+    Bracket IPv6 host literals for embedding in a URL authority.
+
+    Per RFC 3986 §3.2.2, IPv6 hosts in a URI must be enclosed in square
+    brackets so the ``:`` separators do not conflict with the ``host:port``
+    delimiter. Hostnames, IPv4 literals, and already-bracketed v6 literals
+    are returned unchanged. ``None`` is passed through so existing
+    error-logging paths keep their shape.
+    """
+    if host is not None and ":" in host and not host.startswith("["):
+        return f"[{host}]"
+    return host
+
+
 def get_otel_data_exporter(
     *,
     otel_env_config: OtelEnvConfig,
@@ -106,7 +121,7 @@ def get_otel_data_exporter(
 
         endpoint_suffix = "traces" if otel_env_config.data_type == OtelDataType.TRACES else "metrics"
 
-        endpoint_str = f"{protocol}://{host}:{port}/v1/{endpoint_suffix}"
+        endpoint_str = f"{protocol}://{_format_url_host(host)}:{port}/v1/{endpoint_suffix}"
         if otel_env_config.data_type == OtelDataType.TRACES:
             exporter = OTLPSpanExporter(endpoint=endpoint_str)
         else:

shared/observability/tests/observability/metrics/test_otel_logger.py

@@ -394,6 +394,38 @@ def test_timer_start_and_stop_manually_send_true(self, mock_time, name):
                 "grpc",
                 id="type_specific_vars_take_precedence",
             ),
+            pytest.param(
+                {},
+                "::1",
+                "4318",
+                "http://[::1]:4318/v1/metrics",
+                "http",
+                id="airflow_config_ipv6_loopback_is_bracketed",
+            ),
+            pytest.param(
+                {},
+                "2001:db8::1",
+                "4318",
+                "http://[2001:db8::1]:4318/v1/metrics",
+                "http",
+                id="airflow_config_ipv6_literal_is_bracketed",
+            ),
+            pytest.param(
+                {},
+                "[::1]",
+                "4318",
+                "http://[::1]:4318/v1/metrics",
+                "http",
+                id="airflow_config_already_bracketed_ipv6_is_preserved",
+            ),
+            pytest.param(
+                {},
+                "10.0.0.1",
+                "4318",
+                "http://10.0.0.1:4318/v1/metrics",
+                "http",
+                id="airflow_config_ipv4_literal_passes_through_unchanged",
+            ),
         ],
     )
     def test_config_priorities(