feat(gapic): add OpenTelemetry T3 client method span wrapping in gapic_v1.method (D)

packages/google-api-core/google/api_core/gapic_v1/method.py

@@ -18,11 +18,12 @@
 compression, pagination, and long-running operations to gRPC methods.
 """
 
+import contextlib
 import enum
 import functools
-from typing import List, Tuple
+from typing import Any, List, Optional, Tuple
 
-from google.api_core import grpc_helpers
+from google.api_core import _observability, grpc_helpers
 from google.api_core.gapic_v1 import client_info
 from google.api_core.timeout import TimeToDeadlineTimeout
 
@@ -104,6 +105,36 @@ def _extract_metrics_header(metadata) -> Tuple[str, List[Tuple[str, str]]]:
     return metric_str, arbitrary_metadata
 
 
+def _extract_rpc_identity(
+    target: Any, method_name: Optional[str] = None
+) -> Tuple[str, str, str]:
+    """Extract (full_rpc_name, service_name, rpc_method_name) from an explicit method name or target callable.
+
+    Args:
+        target: The underlying callable method.
+        method_name: Optional explicit RPC name (e.g. "/google.cloud.secretmanager.v1.SecretManagerService/AccessSecretVersion").
+
+    Returns:
+        Tuple[str, str, str]: A 3-tuple of (full_rpc_name, service_name, rpc_method_name).
+    """
+    if method_name:
+        method_str = method_name.lstrip("/")
+        service, _, method = method_str.rpartition("/")
+        return method_str, service, method
+
+    raw_method = getattr(target, "_method", None)
+    if raw_method and isinstance(raw_method, (str, bytes)):
+        if isinstance(raw_method, bytes):
+            raw_method = raw_method.decode("utf-8")
+        method_str = raw_method.lstrip("/")
+        service, _, method = method_str.rpartition("/")
+        return method_str, service, method
+
+    service = "google.api_core"
+    method = getattr(target, "__name__", "call")
+    return f"{service}/{method}", service, method
+
+
 class _GapicCallable(object):
     """Callable that applies retry, timeout, and metadata logic.
 
@@ -123,6 +154,14 @@ class _GapicCallable(object):
             provided to the RPC method on every invocation. This is merged with
             any metadata specified during invocation. If ``None``, no
             additional metadata will be passed to the RPC method.
+        method_name (Optional[str]): The optional explicit full RPC method name
+            (e.g. "/google.cloud.secretmanager.v1.SecretManagerService/AccessSecretVersion").
+        tracer_provider (Optional[Any]): Optional custom OpenTelemetry TracerProvider
+            to obtain the tracer from.
+        rpc_system (Optional[str]): The RPC system (defaults to "grpc"). If not "grpc",
+            tracing will not be enabled.
+        is_streaming (bool): Whether the callable is a streaming RPC. Streaming RPC
+            tracing is currently gated and will not produce T3 spans.
     """
 
     def __init__(
@@ -132,11 +171,20 @@ def __init__(
         timeout,
         compression,
         metadata=None,
+        method_name=None,
+        tracer_provider=None,
+        rpc_system="grpc",
+        is_streaming=False,
     ):
         self._target = target
         self._retry = retry
         self._timeout = timeout
         self._compression = compression
+        self._rpc_system = rpc_system
+        self._is_streaming = is_streaming
+        self._rpc_method_name, self._rpc_service, self._rpc_method = (
+            _extract_rpc_identity(target, method_name)
+        )
         # Pre-extract the x-goog-api-client header from the initialized metadata.
         self._x_goog_api_client, remaining = _extract_metrics_header(metadata)
         self._static_metadata = tuple(remaining)
@@ -148,6 +196,35 @@ def __init__(
         else:
             self._default_metadata = self._static_metadata
 
+        # Resolve and cache the OpenTelemetry tracer and attributes once at initialization.
+        # Tracing is gated to non-streaming gRPC calls for Tier 3 method spans.
+        self._tracer = None
+        self._span_name = None
+        self._span_attributes = None
+        if (
+            rpc_system == "grpc"
+            and not is_streaming
+            and _observability.is_otel_capabilities_enabled()
+        ):
+            try:
+                from opentelemetry import trace
+
+                if tracer_provider is not None:
+                    self._tracer = tracer_provider.get_tracer("google.api_core")
+                else:
+                    self._tracer = trace.get_tracer("google.api_core")
+
+                self._span_name = self._rpc_method_name
+                self._span_attributes = {
+                    "rpc.system": "grpc",
+                    "rpc.service": self._rpc_service,
+                    "rpc.method": self._rpc_method,
+                }
+            except Exception:  # pragma: NO COVER
+                self._tracer = None
+                self._span_name = None
+                self._span_attributes = None
+
     def __call__(
         self, *args, timeout=DEFAULT, retry=DEFAULT, compression=DEFAULT, **kwargs
     ):
@@ -186,7 +263,29 @@ def __call__(
         if self._compression is not None:
             kwargs["compression"] = compression
 
-        return wrapped_func(*args, **kwargs)
+        span_context_manager = contextlib.nullcontext()
+        if self._tracer is not None and self._span_name is not None:
+            try:
+                from opentelemetry import trace
+
+                span_context_manager = self._tracer.start_as_current_span(
+                    self._span_name,
+                    kind=trace.SpanKind.CLIENT,
+                    attributes=self._span_attributes,
+                )
+            except Exception:  # pragma: NO COVER
+                span_context_manager = contextlib.nullcontext()
+
+        with span_context_manager as span:
+            try:
+                return wrapped_func(*args, **kwargs)
+            except Exception as exc:
+                if span is not None and hasattr(span, "record_exception"):
+                    from opentelemetry import trace
+
+                    span.record_exception(exc)
+                    span.set_status(trace.StatusCode.ERROR, str(exc))
+                raise
 
 
 def wrap_method(
@@ -197,6 +296,10 @@ def wrap_method(
     client_info=client_info.DEFAULT_CLIENT_INFO,
     *,
     with_call=False,
+    method_name=None,
+    tracer_provider=None,
+    rpc_system="grpc",
+    is_streaming=False,
 ):
     """Wrap an RPC method with common behavior.
 
@@ -280,6 +383,13 @@ def get_topic(name, timeout=None):
             return a tuple of (response, grpc.Call) instead of just the response.
             This is useful for extracting trailing metadata from unary calls.
             Defaults to False.
+        method_name (Optional[str]): Optional explicit full RPC method name
+            (e.g. "/google.cloud.secretmanager.v1.SecretManagerService/AccessSecretVersion").
+        tracer_provider (Optional[Any]): Optional custom OpenTelemetry TracerProvider
+            to obtain the tracer from.
+        rpc_system (Optional[str]): The RPC system (defaults to "grpc"). If not "grpc",
+            tracing will not be enabled.
+        is_streaming (bool): Whether the callable is a streaming RPC. Defaults to False.
 
     Returns:
         Callable: A new callable that takes optional ``retry``, ``timeout``,
@@ -307,5 +417,9 @@ def get_topic(name, timeout=None):
             default_timeout,
             default_compression,
             metadata=user_agent_metadata,
+            method_name=method_name,
+            tracer_provider=tracer_provider,
+            rpc_system=rpc_system,
+            is_streaming=is_streaming,
         )
     )

packages/google-api-core/google/api_core/gapic_v1/method_async.py

@@ -37,6 +37,10 @@ def wrap_method(
     default_compression=None,
     client_info=client_info.DEFAULT_CLIENT_INFO,
     kind=_DEFAULT_ASYNC_TRANSPORT_KIND,
+    method_name=None,
+    tracer_provider=None,
+    rpc_system="grpc",
+    is_streaming=False,
 ):
     """Wrap an async RPC method with common behavior.
 
@@ -57,5 +61,9 @@ def wrap_method(
             default_timeout,
             default_compression,
             metadata=metadata,
+            method_name=method_name,
+            tracer_provider=tracer_provider,
+            rpc_system=rpc_system,
+            is_streaming=is_streaming,
         )
     )

packages/google-api-core/tests/asyncio/gapic/test_method_async.py

@@ -274,3 +274,50 @@ async def test_wrap_method_without_wrap_errors():
         await wrapped_method()
 
         method.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_wrap_method_async_with_otel_tracing(monkeypatch):
+    import sys
+
+    monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
+    fake_call = grpc_helpers_async.FakeUnaryUnaryCall(42)
+    method = mock.Mock(spec=aio.UnaryUnaryMultiCallable, return_value=fake_call)
+
+    mock_span = mock.MagicMock()
+    mock_tracer = mock.MagicMock()
+    mock_tracer.start_as_current_span.return_value.__enter__.return_value = mock_span
+
+    mock_trace = mock.Mock()
+    mock_trace.get_tracer.return_value = mock_tracer
+    mock_trace.SpanKind.CLIENT = "CLIENT"
+
+    with (
+        mock.patch(
+            "google.api_core._observability.is_otel_capabilities_enabled",
+            return_value=True,
+        ),
+        mock.patch.dict(
+            sys.modules,
+            {
+                "opentelemetry": mock.Mock(trace=mock_trace),
+                "opentelemetry.trace": mock_trace,
+            },
+        ),
+    ):
+        wrapped_method = gapic_v1.method_async.wrap_method(
+            method,
+            method_name="google.test.AsyncService/AsyncMethod",
+        )
+        result = await wrapped_method(1, 2, meep="moop")
+
+    assert result == 42
+    mock_tracer.start_as_current_span.assert_called_once_with(
+        "google.test.AsyncService/AsyncMethod",
+        kind="CLIENT",
+        attributes={
+            "rpc.system": "grpc",
+            "rpc.service": "google.test.AsyncService",
+            "rpc.method": "AsyncMethod",
+        },
+    )