feat(gapic): add OpenTelemetry channel tracing to generator templates

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

@@ -53,6 +53,13 @@ try:
 except ImportError:  # pragma: NO COVER
     CLIENT_LOGGING_SUPPORTED = False
 
+# Optional: OpenTelemetry tracing capabilities for grpc channel injection
+# Note: _observability was added in google-api-core 2.36.0+; guard for older versions
+try:
+    from google.api_core import _observability  # type: ignore[attr-defined]
+except ImportError:  # pragma: NO COVER
+    _observability = None  # type: ignore[assignment]
+
 _LOGGER = std_logging.getLogger(__name__)
 
 {% filter sort_lines %}
@@ -314,17 +321,17 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
                 client_cert_source = mtls.default_client_cert_source()
         return client_cert_source
 
-    
+
     def _validate_universe_domain(self):
         """Validates client's and credentials' universe domains are consistent.
-        
+
         Returns:
             bool: True iff the configured universe domain is valid.
 
         Raises:
             ValueError: If the configured universe domain is not valid.
         """
-        
+
         # NOTE (b/349488459): universe validation is disabled until further notice.
         return True
 
@@ -355,21 +362,21 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
     @property
     def api_endpoint(self) -> str:
         """Return the API endpoint used by the client instance.
-        
+
         Returns:
             str: The API endpoint used by the client instance.
         """
         return self._api_endpoint
-    
+
     @property
     def universe_domain(self) -> str:
         """Return the universe domain used by the client instance.
-        
+
         Returns:
             str: The universe domain used by the client instance.
         """
         return self._universe_domain
-    
+
     def __init__(self, *,
             credentials: Optional[ga_credentials.Credentials] = None,
             transport: Optional[Union[str, {{ service.name }}Transport, Callable[..., {{ service.name }}Transport]]] = None,
@@ -397,8 +404,8 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
                 {% endif %}
             client_options (Optional[Union[google.api_core.client_options.ClientOptions, dict]]):
                 Custom options for the client.
-                
-                1. The ``api_endpoint`` property can be used to override the              
+
+                1. The ``api_endpoint`` property can be used to override the
                 default endpoint provided by the client when ``transport`` is
                 not explicitly provided. Only if this property is not set and
                 ``transport`` was not explicitly provided, the endpoint is
@@ -415,7 +422,7 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
                 not provided, the default SSL client certificate will be used if
                 present. If GOOGLE_API_USE_CLIENT_CERTIFICATE is "false" or not
                 set, no client certificate will be used.
-                
+
                 3. The ``universe_domain`` property can be used to override the
                 default "googleapis.com" universe. Note that the ``api_endpoint``
                 property still takes precedence; and ``universe_domain`` is
@@ -473,7 +480,7 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
             self._transport = cast({{ service.name }}Transport, transport)
             self._api_endpoint = self._transport.host
 
-        self._api_endpoint = (self._api_endpoint or 
+        self._api_endpoint = (self._api_endpoint or
             get_api_endpoint(
                 api_override=self._client_options.api_endpoint,
                 universe_domain=self._universe_domain,
@@ -531,19 +538,39 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
                 else cast(Callable[..., {{ service.name }}Transport], transport)
             )
             {% endif %}
+            # When OpenTelemetry tracing is enabled, obtain the channel interceptor
+            # and pass it to the transport.
+            interceptors = []
+            {% if 'grpc' in opts.transport %}
+            if (
+                isinstance(transport_init, type)
+                and issubclass(transport_init, {{ service.grpc_transport_name }})
+                and _observability is not None
+                and (
+                    otel_interceptor := _observability.get_otel_interceptor(
+                        self._client_options
+                    )
+                )
+                is not None
+            ):
+                interceptors.append(otel_interceptor)
+            {% endif %}
+
             # initialize with the provided callable or the passed in class
-            self._transport = transport_init(
-                credentials=credentials,
-                credentials_file=self._client_options.credentials_file,
-                host=self._api_endpoint,
-                scopes=self._client_options.scopes,
-                client_cert_source_for_mtls=self._client_cert_source,
-                quota_project_id=self._client_options.quota_project_id,
-                client_info=client_info,
-                always_use_jwt_access=True,
-                api_audience=self._client_options.api_audience,
-            )
-
+            transport_kwargs = {
+                "credentials": credentials,
+                "credentials_file": self._client_options.credentials_file,
+                "host": self._api_endpoint,
+                "scopes": self._client_options.scopes,
+                "client_cert_source_for_mtls": self._client_cert_source,
+                "quota_project_id": self._client_options.quota_project_id,
+                "client_info": client_info,
+                "always_use_jwt_access": True,
+                "api_audience": self._client_options.api_audience,
+                **({"interceptors": interceptors} if interceptors else {}),
+            }
+            self._transport = transport_init(**transport_kwargs)
+
         if "async" not in str(self._transport):
             if CLIENT_LOGGING_SUPPORTED and _LOGGER.isEnabledFor(std_logging.DEBUG):  # pragma: NO COVER
                 _LOGGER.debug(
@@ -827,7 +854,7 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
             gapic_v1.routing_header.to_grpc_metadata(
                 (("resource", request_pb.resource),)),
         )
-        
+
         # Validate the universe domain.
         self._validate_universe_domain()
 

packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc.py.j2

@@ -8,9 +8,14 @@ import json
 import logging as std_logging
 import pickle
 import warnings
-from typing import Callable, Dict, Optional, Sequence, Tuple, Union
+from typing import Callable, Dict, Optional, Sequence, Tuple, Union, TYPE_CHECKING
 
+import grpc  # type: ignore
 from google.api_core import grpc_helpers
+
+if TYPE_CHECKING:  # pragma: NO COVER
+    # ClientInterceptor was added in google-api-core 2.36.0+; ignore attribute-defined for older api-core versions during type checking
+    from google.api_core.grpc_helpers import ClientInterceptor  # type: ignore[attr-defined]
 {% if service.has_lro %}
 from google.api_core import operations_v1
 {% endif %}
@@ -21,7 +26,6 @@ from google.auth.transport.grpc import SslCredentials  # type: ignore
 from google.protobuf.json_format import MessageToJson
 import google.protobuf.message
 
-import grpc  # type: ignore
 import proto  # type: ignore
 
 {% filter sort_lines %}
@@ -80,7 +84,7 @@ class _LoggingClientInterceptor(grpc.UnaryUnaryClientInterceptor):  # pragma: NO
             grpc_response = {
                 "payload": response_payload,
                 "metadata": metadata,
-                "status": "OK", 
+                "status": "OK",
             }
             _LOGGER.debug(
                 f"Received response for {client_call_details.method}.",
@@ -123,6 +127,14 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
             client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
             always_use_jwt_access: Optional[bool] = False,
             api_audience: Optional[str] = None,
+            interceptors: Optional[
+                Sequence[
+                    Union[
+                        "ClientInterceptor",
+                        Callable[[grpc.Channel], grpc.Channel],
+                    ]
+                ]
+            ] = None,
             ) -> None:
         """Instantiate the transport.
 
@@ -143,7 +155,7 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
                 ignored if a ``channel`` instance is provided.
             channel (Optional[Union[grpc.Channel, Callable[..., grpc.Channel]]]):
                 A ``Channel`` instance through which to make calls, or a Callable
-                that constructs and returns one. If set to None, ``self.create_channel`` 
+                that constructs and returns one. If set to None, ``self.create_channel``
                 is used to create the channel. If a Callable is given, it will be called
                 with the same arguments as used in ``self.create_channel``.
             api_mtls_endpoint (Optional[str]): Deprecated. The mutual TLS endpoint.
@@ -173,6 +185,9 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
                 to the service that will be set when using certain 3rd party
                 authentication flows. Audience is typically a resource identifier.
                 If not set, the host value will be used as a default.
+            interceptors (Optional[Sequence[Union[ClientInterceptor, Callable[[grpc.Channel], grpc.Channel]]]]):
+                Additional interceptors (or callables that apply interceptors) to apply to the
+                gRPC channel.
 
         Raises:
           google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
@@ -252,6 +267,13 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
                 ],
             )
 
+        apply_interceptors = getattr(
+            grpc_helpers,
+            "apply_channel_interceptors",
+            lambda channel, interceptors: channel,
+        )
+        self._grpc_channel = apply_interceptors(self._grpc_channel, interceptors)
+
         self._interceptor = _LoggingClientInterceptor()
         self._logged_channel =  grpc.intercept_channel(self._grpc_channel, self._interceptor)
 

packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2

@@ -904,6 +904,94 @@ def test_{{ service.client_name|snake_case }}_client_options_from_dict():
         )
 
 
+def test_{{ service.client_name|snake_case }}_otel_channel_injection_enabled():
+    mock_interceptor = mock.Mock()
+    mock_obs = mock.Mock()
+    mock_obs.get_otel_interceptor.return_value = mock_interceptor
+    with (
+        mock.patch(
+            "{{ (api.naming.module_namespace + (api.naming.versioned_module_name,) + service.meta.address.subpackage)|join('.') }}.services.{{ service.name|snake_case }}.client._observability",
+            mock_obs,
+        ),
+        mock.patch.object(
+            transports.{{ service.grpc_transport_name }}, "__init__", return_value=None
+        ) as patched_transport_init,
+    ):
+        client = {{ service.client_name }}(transport="grpc")
+
+        mock_obs.get_otel_interceptor.assert_called_once_with(client._client_options)
+        called_kwargs = patched_transport_init.call_args.kwargs
+        assert called_kwargs.get("interceptors") == [mock_interceptor]
+
+
+def test_{{ service.client_name|snake_case }}_otel_channel_injection_disabled():
+    mock_obs = mock.Mock()
+    mock_obs.get_otel_interceptor.return_value = None
+    with (
+        mock.patch(
+            "{{ (api.naming.module_namespace + (api.naming.versioned_module_name,) + service.meta.address.subpackage)|join('.') }}.services.{{ service.name|snake_case }}.client._observability",
+            mock_obs,
+        ),
+        mock.patch.object(
+            transports.{{ service.grpc_transport_name }}, "__init__", return_value=None
+        ) as patched_transport_init,
+    ):
+        client = {{ service.client_name }}(transport="grpc")
+
+        mock_obs.get_otel_interceptor.assert_called_once_with(client._client_options)
+        called_kwargs = patched_transport_init.call_args.kwargs
+        assert not called_kwargs.get("interceptors", [])
+
+
+def test_{{ service.name|snake_case }}_grpc_transport_channel_interceptors():
+    mock_interceptor = mock.Mock()
+    mock_channel = mock.Mock()
+
+    with (
+        mock.patch.object(
+            transports.{{ service.grpc_transport_name }},
+            "create_channel",
+            return_value=mock_channel,
+        ),
+        mock.patch.object(
+            grpc_helpers,
+            "apply_channel_interceptors",
+            return_value=mock_channel,
+            create=True,
+        ) as mock_apply_interceptors,
+    ):
+        transport = transports.{{ service.grpc_transport_name }}(
+            credentials=ga_credentials.AnonymousCredentials(),
+            interceptors=[mock_interceptor],
+        )
+
+        mock_apply_interceptors.assert_called_once_with(
+            mock_channel, [mock_interceptor]
+        )
+        assert transport.grpc_channel == mock_channel
+
+
+def test_{{ service.name|snake_case }}_grpc_transport_custom_channel_interceptors():
+    mock_interceptor = mock.Mock()
+    mock_custom_channel = mock.Mock(spec=grpc.Channel)
+
+    with mock.patch.object(
+        grpc_helpers,
+        "apply_channel_interceptors",
+        return_value=mock_custom_channel,
+        create=True,
+    ) as mock_apply_interceptors:
+        transport = transports.{{ service.grpc_transport_name }}(
+            channel=mock_custom_channel,
+            interceptors=[mock_interceptor],
+        )
+
+        mock_apply_interceptors.assert_called_once_with(
+            mock_custom_channel, [mock_interceptor]
+        )
+        assert transport.grpc_channel == mock_custom_channel
+
+
 @pytest.mark.parametrize("client_class,transport_class,transport_name,grpc_helpers", [
     ({{ service.client_name }}, transports.{{ service.grpc_transport_name }}, "grpc", grpc_helpers),
     ({{ service.async_client_name }}, transports.{{ service.grpc_asyncio_transport_name }}, "grpc_asyncio", grpc_helpers_async),

packages/gapic-generator/noxfile.py

@@ -18,18 +18,18 @@
 #   PIP_INDEX_URL=https://pypi.org/simple nox
 
 from __future__ import absolute_import
-from concurrent.futures import ThreadPoolExecutor
-from pathlib import Path
+
 import os
+import shutil
 import sys
 import tempfile
 import typing
-import nox  # type: ignore
-
+from concurrent.futures import ThreadPoolExecutor
 from contextlib import contextmanager
 from os import path
-import shutil
+from pathlib import Path
 
+import nox  # type: ignore
 
 nox.options.error_on_missing_interpreters = True
 
@@ -407,6 +407,11 @@ def showcase(
         # Use pytest-asyncio<1.0.0 while we investigate the recent failure described in
         # https://github.com/googleapis/gapic-generator-python/issues/2399
         session.install("pytest", "pytest-asyncio<1.0.0")
+        session.install(
+            "opentelemetry-api",
+            "opentelemetry-sdk",
+            "opentelemetry-instrumentation-grpc",
+        )
         test_directory = Path("tests", "system")
         ignore_file = env.get("IGNORE_FILE")
         pytest_command = [
@@ -498,7 +503,13 @@ def showcase_pqc(
     with showcase_library(session, templates=templates, other_opts=other_opts):
         session.install("pytest", "pytest-asyncio")
         session.install("--upgrade", "grpcio>=1.83.0", "grpcio-status>=1.83.0")
-        session.run("py.test", "--quiet", "--tls", *(session.posargs or ["tests/system/test_pqc.py"]), env=env)
+        session.run(
+            "py.test",
+            "--quiet",
+            "--tls",
+            *(session.posargs or ["tests/system/test_pqc.py"]),
+            env=env,
+        )
 
 
 def run_showcase_unit_tests(session, fail_under=100, rest_async_io_enabled=False):
@@ -508,6 +519,8 @@ def run_showcase_unit_tests(session, fail_under=100, rest_async_io_enabled=False
         "pytest-cov",
         "pytest-xdist",
         "pytest-asyncio",
+        "opentelemetry-api",
+        "opentelemetry-sdk",
     )
     # Freeze and print python environment package versions
     session.run("python", "-m", "pip", "freeze")