feat(observability): implement universal 4-path OpenTelemetry tracing

packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/_compat.py.j2

@@ -5,12 +5,12 @@
 """A compatibility module for older versions of google-api-core."""
 {% set has_auto_populated_fields = api.all_method_settings.values()|map(attribute="auto_populated_fields", default=[])|select|list %}
 {# TODO(https://github.com/googleapis/google-cloud-python/issues/17813):
-Clean up this file/functions when the minimum supported version of 
+Clean up this file/functions when the minimum supported version of
 google-api-core has the functions in `_compat.py.j2`. #}
-{# TODO(https://github.com/googleapis/google-cloud-python/issues/17884): 
-Add conditional logic to check if static code exists in google-api-core and use it from there, 
+{# TODO(https://github.com/googleapis/google-cloud-python/issues/17884):
+Add conditional logic to check if static code exists in google-api-core and use it from there,
 falling back to the local implementation if not present. #}
-{# TODO(https://github.com/googleapis/google-cloud-python/issues/17883): 
+{# TODO(https://github.com/googleapis/google-cloud-python/issues/17883):
 Backfill compatibility functions being removed from the client layer. #}
 
 import os
@@ -20,23 +20,182 @@ import uuid
 import google.protobuf.message
 {% endif %}
 
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Sequence, Tuple
 {% if has_auto_populated_fields %}
 from typing import TYPE_CHECKING, Union
-
-if TYPE_CHECKING:  # pragma: NO COVER
-    import proto  # type: ignore[import-untyped]
 {% endif %}
 
+from google.api_core import gapic_v1
 from google.api_core import path_template
 from google.api_core.universe import EmptyUniverseError
 from google.auth.exceptions import MutualTLSChannelError
 from google.protobuf import json_format
 from urllib.parse import urlparse, urlunparse
 
+try:
+    import grpc  # type: ignore[import-untyped]
+except ImportError:  # pragma: NO COVER
+    grpc = None  # type: ignore[assignment]
+
+{% if has_auto_populated_fields %}
+if TYPE_CHECKING:  # pragma: NO COVER
+    import proto  # type: ignore[import-untyped]
+{% endif %}
+
+# The _observability module and OpenTelemetry tracing in wrap_method were introduced
+# together in google-api-core 2.36.0+.
+# On older versions of google-api-core or when type-checking against them,
+# mypy may flag attr-defined or assignment errors when fallback to None occurs.
+try:
+    from google.api_core import _observability  # type: ignore[attr-defined]
+except ImportError:  # pragma: NO COVER
+    _observability = None  # type: ignore[assignment]
+
+# OpenTelemetry method tracing arguments (client_options, method_name, is_streaming, kind)
+# were introduced to gapic_v1.method.wrap_method and gapic_v1.method_async.wrap_method
+# in google-api-core 2.36.0+ alongside the _observability module.
+# Once the repository's minimum constraint for google-api-core is bumped to >= 2.36.0,
+# WRAP_METHOD_SUPPORTS_TRACING, the fallback _FallbackTraceContext, and base transport fallbacks
+# can be safely removed.
+if _observability is not None and hasattr(_observability, "trace_http_request"):
+    trace_http_request = _observability.trace_http_request
+    WRAP_METHOD_SUPPORTS_TRACING = True
+else:  # pragma: NO COVER
+    WRAP_METHOD_SUPPORTS_TRACING = False
+
+    # Fallback for older versions of google-api-core without HTTP tracing.
+    class _FallbackTraceContext:
+        def __enter__(self) -> "_FallbackTraceContext":
+            return self
+
+        def __exit__(self, *args: Any) -> None:
+            pass
+
+        def record_response(self, response: Any) -> None:
+            pass
+
+        record_http_response = record_response
+
+        def record_error(self, exc: BaseException | None) -> None:
+            pass
+
+        record_http_error = record_error
+
+    def trace_http_request(*args: Any, **kwargs: Any) -> _FallbackTraceContext:
+        return _FallbackTraceContext()
+
+# The `kind` parameter in gapic_v1.method_async.wrap_method was introduced in
+# google-api-core 2.29.0 (PR #688) alongside _DEFAULT_ASYNC_TRANSPORT_KIND to prevent
+# async REST callables from being wrapped with gRPC error handlers.
+# In environments running older google-api-core (< 2.29.0) where tracing is also absent,
+# passing `kind` will trigger a TypeError unless stripped.
+# Once the minimum constraint for google-api-core is bumped to
+# >= 2.29.0, this check and the corresponding fallback in base transports can be safely removed.
+ASYNC_WRAP_METHOD_SUPPORTS_KIND = WRAP_METHOD_SUPPORTS_TRACING or hasattr(
+    gapic_v1.method_async, "_DEFAULT_ASYNC_TRANSPORT_KIND"
+)
+
+try:
+    from google.api_core import grpc_helpers
+except ImportError:  # pragma: NO COVER
+    grpc_helpers = None  # type: ignore[assignment]
+
+try:
+    from google.api_core import grpc_helpers_async
+except ImportError:  # pragma: NO COVER
+    grpc_helpers_async = None  # type: ignore[assignment]
+
+
+def apply_channel_interceptors(
+    channel: Any, interceptors: Optional[Sequence[Any]] = None
+) -> Any:
+    """Applies interceptors to a synchronous gRPC channel with fallback for older google-api-core."""
+    if grpc_helpers is not None and hasattr(grpc_helpers, "apply_channel_interceptors"):
+        return grpc_helpers.apply_channel_interceptors(channel, interceptors)  # pragma: NO COVER
+    if not interceptors:  # pragma: NO COVER
+        return channel
+    for interceptor in reversed(list(interceptors)):  # pragma: NO COVER
+        if callable(interceptor):
+            channel = interceptor(channel)
+        elif grpc is not None:
+            channel = grpc.intercept_channel(channel, interceptor)
+    return channel  # pragma: NO COVER
+
+
+def _fallback_apply_interceptors(
+    channel: Any, interceptors: Optional[Sequence[Any]] = None
+) -> Any:  # pragma: NO COVER
+    """Fallback helper to attach interceptors to an existing async gRPC channel.
+
+    Synchronous gRPC channels allow post-creation wrapping via
+    `grpc.intercept_channel(channel, *interceptors)`. However, asynchronous
+    channels (`grpc.aio.Channel`) in grpcio are immutable post-creation and
+    normally require interceptors at initialization.
+
+    When generated client libraries run against older versions of `google-api-core`
+    (< 2.36.0) where `grpc_helpers_async.apply_channel_interceptors` is not yet
+    available, this fallback mutates the channel's internal interceptor lists
+    directly so telemetry and auth interceptors remain operational.
+
+    Args:
+        channel (Any): The gRPC channel instance (typically a `grpc.aio.Channel`
+            or mock).
+        interceptors (Optional[Sequence[Any]]): A sequence of gRPC interceptor
+            instances or callables to register on the channel.
+
+    Returns:
+        Any: The channel instance with interceptors registered.
+    """
+    mapping = (
+        ("intercept_unary_unary", "_unary_unary_interceptors"),
+        ("intercept_unary_stream", "_unary_stream_interceptors"),
+        ("intercept_stream_unary", "_stream_unary_interceptors"),
+        ("intercept_stream_stream", "_stream_stream_interceptors"),
+    )
+    if interceptors:
+        for interceptor in interceptors:
+            matched = False
+
+            # Path 1: Interface-matched routing.
+            # Inspect whether the interceptor implements specific gRPC method hooks
+            # (unary-unary, unary-stream, stream-unary, stream-stream). If the channel
+            # maintains a corresponding internal list, append the interceptor (preventing duplicates).
+            # A single interceptor can handle multiple call patterns, so check all mappings.
+            for method_name, attr_name in mapping:
+                if hasattr(interceptor, method_name) and hasattr(channel, attr_name):
+                    target_list = getattr(channel, attr_name)
+                    if isinstance(target_list, list) and interceptor not in target_list:
+                        target_list.append(interceptor)
+                    matched = True
+
+            # Path 2: Duck-typed / callable fallback.
+            # If the interceptor did not implement any of the specific method signatures
+            # above (e.g., generic callables, untyped wrappers, or test doubles), fall back
+            # to registering it in the primary unary-unary interceptor list if present.
+            if not matched and hasattr(channel, "_unary_unary_interceptors"):
+                unary_interceptors = channel._unary_unary_interceptors
+                if isinstance(unary_interceptors, list) and interceptor not in unary_interceptors:
+                    unary_interceptors.append(interceptor)
+    return channel
+
+
+def apply_async_channel_interceptors(
+    channel: Any, interceptors: Optional[Sequence[Any]] = None
+) -> Any:
+    """Applies interceptors to an asynchronous gRPC channel with fallback for older google-api-core."""
+    if grpc_helpers_async is not None and hasattr(
+        grpc_helpers_async, "apply_channel_interceptors"
+    ):
+        return grpc_helpers_async.apply_channel_interceptors(channel, interceptors)  # pragma: NO COVER
+    return _fallback_apply_interceptors(channel, interceptors)  # pragma: NO COVER
+
+
+apply_channel_interceptors_async = apply_async_channel_interceptors
+
+
 try:
     # note: `#type: ignore` is added because the return type for `should_use_client_cert`
-    # is different than that of the fallback implementation below. This will be removed once 
+    # is different than that of the fallback implementation below. This will be removed once
     # we bump the minimum supported version of google-auth.
     from google.auth.transport.mtls import should_use_client_cert  # type: ignore
 except ImportError:  # pragma: NO COVER

packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/_shared_macros.j2

@@ -159,30 +159,45 @@ def _get_http_options():
     session,
     timeout,
     transcoded_request,
-    body=None):
-
+    body=None,
+    client_options=None,
+    url_template=None):
+    """Execute the HTTP request over the transport session with
+    OpenTelemetry tracing and metadata propagation."""
+
     uri = transcoded_request['uri']
     method = transcoded_request['method']
     headers = dict(metadata)
     headers['Content-Type'] = 'application/json'
-    response = {{ await_prefix }}getattr(session, method)(
-        "{host}{uri}".format(host=host, uri=uri),
-        timeout=timeout,
+    url = "{host}{uri}".format(host=host, uri=uri)
+
+    with trace_http_request(
+        client_options=client_options,
+        method=method,
+        url=url,
+        url_template=url_template,
         headers=headers,
-        params=rest_helpers.flatten_query_params(query_params, strict=True),
-        {% if body_spec %}
-        data=body,
-        {% endif %}
-        {% if not is_async and is_streaming_method %}
-        {# NOTE: The underlying `requests` library used for making a sync request
-         # requires us to set `stream=True` to avoid loading the entire response
-         # into memory at once. For an async request, given its nature where it
-         # reads data chunk by chunk, this is not required.
-        #}
-        stream=True,
-        {% endif %}
-        )
-    return response
+        body=body,
+    ) as trace_ctx:
+        response = {{ await_prefix }}getattr(session, method)(
+            url,
+            timeout=timeout,
+            headers=headers,
+            params=rest_helpers.flatten_query_params(query_params, strict=True),
+            {% if body_spec %}
+            data=body,
+            {% endif %}
+            {% if not is_async and is_streaming_method %}
+            {# NOTE: The underlying `requests` library used for making a sync request
+             # requires us to set `stream=True` to avoid loading the entire response
+             # into memory at once. For an async request, given its nature where it
+             # reads data chunk by chunk, this is not required.
+            #}
+            stream=True,
+            {% endif %}
+            )
+        trace_ctx.record_response(response)
+        return response
 {% endmacro %}
 
 {# rest_call_method_common includes the common code for a rest __call__ method to be
@@ -240,13 +255,25 @@ def _get_http_options():
         )
 
     # Send the request
-    response = {{ await_prefix }}{{ async_class_prefix }}{{ service_name }}RestTransport._{{method_name}}._get_response(self._host, metadata, query_params, self._session, timeout, transcoded_request{% if body_spec %}, body{% endif %})
+    response = {{ await_prefix }}{{ async_class_prefix }}{{ service_name }}RestTransport._{{method_name}}._get_response(
+        self._host,
+        metadata,
+        query_params,
+        self._session,
+        timeout,
+        transcoded_request,
+        {% if body_spec %}
+        body,
+        {% endif %}
+        client_options=getattr(self, "_client_options", None),
+        url_template=http_options[0].get("uri") if http_options else None,
+    )
 
     # In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
     # subclass.
     if response.status_code >= 400:
-        {# Note: format_http_response_error takes in more parameters than from_http_response and the 
-        latter only supports a response of type requests.Response. 
+        {# Note: format_http_response_error takes in more parameters than from_http_response and the
+        latter only supports a response of type requests.Response.
         TODO: Clean up the sync response error handling and use format_http_response_error.
         See issue: https://github.com/googleapis/gapic-generator-python/issues/2116. #}
         {% if is_async %}
@@ -364,10 +391,13 @@ return {{ client_output_ident }}(
 {% macro prep_wrapped_messages_async_method(api, service, is_rest_asyncio=False) %}
 {% set rest_async_io_enabled = api.all_library_settings[api.naming.proto_package].python_settings.experimental_features.rest_async_io_enabled %}
 def _prep_wrapped_messages(self, client_info):
-    """ Precompute the wrapped methods, overriding the base class method to use async wrappers."""
+    """Precompute and cache wrapped methods for async RPC dispatch.
+
+    Overrides the base class method to use asynchronous wrappers and retries.
+    """
     self._wrapped_methods = {
         {% for method in service.methods.values() if not is_rest_asyncio or rest_async_io_enabled or method.is_resumable_upload %}
-        self.{{ method.transport_safe_name|snake_case }}: self._wrap_method(
+        self.{{ method.transport_safe_name|snake_case }}: self._wrap_async_method(
             self.{{ method.transport_safe_name|snake_case }},
             {% if method.retry %}
             default_retry=retries.AsyncRetry(
@@ -390,17 +420,22 @@ def _prep_wrapped_messages(self, client_info):
             {% endif %}
             default_timeout={{ method.timeout }},
             client_info=client_info,
+            method_name="{{ '.'.join(method.meta.address.package) }}.{{ service.name }}/{{ method.name }}",
+            {% if method.client_streaming or method.server_streaming %}
+            is_streaming=True,
+            {% endif %}
         ),
         {% endfor %}{# service.methods.values() #}
         {% if not is_rest_asyncio or rest_async_io_enabled %}
         {% for method_name in api.mixin_api_methods.keys() %}
         {# TODO(https://github.com/googleapis/gapic-generator-python/issues/2197): Use `transport_safe_name` similar
          # to what we do for non-mixin methods above.
         #}
-        self.{{ method_name|snake_case }}: self._wrap_method(
+        self.{{ method_name|snake_case }}: self._wrap_async_method(
             self.{{ method_name|snake_case }},
             default_timeout=None,
             client_info=client_info,
+            method_name="{{ api.mixin_api_signatures[method_name].rpc_name }}",
         ),
         {% endfor %}{# method_name in api.mixin_api_methods.keys() #}
         {% endif %}
@@ -503,11 +538,11 @@ class {{ async_method_name_prefix }}{{ service.name }}RestInterceptor:
 
         Override in a subclass to read or manipulate the response or metadata after it
         is returned by the {{ service.name }} server but before it is returned to user code.
-        
+
         We recommend only using this `post_{{ method.name|snake_case }}_with_metadata`
         interceptor in new development instead of the `post_{{ method.name|snake_case }}` interceptor.
         When both interceptors are used, this `post_{{ method.name|snake_case }}_with_metadata` interceptor runs after the
-        `post_{{ method.name|snake_case }}` interceptor. The (possibly modified) response returned by 
+        `post_{{ method.name|snake_case }}` interceptor. The (possibly modified) response returned by
         `post_{{ method.name|snake_case }}` will be passed to
         `post_{{ method.name|snake_case }}_with_metadata`.
         """
@@ -555,15 +590,15 @@ class {{ async_method_name_prefix }}{{ service.name }}RestInterceptor:
 
 @property
 def {{ name|snake_case }}(self):
-    return self.{{ name|make_private }}(self._session, self._host, self._interceptor) # type: ignore
+    return self.{{ name|make_private }}(self._session, self._host, self._interceptor, getattr(self, "_client_options", None)) # type: ignore
 
 class {{ name|make_private }}(_Base{{ service.name }}RestTransport._Base{{name}}, {{ async_method_name_prefix }}{{service.name}}RestStub):
     def __hash__(self):
         return hash("{{ async_method_name_prefix }}{{ service.name }}RestTransport.{{ name }}")
 
     {% set body_spec = api.mixin_http_options["{}".format(name)][0].body %}
     {{ response_method(body_spec, is_async=is_async, is_streaming_method=None) | indent(4) }}
-    
+
     {{ async_prefix }}def __call__(self,
         request: {{ sig.request_type }}, *,
         retry: OptionalRetry=gapic_v1.method.DEFAULT,
@@ -589,7 +624,7 @@ class {{ name|make_private }}(_Base{{ service.name }}RestTransport._Base{{name}}
         {% endif %}
         """
         {{ rest_call_method_common(body_spec, name, service, is_async)|indent(4) }}
-        
+
         {% if sig.response_type == "None" %}
         return {{ await_prefix }}self._interceptor.post_{{ name|snake_case }}(None)
         {% else %}
@@ -611,7 +646,7 @@ class {{ name|make_private }}(_Base{{ service.name }}RestTransport._Base{{name}}
                 "payload": response_payload,
                 "headers":  dict(response.headers),
                 "status": response.status_code,
-            }        
+            }
             _LOGGER.debug(
                 "Received response for {{ service.meta.address.proto_package_versioned }}.{{ service.async_client_name }}.{{ name }}",
                 extra = {
@@ -633,7 +668,7 @@ class {{ name|make_private }}(_Base{{ service.name }}RestTransport._Base{{name}}
 
 {% macro client_method_metadata_default_value() %}(){% endmacro %}
 
-{% macro client_method_metadata_argument_doc() %}metadata ({{ client_method_metadata_type() }}): Key/value pairs which should be 
+{% macro client_method_metadata_argument_doc() %}metadata ({{ client_method_metadata_type() }}): Key/value pairs which should be
             sent along with the request as metadata. Normally, each value must be of type `str`,
             but for metadata keys ending with the suffix `-bin`, the corresponding values must
             be of type `bytes`.{% endmacro %}