feat(core): add request hook to enrich T4 network spans with GCP resource attributes (E)

packages/google-api-core/google/api_core/_observability.py

@@ -18,14 +18,15 @@
 
 from __future__ import annotations
 
+import urllib.parse
 from typing import TYPE_CHECKING, Any, Callable, Sequence
 
 from google.api_core import _feature_gating_helpers
 from google.api_core.client_options import ClientOptions
 
 if TYPE_CHECKING:
     # flake8: grpc, trace, and ClientInterceptor are imported only for static analysis and type annotations
-    # The `# noqa: F401` comment avoids flake8 "imported but not used" errors.
+    # The 'noqa: F401' comment avoids flake8 "imported but not used" errors.
     import grpc  # noqa: F401
     import opentelemetry.trace  # noqa: F401
 
@@ -64,6 +65,152 @@ def is_otel_capabilities_enabled(
     return False
 
 
+def _extract_endpoint_attributes(
+    client_options: ClientOptions | dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Extracts server.address, server.port (if non-default), and url.domain from client options if present.
+
+    Args:
+        client_options: The client options object or dictionary.
+
+    Returns:
+        dict[str, Any]: A dictionary containing url.domain and, if an api_endpoint is configured,
+            server.address and non-default server.port.
+    """
+    attrs: dict[str, Any] = {}
+    endpoint = None
+    universe_domain = None
+
+    if isinstance(client_options, dict):
+        endpoint = client_options.get("api_endpoint")
+        universe_domain = client_options.get("universe_domain")
+    elif client_options is not None:
+        endpoint = getattr(client_options, "api_endpoint", None)
+        universe_domain = getattr(client_options, "universe_domain", None)
+
+    attrs["url.domain"] = universe_domain or "googleapis.com"
+
+    if endpoint and isinstance(endpoint, str):
+        target = endpoint if "//" in endpoint else f"//{endpoint}"
+        parsed = urllib.parse.urlsplit(target)
+        if parsed.hostname:
+            attrs["server.address"] = parsed.hostname
+        if parsed.port:
+            scheme = parsed.scheme.lower()
+            is_default_port = (parsed.port == 443 and scheme in ("https", "")) or (
+                parsed.port == 80 and scheme == "http"
+            )
+            if not is_default_port:
+                attrs["server.port"] = parsed.port
+    return attrs
+
+
+def _extract_grpc_request_attributes(request: Any) -> dict[str, Any]:
+    """Extracts Google Cloud T4 semantic and resource attributes from a gRPC request object.
+
+    Args:
+        request: The gRPC request object.
+
+    Returns:
+        dict[str, Any]: A dictionary of semantic attributes.
+    """
+    attrs: dict[str, Any] = {
+        "rpc.system.name": "grpc",
+    }
+    if request is None:
+        return attrs
+
+    resend_count = getattr(request, "resend_count", None)
+    if isinstance(resend_count, int) and resend_count > 0:
+        attrs["gcp.grpc.resend_count"] = resend_count
+
+    return attrs
+
+
+def _make_grpc_client_request_hook(
+    endpoint_attrs: dict[str, Any] | None = None,
+) -> Callable[[Any, Any], None]:
+    """Creates an OpenTelemetry gRPC client request hook with optional endpoint attributes.
+
+    Args:
+        endpoint_attrs: Optional static endpoint attributes to attach to every span.
+
+    Returns:
+        Callable[[Any, Any], None]: The request hook callback.
+    """
+    static_attrs = dict(endpoint_attrs) if endpoint_attrs else {}
+
+    def client_request_hook(span: Any, request: Any) -> None:
+        if span is None or not getattr(span, "is_recording", lambda: True)():
+            return
+
+        # Upstream opentelemetry-instrumentation-grpc names spans with a leading slash
+        # (e.g. "/package.Service/Method") and sets only the short name on rpc.method.
+        # Normalize span.name and rpc.method to the fully-qualified name without leading slash.
+        span_name = getattr(span, "name", None)
+        clean_method_name = None
+        if isinstance(span_name, str) and span_name.startswith("/"):
+            clean_method_name = span_name.lstrip("/")
+            if hasattr(span, "update_name"):
+                span.update_name(clean_method_name)
+
+        # Remove duplicate legacy rpc.system attribute set by stock instrumentation
+        # in favor of modern rpc.system.name ("grpc") per PRD changelog.
+        span_attributes = getattr(span, "_attributes", None)
+        if hasattr(span_attributes, "pop"):
+            span_attributes.pop("rpc.system", None)
+
+        attrs = _extract_grpc_request_attributes(request)
+        if clean_method_name:
+            attrs["rpc.method"] = clean_method_name
+        if static_attrs:
+            attrs.update(static_attrs)
+        for key, value in attrs.items():
+            span.set_attribute(key, value)
+
+    return client_request_hook
+
+
+_grpc_client_request_hook = _make_grpc_client_request_hook()
+
+
+def _grpc_client_response_hook(span: Any, response: Any) -> None:
+    """OpenTelemetry gRPC client response hook to record response status code.
+
+    Args:
+        span: The OpenTelemetry span.
+        response: The gRPC response object or details.
+    """
+    if span is None or not hasattr(span, "set_attribute"):
+        return
+
+    status = getattr(span, "status", None)
+    status_code = getattr(status, "status_code", None)
+    try:
+        from opentelemetry.trace.status import StatusCode
+
+        if status_code == StatusCode.ERROR:
+            span_attrs = (
+                getattr(span, "attributes", None)
+                or getattr(span, "_attributes", None)
+                or {}
+            )
+            grpc_code = span_attrs.get("rpc.grpc.status_code")
+            if grpc_code is not None:
+                from google.api_core import exceptions
+
+                if grpc_code in exceptions._INT_TO_GRPC_CODE:
+                    span.set_attribute(
+                        "rpc.response.status_code",
+                        exceptions._INT_TO_GRPC_CODE[grpc_code].name,
+                    )
+                    return
+    except Exception:
+        pass
+
+    span.set_attribute("rpc.response.status_code", "OK")
+
+
 def _get_tracer_provider(
     client_options: ClientOptions | dict[str, Any] | None = None,
 ) -> opentelemetry.trace.TracerProvider | None:
@@ -101,8 +248,13 @@ def get_otel_interceptor(
 
     import opentelemetry.instrumentation.grpc as otel_grpc  # type: ignore[import-not-found]
 
+    endpoint_attrs = _extract_endpoint_attributes(client_options)
+    request_hook = _make_grpc_client_request_hook(endpoint_attrs)
+
     interceptor: ClientInterceptor = otel_grpc.client_interceptor(
-        tracer_provider=_get_tracer_provider(client_options)
+        tracer_provider=_get_tracer_provider(client_options),
+        request_hook=request_hook,
+        response_hook=_grpc_client_response_hook,
     )
 
     def otel_interceptor(channel: grpc.Channel) -> grpc.Channel:
@@ -130,6 +282,11 @@ def get_otel_async_interceptor(
     # Ignored by mypy: Optional dependency only loaded if early-return is skipped
     import opentelemetry.instrumentation.grpc as otel_grpc  # type: ignore[import-not-found]
 
+    endpoint_attrs = _extract_endpoint_attributes(client_options)
+    request_hook = _make_grpc_client_request_hook(endpoint_attrs)
+
     return otel_grpc.aio_client_interceptors(
-        tracer_provider=_get_tracer_provider(client_options)
+        tracer_provider=_get_tracer_provider(client_options),
+        request_hook=request_hook,
+        response_hook=_grpc_client_response_hook,
     )