Merge branch 'main' into proper-buckets-http-stable-django

Author: xrmx

CHANGELOG.md

@@ -17,6 +17,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#3012](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3012))
 - `opentelemetry-instrumentation-django`: proper bucket boundaries in stable semconv http duration
   ([#3524](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3524))
+- `opentelemetry-instrumentation-grpc`: support non-list interceptors
+  ([#3520](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3520))
 
 ### Breaking changes
 

instrumentation/opentelemetry-instrumentation-elasticsearch/tests/test_elasticsearch.py

@@ -31,7 +31,10 @@
     ElasticsearchInstrumentor,
 )
 from opentelemetry.instrumentation.elasticsearch.utils import sanitize_body
-from opentelemetry.semconv.trace import SpanAttributes
+from opentelemetry.semconv._incubating.attributes.db_attributes import (
+    DB_STATEMENT,
+    DB_SYSTEM,
+)
 from opentelemetry.test.test_base import TestBase
 from opentelemetry.trace import StatusCode
 
@@ -76,15 +79,15 @@ def get_elasticsearch_client(*args, **kwargs):
 )
 class TestElasticsearchIntegration(TestBase):
     search_attributes = {
-        SpanAttributes.DB_SYSTEM: "elasticsearch",
+        DB_SYSTEM: "elasticsearch",
         "elasticsearch.url": "/test-index/_search",
         "elasticsearch.method": helpers.dsl_search_method,
         "elasticsearch.target": "test-index",
-        SpanAttributes.DB_STATEMENT: str({"query": {"bool": {"filter": "?"}}}),
+        DB_STATEMENT: str({"query": {"bool": {"filter": "?"}}}),
     }
 
     create_attributes = {
-        SpanAttributes.DB_SYSTEM: "elasticsearch",
+        DB_SYSTEM: "elasticsearch",
         "elasticsearch.url": "/test-index",
         "elasticsearch.method": "HEAD",
     }
@@ -361,13 +364,13 @@ def test_dsl_create(self, request_mock):
         )
 
         attributes = {
-            SpanAttributes.DB_SYSTEM: "elasticsearch",
+            DB_SYSTEM: "elasticsearch",
             "elasticsearch.url": "/test-index",
             "elasticsearch.method": "PUT",
         }
         self.assertSpanHasAttributes(span2, attributes)
         self.assertEqual(
-            literal_eval(span2.attributes[SpanAttributes.DB_STATEMENT]),
+            literal_eval(span2.attributes[DB_STATEMENT]),
             helpers.dsl_create_statement,
         )
 
@@ -408,13 +411,13 @@ def test_dsl_index(self, request_mock):
         span = spans[0]
         self.assertEqual(span.name, helpers.dsl_index_span_name)
         attributes = {
-            SpanAttributes.DB_SYSTEM: "elasticsearch",
+            DB_SYSTEM: "elasticsearch",
             "elasticsearch.url": helpers.dsl_index_url,
             "elasticsearch.method": "PUT",
         }
         self.assertSpanHasAttributes(span, attributes)
         self.assertEqual(
-            literal_eval(span.attributes[SpanAttributes.DB_STATEMENT]),
+            literal_eval(span.attributes[DB_STATEMENT]),
             {
                 "body": "A few words here, a few words there",
                 "title": "About searching",

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

@@ -334,7 +334,8 @@ def _instrument(self, **kwargs):
         tracer_provider = kwargs.get("tracer_provider")
 
         def server(*args, **kwargs):
-            if "interceptors" in kwargs:
+            if "interceptors" in kwargs and kwargs["interceptors"]:
+                kwargs["interceptors"] = list(kwargs["interceptors"])
                 # add our interceptor as the first
                 kwargs["interceptors"].insert(
                     0,
@@ -348,6 +349,7 @@ def server(*args, **kwargs):
                         tracer_provider=tracer_provider, filter_=self._filter
                     )
                 ]
+
             return self._original_func(*args, **kwargs)
 
         grpc.server = server
@@ -386,7 +388,8 @@ def _instrument(self, **kwargs):
         tracer_provider = kwargs.get("tracer_provider")
 
         def server(*args, **kwargs):
-            if "interceptors" in kwargs:
+            if "interceptors" in kwargs and kwargs["interceptors"]:
+                kwargs["interceptors"] = list(kwargs["interceptors"])
                 # add our interceptor as the first
                 kwargs["interceptors"].insert(
                     0,
@@ -516,6 +519,7 @@ def instrumentation_dependencies(self) -> Collection[str]:
 
     def _add_interceptors(self, tracer_provider, kwargs):
         if "interceptors" in kwargs and kwargs["interceptors"]:
+            kwargs["interceptors"] = list(kwargs["interceptors"])
             kwargs["interceptors"] = (
                 aio_client_interceptors(
                     tracer_provider=tracer_provider,

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

@@ -58,13 +58,9 @@ async def ServerStreamingMethod(self, request, context):
 
 
 async def run_with_test_server(
-    runnable, servicer=Servicer(), add_interceptor=True
+    runnable, servicer=Servicer(), interceptors=None
 ):
-    if add_interceptor:
-        interceptors = [aio_server_interceptor()]
-        server = grpc.aio.server(interceptors=interceptors)
-    else:
-        server = grpc.aio.server()
+    server = grpc.aio.server(interceptors=interceptors)
 
     add_GRPCTestServerServicer_to_server(servicer, server)
 
@@ -95,7 +91,7 @@ async def request(channel):
                 msg = request.SerializeToString()
                 return await channel.unary_unary(rpc_call)(msg)
 
-            await run_with_test_server(request, add_interceptor=False)
+            await run_with_test_server(request)
 
             spans_list = self.memory_exporter.get_finished_spans()
             self.assertEqual(len(spans_list), 1)
@@ -140,7 +136,7 @@ async def request(channel):
             msg = request.SerializeToString()
             return await channel.unary_unary(rpc_call)(msg)
 
-        await run_with_test_server(request, add_interceptor=False)
+        await run_with_test_server(request)
 
         spans_list = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans_list), 0)
@@ -154,7 +150,9 @@ async def request(channel):
             msg = request.SerializeToString()
             return await channel.unary_unary(rpc_call)(msg)
 
-        await run_with_test_server(request)
+        await run_with_test_server(
+            request, interceptors=[aio_server_interceptor()]
+        )
 
         spans_list = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans_list), 1)
@@ -206,7 +204,11 @@ async def request(channel):
             msg = request.SerializeToString()
             return await channel.unary_unary(rpc_call)(msg)
 
-        await run_with_test_server(request, servicer=TwoSpanServicer())
+        await run_with_test_server(
+            request,
+            servicer=TwoSpanServicer(),
+            interceptors=[aio_server_interceptor()],
+        )
 
         spans_list = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans_list), 2)
@@ -253,7 +255,9 @@ async def request(channel):
             async for response in channel.unary_stream(rpc_call)(msg):
                 print(response)
 
-        await run_with_test_server(request)
+        await run_with_test_server(
+            request, interceptors=[aio_server_interceptor()]
+        )
 
         spans_list = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans_list), 1)
@@ -307,7 +311,11 @@ async def request(channel):
             async for response in channel.unary_stream(rpc_call)(msg):
                 print(response)
 
-        await run_with_test_server(request, servicer=TwoSpanServicer())
+        await run_with_test_server(
+            request,
+            servicer=TwoSpanServicer(),
+            interceptors=[aio_server_interceptor()],
+        )
 
         spans_list = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans_list), 2)
@@ -367,7 +375,11 @@ async def request(channel):
         lifetime_servicer = SpanLifetimeServicer()
         active_span_before_call = trace.get_current_span()
 
-        await run_with_test_server(request, servicer=lifetime_servicer)
+        await run_with_test_server(
+            request,
+            servicer=lifetime_servicer,
+            interceptors=[aio_server_interceptor()],
+        )
 
         active_span_in_handler = lifetime_servicer.span
         active_span_after_call = trace.get_current_span()
@@ -390,7 +402,9 @@ async def sequential_requests(channel):
             await request(channel)
             await request(channel)
 
-        await run_with_test_server(sequential_requests)
+        await run_with_test_server(
+            sequential_requests, interceptors=[aio_server_interceptor()]
+        )
 
         spans_list = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans_list), 2)
@@ -450,7 +464,9 @@ async def concurrent_requests(channel):
             await asyncio.gather(request(channel), request(channel))
 
         await run_with_test_server(
-            concurrent_requests, servicer=LatchedServicer()
+            concurrent_requests,
+            servicer=LatchedServicer(),
+            interceptors=[aio_server_interceptor()],
         )
 
         spans_list = self.memory_exporter.get_finished_spans()
@@ -504,7 +520,11 @@ async def request(channel):
             self.assertEqual(cm.exception.code(), grpc.StatusCode.INTERNAL)
             self.assertEqual(cm.exception.details(), failure_message)
 
-        await run_with_test_server(request, servicer=AbortServicer())
+        await run_with_test_server(
+            request,
+            servicer=AbortServicer(),
+            interceptors=[aio_server_interceptor()],
+        )
 
         spans_list = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans_list), 1)
@@ -569,7 +589,11 @@ async def request(channel):
             )
             self.assertEqual(cm.exception.details(), failure_message)
 
-        await run_with_test_server(request, servicer=AbortServicer())
+        await run_with_test_server(
+            request,
+            servicer=AbortServicer(),
+            interceptors=[aio_server_interceptor()],
+        )
 
         spans_list = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans_list), 1)
@@ -602,6 +626,60 @@ async def request(channel):
             },
         )
 
+    async def test_non_list_interceptors(self):
+        """Check that we handle non-list interceptors correctly."""
+
+        grpc_server_instrumentor = GrpcAioInstrumentorServer()
+        grpc_server_instrumentor.instrument()
+
+        try:
+            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)
+
+            class MockInterceptor(grpc.aio.ServerInterceptor):
+                async def intercept_service(
+                    self, continuation, handler_call_details
+                ):
+                    return await continuation(handler_call_details)
+
+            await run_with_test_server(
+                request, interceptors=(MockInterceptor(),)
+            )
+
+        finally:
+            grpc_server_instrumentor.uninstrument()
+
+        spans_list = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans_list), 1)
+        span = spans_list[0]
+
+        self.assertEqual(span.name, rpc_call)
+        self.assertIs(span.kind, trace.SpanKind.SERVER)
+
+        # Check version and name in span's instrumentation info
+        self.assertEqualSpanInstrumentationScope(
+            span, opentelemetry.instrumentation.grpc
+        )
+
+        # Check attributes
+        self.assertSpanHasAttributes(
+            span,
+            {
+                SpanAttributes.NET_PEER_IP: "[::1]",
+                SpanAttributes.NET_PEER_NAME: "localhost",
+                SpanAttributes.RPC_METHOD: "SimpleMethod",
+                SpanAttributes.RPC_SERVICE: "GRPCTestServer",
+                SpanAttributes.RPC_SYSTEM: "grpc",
+                SpanAttributes.RPC_GRPC_STATUS_CODE: grpc.StatusCode.OK.value[
+                    0
+                ],
+            },
+        )
+
 
 def get_latch(num):
     """Get a countdown latch function for use in n threads."""