feat(observability): suppress redundant gRPC auto-instrumentation spans

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

@@ -81,9 +81,12 @@ else:  # pragma: NO COVER
 
         record_http_error = record_error
 
-    def trace_http_request(*args: Any, **kwargs: Any) -> _FallbackTraceContext:
+    # TODO (Issue #18657): (mypy) remove pragma once google-api-core >= 2.42.0
+    # ignores the fallback function redefinition of trace_http_request
+    def trace_http_request(*args: Any, **kwargs: Any) -> _FallbackTraceContext:  # type: ignore[misc]
         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.
@@ -123,7 +126,9 @@ def apply_channel_interceptors(
 
 
 def _fallback_apply_interceptors(
-    channel: Any, interceptors: Optional[Sequence[Any]] = None
+    channel: Any,
+    interceptors: Optional[Sequence[Any]] = None,
+    prepend: bool = False,
 ) -> Any:  # pragma: NO COVER
     """Fallback helper to attach interceptors to an existing async gRPC channel.
 
@@ -142,6 +147,8 @@ def _fallback_apply_interceptors(
             or mock).
         interceptors (Optional[Sequence[Any]]): A sequence of gRPC interceptor
             instances or callables to register on the channel.
+        prepend (bool): Whether to prepend interceptors to the beginning of the
+            channel's pipeline (e.g. for outer tracing spans) instead of appending.
 
     Returns:
         Any: The channel instance with interceptors registered.
@@ -153,41 +160,69 @@ def _fallback_apply_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)
+        # Path 1: Typed interceptors matching specific gRPC RPC methods.
+        for method_name, attr_name in mapping:
+            target_list = getattr(channel, attr_name, None)
+            if not isinstance(target_list, list):
+                continue
+
+            matching = [
+                i
+                for i in interceptors
+                if hasattr(i, method_name) and i not in target_list
+            ]
+            if matching:
+                if prepend:
+                    target_list[:0] = matching
+                else:
+                    target_list.extend(matching)
+
+        # Path 2: Fallback for generic/untyped interceptors.
+        # Interceptors that do not define any of the 4 standard gRPC method signatures
+        # (such as generic callables, custom wrappers, or test doubles) are routed to
+        # the primary unary-unary interceptor list.
+        unary_list = getattr(channel, "_unary_unary_interceptors", None)
+        if isinstance(unary_list, list):
+            standard_methods = (
+                "intercept_unary_unary",
+                "intercept_unary_stream",
+                "intercept_stream_unary",
+                "intercept_stream_stream",
+            )
+            fallback_items = [
+                i
+                for i in interceptors
+                if not any(hasattr(i, m) for m in standard_methods)
+                and i not in unary_list
+            ]
+            if fallback_items:
+                if prepend:
+                    unary_list[:0] = fallback_items
+                else:
+                    unary_list.extend(fallback_items)
     return channel
 
 
 def apply_async_channel_interceptors(
-    channel: Any, interceptors: Optional[Sequence[Any]] = None
-) -> Any:
+    channel: Any,
+    interceptors: Optional[Sequence[Any]] = None,
+    prepend: bool = False,
+) -> Any:  # pragma: NO COVER
     """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
+        if not prepend:
+            return grpc_helpers_async.apply_channel_interceptors(channel, interceptors)  # pragma: NO COVER
+        try:
+            return grpc_helpers_async.apply_channel_interceptors(  # pragma: NO COVER
+                channel, interceptors, prepend=True  # type: ignore[call-arg]
+            )
+        except TypeError:  # pragma: NO COVER
+            # Fallback for environments running older google-api-core (< 2.43.0)
+            # where grpc_helpers_async.apply_channel_interceptors lacks `prepend`.
+            return _fallback_apply_interceptors(channel, interceptors, prepend=True)
+    return _fallback_apply_interceptors(channel, interceptors, prepend=prepend)  # pragma: NO COVER
 
 
 apply_channel_interceptors_async = apply_async_channel_interceptors

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

@@ -333,21 +333,28 @@ class {{ service.grpc_asyncio_transport_name }}({{ service.name }}Transport):
                 ],
             )
 
+        # 1. User & Logging Interceptors (appended to channel)
         channel_interceptors = list(interceptors) if interceptors else []
         self._interceptor = _LoggingClientAIOInterceptor()
         channel_interceptors.append(self._interceptor)
+        self._grpc_channel = apply_async_channel_interceptors(
+            self._grpc_channel, channel_interceptors
+        )
 
+        # 2. OpenTelemetry Tracing & Suppression Interceptors (prepended to channel)
+        # Must be placed at the front of the channel's pipeline so that our SDK spans
+        # wrap outside and downstream generic auto-instrumentation is suppressed.
         if (
             _observability is not None
             and (otel_interceptors := _observability.get_otel_async_interceptor(self._client_options)) is not None
         ):
             # NOTE: Coverage tool ignores async interceptors in environments running
             # legacy google-api-core (< 2.36.0) where OpenTelemetry is unavailable.
-            # Lifecycle: Can be lifted once lowest constraints require google-api-core >= 2.36.0.
+            # TODO (Issue #18657): Remove once lowest constraints require google-api-core >= 2.36.0.
             otel_list = otel_interceptors if isinstance(otel_interceptors, (list, tuple)) else [otel_interceptors]  # pragma: NO COVER
-            channel_interceptors.extend(otel_list)  # pragma: NO COVER
-
-        self._grpc_channel = apply_async_channel_interceptors(self._grpc_channel, channel_interceptors)
+            self._grpc_channel = apply_async_channel_interceptors(
+                self._grpc_channel, otel_list, prepend=True
+            )  # pragma: NO COVER
 
         self._logged_channel = self._grpc_channel
         # Wrap messages. This must be done after self._logged_channel exists

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

@@ -609,5 +609,7 @@ def test_apply_async_channel_interceptors_compat():
     dummy_channel = mock.Mock()
     result = apply_async_channel_interceptors(dummy_channel, None)
     assert result is dummy_channel
+    result_prepend = apply_async_channel_interceptors(dummy_channel, None, prepend=True)
+    assert result_prepend is dummy_channel
 
 {% endblock %}

packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/_compat.py

@@ -72,9 +72,12 @@ def record_error(self, exc: BaseException | None) -> None:
 
         record_http_error = record_error
 
-    def trace_http_request(*args: Any, **kwargs: Any) -> _FallbackTraceContext:
+    # TODO (Issue #18657): (mypy) remove pragma once google-api-core >= 2.42.0
+    # ignores the fallback function redefinition of trace_http_request
+    def trace_http_request(*args: Any, **kwargs: Any) -> _FallbackTraceContext:  # type: ignore[misc]
         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.
@@ -114,7 +117,9 @@ def apply_channel_interceptors(
 
 
 def _fallback_apply_interceptors(
-    channel: Any, interceptors: Optional[Sequence[Any]] = None
+    channel: Any,
+    interceptors: Optional[Sequence[Any]] = None,
+    prepend: bool = False,
 ) -> Any:  # pragma: NO COVER
     """Fallback helper to attach interceptors to an existing async gRPC channel.
 
@@ -133,6 +138,8 @@ def _fallback_apply_interceptors(
             or mock).
         interceptors (Optional[Sequence[Any]]): A sequence of gRPC interceptor
             instances or callables to register on the channel.
+        prepend (bool): Whether to prepend interceptors to the beginning of the
+            channel's pipeline (e.g. for outer tracing spans) instead of appending.
 
     Returns:
         Any: The channel instance with interceptors registered.
@@ -144,41 +151,69 @@ def _fallback_apply_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)
+        # Path 1: Typed interceptors matching specific gRPC RPC methods.
+        for method_name, attr_name in mapping:
+            target_list = getattr(channel, attr_name, None)
+            if not isinstance(target_list, list):
+                continue
+
+            matching = [
+                i
+                for i in interceptors
+                if hasattr(i, method_name) and i not in target_list
+            ]
+            if matching:
+                if prepend:
+                    target_list[:0] = matching
+                else:
+                    target_list.extend(matching)
+
+        # Path 2: Fallback for generic/untyped interceptors.
+        # Interceptors that do not define any of the 4 standard gRPC method signatures
+        # (such as generic callables, custom wrappers, or test doubles) are routed to
+        # the primary unary-unary interceptor list.
+        unary_list = getattr(channel, "_unary_unary_interceptors", None)
+        if isinstance(unary_list, list):
+            standard_methods = (
+                "intercept_unary_unary",
+                "intercept_unary_stream",
+                "intercept_stream_unary",
+                "intercept_stream_stream",
+            )
+            fallback_items = [
+                i
+                for i in interceptors
+                if not any(hasattr(i, m) for m in standard_methods)
+                and i not in unary_list
+            ]
+            if fallback_items:
+                if prepend:
+                    unary_list[:0] = fallback_items
+                else:
+                    unary_list.extend(fallback_items)
     return channel
 
 
 def apply_async_channel_interceptors(
-    channel: Any, interceptors: Optional[Sequence[Any]] = None
-) -> Any:
+    channel: Any,
+    interceptors: Optional[Sequence[Any]] = None,
+    prepend: bool = False,
+) -> Any:  # pragma: NO COVER
     """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
+        if not prepend:
+            return grpc_helpers_async.apply_channel_interceptors(channel, interceptors)  # pragma: NO COVER
+        try:
+            return grpc_helpers_async.apply_channel_interceptors(  # pragma: NO COVER
+                channel, interceptors, prepend=True  # type: ignore[call-arg]
+            )
+        except TypeError:  # pragma: NO COVER
+            # Fallback for environments running older google-api-core (< 2.43.0)
+            # where grpc_helpers_async.apply_channel_interceptors lacks `prepend`.
+            return _fallback_apply_interceptors(channel, interceptors, prepend=True)
+    return _fallback_apply_interceptors(channel, interceptors, prepend=prepend)  # pragma: NO COVER
 
 
 apply_channel_interceptors_async = apply_async_channel_interceptors

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

@@ -319,21 +319,28 @@ def __init__(self, *,
                 ],
             )
 
+        # 1. User & Logging Interceptors (appended to channel)
         channel_interceptors = list(interceptors) if interceptors else []
         self._interceptor = _LoggingClientAIOInterceptor()
         channel_interceptors.append(self._interceptor)
+        self._grpc_channel = apply_async_channel_interceptors(
+            self._grpc_channel, channel_interceptors
+        )
 
+        # 2. OpenTelemetry Tracing & Suppression Interceptors (prepended to channel)
+        # Must be placed at the front of the channel's pipeline so that our SDK spans
+        # wrap outside and downstream generic auto-instrumentation is suppressed.
         if (
             _observability is not None
             and (otel_interceptors := _observability.get_otel_async_interceptor(self._client_options)) is not None
         ):
             # NOTE: Coverage tool ignores async interceptors in environments running
             # legacy google-api-core (< 2.36.0) where OpenTelemetry is unavailable.
-            # Lifecycle: Can be lifted once lowest constraints require google-api-core >= 2.36.0.
+            # TODO (Issue #18657): Remove once lowest constraints require google-api-core >= 2.36.0.
             otel_list = otel_interceptors if isinstance(otel_interceptors, (list, tuple)) else [otel_interceptors]  # pragma: NO COVER
-            channel_interceptors.extend(otel_list)  # pragma: NO COVER
-
-        self._grpc_channel = apply_async_channel_interceptors(self._grpc_channel, channel_interceptors)
+            self._grpc_channel = apply_async_channel_interceptors(
+                self._grpc_channel, otel_list, prepend=True
+            )  # pragma: NO COVER
 
         self._logged_channel = self._grpc_channel
         # Wrap messages. This must be done after self._logged_channel exists

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

@@ -502,3 +502,5 @@ def test_apply_async_channel_interceptors_compat():
     dummy_channel = mock.Mock()
     result = apply_async_channel_interceptors(dummy_channel, None)
     assert result is dummy_channel
+    result_prepend = apply_async_channel_interceptors(dummy_channel, None, prepend=True)
+    assert result_prepend is dummy_channel