feat(gapic): add async gRPC OpenTelemetry channel tracing

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

@@ -554,6 +554,18 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
                 is not None
             ):
                 interceptors.append(otel_interceptor)
+            elif (
+                isinstance(transport_init, type)
+                and issubclass(transport_init, {{ service.grpc_asyncio_transport_name }})
+                and _observability is not None
+                and (
+                    otel_async_interceptors := _observability.get_otel_async_interceptor(
+                        self._client_options
+                    )
+                )
+                is not None
+            ):
+                interceptors.extend(otel_async_interceptors)
             {% endif %}
 
             # initialize with the provided callable or the passed in class

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

@@ -171,6 +171,7 @@ class {{ service.grpc_asyncio_transport_name }}({{ 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[aio.ClientInterceptor]] = None,
             ) -> None:
         """Instantiate the transport.
 
@@ -222,6 +223,8 @@ class {{ service.grpc_asyncio_transport_name }}({{ 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[aio.ClientInterceptor]]):
+                Additional interceptors to apply to the gRPC channel.
 
         Raises:
             google.auth.exceptions.MutualTlsChannelError: If mutual TLS transport
@@ -300,6 +303,13 @@ class {{ service.grpc_asyncio_transport_name }}({{ service.name }}Transport):
                 ],
             )
 
+        apply_interceptors = getattr(
+            grpc_helpers_async,
+            "apply_channel_interceptors",
+            lambda channel, interceptors: channel,
+        )
+        self._grpc_channel = apply_interceptors(self._grpc_channel, interceptors)
+
         self._interceptor = _LoggingClientAIOInterceptor()
         self._grpc_channel._unary_unary_interceptors.append(self._interceptor)
         self._logged_channel = self._grpc_channel

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

@@ -992,6 +992,96 @@ def test_{{ service.name|snake_case }}_grpc_transport_custom_channel_interceptor
         assert transport.grpc_channel == mock_custom_channel
 
 
+def test_{{ service.async_client_name|snake_case }}_otel_channel_injection_enabled():
+    mock_interceptor = mock.Mock()
+    mock_obs = mock.Mock()
+    mock_obs.get_otel_async_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_asyncio_transport_name }}, "__init__", return_value=None
+        ) as patched_transport_init,
+    ):
+        client = {{ service.async_client_name }}()
+
+        mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options)
+        called_kwargs = patched_transport_init.call_args.kwargs
+        assert called_kwargs.get("interceptors") == [mock_interceptor]
+
+
+def test_{{ service.async_client_name|snake_case }}_otel_channel_injection_disabled():
+    mock_obs = mock.Mock()
+    mock_obs.get_otel_async_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_asyncio_transport_name }}, "__init__", return_value=None
+        ) as patched_transport_init,
+    ):
+        client = {{ service.async_client_name }}()
+
+        mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options)
+        called_kwargs = patched_transport_init.call_args.kwargs
+        assert not called_kwargs.get("interceptors", [])
+
+
+def test_{{ service.name|snake_case }}_grpc_asyncio_transport_channel_interceptors():
+    mock_interceptor = mock.Mock()
+    mock_channel = mock.Mock()
+
+    with (
+        mock.patch.object(
+            transports.{{ service.grpc_asyncio_transport_name }},
+            "create_channel",
+            return_value=mock_channel,
+        ),
+        mock.patch.object(
+            grpc_helpers_async,
+            "apply_channel_interceptors",
+            return_value=mock_channel,
+            create=True,
+        ) as mock_apply_interceptors,
+    ):
+        transport = transports.{{ service.grpc_asyncio_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_asyncio_transport_custom_channel_interceptors():
+    mock_interceptor = mock.Mock()
+    mock_custom_channel = mock.Mock(spec=aio.Channel)
+    mock_custom_channel._unary_unary_interceptors = []
+
+    with mock.patch.object(
+        grpc_helpers_async,
+        "apply_channel_interceptors",
+        return_value=mock_custom_channel,
+        create=True,
+    ) as mock_apply_interceptors:
+        transport = transports.{{ service.grpc_asyncio_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

@@ -412,6 +412,9 @@ def showcase(
             "opentelemetry-sdk",
             "opentelemetry-instrumentation-grpc",
         )
+        local_core = Path(__file__).parent.parent / "google-api-core"
+        if local_core.exists() and (local_core / "setup.py").exists():
+            session.install("-e", str(local_core))
         test_directory = Path("tests", "system")
         ignore_file = env.get("IGNORE_FILE")
         pytest_command = [

packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/client.py

@@ -567,6 +567,18 @@ def __init__(self, *,
                 is not None
             ):
                 interceptors.append(otel_interceptor)
+            elif (
+                isinstance(transport_init, type)
+                and issubclass(transport_init, AssetServiceGrpcAsyncIOTransport)
+                and _observability is not None
+                and (
+                    otel_async_interceptors := _observability.get_otel_async_interceptor(
+                        self._client_options
+                    )
+                )
+                is not None
+            ):
+                interceptors.extend(otel_async_interceptors)
 
             # initialize with the provided callable or the passed in class
             transport_kwargs = {

packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/transports/grpc_asyncio.py

@@ -179,6 +179,7 @@ def __init__(self, *,
             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[aio.ClientInterceptor]] = None,
             ) -> None:
         """Instantiate the transport.
 
@@ -230,6 +231,8 @@ def __init__(self, *,
                 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[aio.ClientInterceptor]]):
+                Additional interceptors to apply to the gRPC channel.
 
         Raises:
             google.auth.exceptions.MutualTlsChannelError: If mutual TLS transport
@@ -306,6 +309,13 @@ def __init__(self, *,
                 ],
             )
 
+        apply_interceptors = getattr(
+            grpc_helpers_async,
+            "apply_channel_interceptors",
+            lambda channel, interceptors: channel,
+        )
+        self._grpc_channel = apply_interceptors(self._grpc_channel, interceptors)
+
         self._interceptor = _LoggingClientAIOInterceptor()
         self._grpc_channel._unary_unary_interceptors.append(self._interceptor)
         self._logged_channel = self._grpc_channel

packages/gapic-generator/tests/integration/goldens/asset/tests/unit/gapic/asset_v1/test_asset_service.py

@@ -848,6 +848,95 @@ def test_asset_service_grpc_transport_custom_channel_interceptors():
         assert transport.grpc_channel == mock_custom_channel
 
 
+def test_asset_service_async_client_otel_channel_injection_enabled():
+    mock_interceptor = mock.Mock()
+    mock_obs = mock.Mock()
+    mock_obs.get_otel_async_interceptor.return_value = [mock_interceptor]
+    with (
+        mock.patch(
+            "google.cloud.asset_v1.services.asset_service.client._observability",
+            mock_obs,
+        ),
+        mock.patch.object(
+            transports.AssetServiceGrpcAsyncIOTransport, "__init__", return_value=None
+        ) as patched_transport_init,
+    ):
+        client = AssetServiceAsyncClient()
+
+        mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options)
+        called_kwargs = patched_transport_init.call_args.kwargs
+        assert called_kwargs.get("interceptors") == [mock_interceptor]
+
+
+def test_asset_service_async_client_otel_channel_injection_disabled():
+    mock_obs = mock.Mock()
+    mock_obs.get_otel_async_interceptor.return_value = None
+    with (
+        mock.patch(
+            "google.cloud.asset_v1.services.asset_service.client._observability",
+            mock_obs,
+        ),
+        mock.patch.object(
+            transports.AssetServiceGrpcAsyncIOTransport, "__init__", return_value=None
+        ) as patched_transport_init,
+    ):
+        client = AssetServiceAsyncClient()
+
+        mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options)
+        called_kwargs = patched_transport_init.call_args.kwargs
+        assert not called_kwargs.get("interceptors", [])
+
+
+def test_asset_service_grpc_asyncio_transport_channel_interceptors():
+    mock_interceptor = mock.Mock()
+    mock_channel = mock.Mock()
+
+    with (
+        mock.patch.object(
+            transports.AssetServiceGrpcAsyncIOTransport,
+            "create_channel",
+            return_value=mock_channel,
+        ),
+        mock.patch.object(
+            grpc_helpers_async,
+            "apply_channel_interceptors",
+            return_value=mock_channel,
+            create=True,
+        ) as mock_apply_interceptors,
+    ):
+        transport = transports.AssetServiceGrpcAsyncIOTransport(
+            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_asset_service_grpc_asyncio_transport_custom_channel_interceptors():
+    mock_interceptor = mock.Mock()
+    mock_custom_channel = mock.Mock(spec=aio.Channel)
+    mock_custom_channel._unary_unary_interceptors = []
+
+    with mock.patch.object(
+        grpc_helpers_async,
+        "apply_channel_interceptors",
+        return_value=mock_custom_channel,
+        create=True,
+    ) as mock_apply_interceptors:
+        transport = transports.AssetServiceGrpcAsyncIOTransport(
+            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", [
     (AssetServiceClient, transports.AssetServiceGrpcTransport, "grpc", grpc_helpers),
     (AssetServiceAsyncClient, transports.AssetServiceGrpcAsyncIOTransport, "grpc_asyncio", grpc_helpers_async),

packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/client.py

@@ -504,6 +504,18 @@ def __init__(self, *,
                 is not None
             ):
                 interceptors.append(otel_interceptor)
+            elif (
+                isinstance(transport_init, type)
+                and issubclass(transport_init, IAMCredentialsGrpcAsyncIOTransport)
+                and _observability is not None
+                and (
+                    otel_async_interceptors := _observability.get_otel_async_interceptor(
+                        self._client_options
+                    )
+                )
+                is not None
+            ):
+                interceptors.extend(otel_async_interceptors)
 
             # initialize with the provided callable or the passed in class
             transport_kwargs = {

packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/transports/grpc_asyncio.py

@@ -185,6 +185,7 @@ def __init__(self, *,
             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[aio.ClientInterceptor]] = None,
             ) -> None:
         """Instantiate the transport.
 
@@ -236,6 +237,8 @@ def __init__(self, *,
                 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[aio.ClientInterceptor]]):
+                Additional interceptors to apply to the gRPC channel.
 
         Raises:
             google.auth.exceptions.MutualTlsChannelError: If mutual TLS transport
@@ -311,6 +314,13 @@ def __init__(self, *,
                 ],
             )
 
+        apply_interceptors = getattr(
+            grpc_helpers_async,
+            "apply_channel_interceptors",
+            lambda channel, interceptors: channel,
+        )
+        self._grpc_channel = apply_interceptors(self._grpc_channel, interceptors)
+
         self._interceptor = _LoggingClientAIOInterceptor()
         self._grpc_channel._unary_unary_interceptors.append(self._interceptor)
         self._logged_channel = self._grpc_channel