opentelemetry-instrumentation-grpc: respect suppressed instrumentation in server interceptors

.changelog/4870.fixed

@@ -0,0 +1 @@
+`opentelemetry-instrumentation-grpc`: respect suppressed instrumentation in server interceptors

instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_aio_server.py

@@ -10,6 +10,7 @@
 except ImportError:
     from wrapt import ObjectProxy as BaseObjectProxy
 
+from opentelemetry.instrumentation.utils import is_instrumentation_enabled
 from opentelemetry.semconv._incubating.attributes.rpc_attributes import (
     RPC_GRPC_STATUS_CODE,
 )
@@ -71,6 +72,9 @@ class OpenTelemetryAioServerInterceptor(
     """
 
     async def intercept_service(self, continuation, handler_call_details):
+        if not is_instrumentation_enabled():
+            return await continuation(handler_call_details)
+
         if self._filter is not None and not self._filter(handler_call_details):
             return await continuation(handler_call_details)
 

instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_server.py

@@ -18,6 +18,7 @@
 
 from opentelemetry import trace
 from opentelemetry.context import attach, detach
+from opentelemetry.instrumentation.utils import is_instrumentation_enabled
 from opentelemetry.propagate import extract
 from opentelemetry.semconv._incubating.attributes.net_attributes import (
     NET_PEER_IP,
@@ -268,6 +269,9 @@ def _start_span(
         )
 
     def intercept_service(self, continuation, handler_call_details):
+        if not is_instrumentation_enabled():
+            return continuation(handler_call_details)
+
         if self._filter is not None and not self._filter(handler_call_details):
             return continuation(handler_call_details)
 

instrumentation/opentelemetry-instrumentation-grpc/tests/test_aio_server_interceptor.py

@@ -12,6 +12,7 @@
     GrpcAioInstrumentorServer,
     aio_server_interceptor,
 )
+from opentelemetry.instrumentation.utils import suppress_instrumentation
 from opentelemetry.sdk import trace as trace_sdk
 from opentelemetry.semconv._incubating.attributes.net_attributes import (
     NET_PEER_IP,
@@ -55,6 +56,15 @@ async def ServerStreamingMethod(self, request, context):
             )
 
 
+class SuppressAioRpcInterceptor(grpc.aio.ServerInterceptor):
+    """An interceptor which runs the rest of the server interceptor chain
+    with instrumentation suppressed."""
+
+    async def intercept_service(self, continuation, handler_call_details):
+        with suppress_instrumentation():
+            return await continuation(handler_call_details)
+
+
 async def run_with_test_server(
     runnable, servicer=Servicer(), interceptors=None
 ):
@@ -339,6 +349,49 @@ async def request(channel):
             parent_span.context.trace_id, child_span.context.trace_id
         )
 
+    async def test_suppress_instrumentation(self):
+        """Check that no span is created for a call when instrumentation
+        is suppressed."""
+        rpc_call = "/GRPCTestServer/SimpleMethod"
+
+        async def request(channel):
+            request = Request(client_id=1, request_data="test")
+            msg = request.SerializeToString()
+            return await channel.unary_unary(rpc_call)(msg)
+
+        await run_with_test_server(
+            request,
+            interceptors=[
+                SuppressAioRpcInterceptor(),
+                aio_server_interceptor(),
+            ],
+        )
+
+        spans_list = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans_list), 0)
+
+    async def test_suppress_instrumentation_streaming(self):
+        """Check that no span is created for a streaming call when
+        instrumentation is suppressed."""
+        rpc_call = "/GRPCTestServer/ServerStreamingMethod"
+
+        async def request(channel):
+            request = Request(client_id=1, request_data="test")
+            msg = request.SerializeToString()
+            async for response in channel.unary_stream(rpc_call)(msg):
+                print(response)
+
+        await run_with_test_server(
+            request,
+            interceptors=[
+                SuppressAioRpcInterceptor(),
+                aio_server_interceptor(),
+            ],
+        )
+
+        spans_list = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans_list), 0)
+
     async def test_span_lifetime(self):
         """Verify that the interceptor captures sub spans within the given
         trace"""

instrumentation/opentelemetry-instrumentation-grpc/tests/test_server_interceptor.py

@@ -18,6 +18,7 @@
     GrpcInstrumentorServer,
     server_interceptor,
 )
+from opentelemetry.instrumentation.utils import suppress_instrumentation
 from opentelemetry.sdk import trace as trace_sdk
 from opentelemetry.semconv._incubating.attributes.net_attributes import (
     NET_PEER_IP,
@@ -59,6 +60,15 @@ def service(self, handler_call_details):
         return UnaryUnaryMethodHandler(self._unary_unary_handler)
 
 
+class SuppressRpcInterceptor(grpc.ServerInterceptor):
+    """An interceptor which runs the rest of the server interceptor chain
+    with instrumentation suppressed."""
+
+    def intercept_service(self, continuation, handler_call_details):
+        with suppress_instrumentation():
+            return continuation(handler_call_details)
+
+
 class Servicer(GRPCTestServerServicer):
     """Our test servicer"""
 
@@ -389,6 +399,54 @@ def ServerStreamingMethod(self, request, context):
             parent_span.context.trace_id, child_span.context.trace_id
         )
 
+    def test_suppress_instrumentation(self):
+        """Check that no span is created for a call when instrumentation
+        is suppressed."""
+
+        interceptor = server_interceptor()
+
+        with self.server(
+            max_workers=1,
+            interceptors=[SuppressRpcInterceptor(), interceptor],
+        ) as (server, channel):
+            add_GRPCTestServerServicer_to_server(Servicer(), server)
+
+            rpc_call = "/GRPCTestServer/SimpleMethod"
+            request = Request(client_id=1, request_data="test")
+            msg = request.SerializeToString()
+            try:
+                server.start()
+                channel.unary_unary(rpc_call)(msg)
+            finally:
+                server.stop(None)
+
+        spans_list = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans_list), 0)
+
+    def test_suppress_instrumentation_streaming(self):
+        """Check that no span is created for a streaming call when
+        instrumentation is suppressed."""
+
+        interceptor = server_interceptor()
+
+        with self.server(
+            max_workers=1,
+            interceptors=[SuppressRpcInterceptor(), interceptor],
+        ) as (server, channel):
+            add_GRPCTestServerServicer_to_server(Servicer(), server)
+
+            rpc_call = "/GRPCTestServer/ServerStreamingMethod"
+            request = Request(client_id=1, request_data="test")
+            msg = request.SerializeToString()
+            try:
+                server.start()
+                list(channel.unary_stream(rpc_call)(msg))
+            finally:
+                server.stop(None)
+
+        spans_list = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans_list), 0)
+
     def test_span_lifetime(self):
         """Check that the span is active for the duration of the call."""