Skip to content

Commit f9d9f19

Browse files
sentrivanasentrivana
authored
fix(grpc): Allow non-list interceptors (open-telemetry#3520)
* fix(grpc): Support non-list interceptors The GRPC integration assumes that if interceptors are provided, they will be a list, when GRPC itself types them as Sequence. With this change, we're making the codepaths using interceptors more robust by explicitly turning them into lists before manipulating them. * Add a changelog entry * . * Run ruff * Simplify
1 parent 0db4d0b commit f9d9f19

File tree

4 files changed

+189
-50
lines changed

4 files changed

+189
-50
lines changed

‎CHANGELOG.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
1515

1616
- `opentelemetry-instrumentation-fastapi`: fix wrapping of middlewares
1717
([#3012](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3012))
18+
- `opentelemetry-instrumentation-grpc`: support non-list interceptors
19+
([#3520](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3520))
1820

1921
### Breaking changes
2022

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

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -334,7 +334,8 @@ def _instrument(self, **kwargs):
334334
tracer_provider = kwargs.get("tracer_provider")
335335

336336
def server(*args, **kwargs):
337-
if "interceptors" in kwargs:
337+
if "interceptors" in kwargs and kwargs["interceptors"]:
338+
kwargs["interceptors"] = list(kwargs["interceptors"])
338339
# add our interceptor as the first
339340
kwargs["interceptors"].insert(
340341
0,
@@ -348,6 +349,7 @@ def server(*args, **kwargs):
348349
tracer_provider=tracer_provider, filter_=self._filter
349350
)
350351
]
352+
351353
return self._original_func(*args, **kwargs)
352354

353355
grpc.server = server
@@ -386,7 +388,8 @@ def _instrument(self, **kwargs):
386388
tracer_provider = kwargs.get("tracer_provider")
387389

388390
def server(*args, **kwargs):
389-
if "interceptors" in kwargs:
391+
if "interceptors" in kwargs and kwargs["interceptors"]:
392+
kwargs["interceptors"] = list(kwargs["interceptors"])
390393
# add our interceptor as the first
391394
kwargs["interceptors"].insert(
392395
0,
@@ -516,6 +519,7 @@ def instrumentation_dependencies(self) -> Collection[str]:
516519

517520
def _add_interceptors(self, tracer_provider, kwargs):
518521
if "interceptors" in kwargs and kwargs["interceptors"]:
522+
kwargs["interceptors"] = list(kwargs["interceptors"])
519523
kwargs["interceptors"] = (
520524
aio_client_interceptors(
521525
tracer_provider=tracer_provider,

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

Lines changed: 95 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -58,13 +58,9 @@ async def ServerStreamingMethod(self, request, context):
5858

5959

6060
async def run_with_test_server(
61-
runnable, servicer=Servicer(), add_interceptor=True
61+
runnable, servicer=Servicer(), interceptors=None
6262
):
63-
if add_interceptor:
64-
interceptors = [aio_server_interceptor()]
65-
server = grpc.aio.server(interceptors=interceptors)
66-
else:
67-
server = grpc.aio.server()
63+
server = grpc.aio.server(interceptors=interceptors)
6864

6965
add_GRPCTestServerServicer_to_server(servicer, server)
7066

@@ -95,7 +91,7 @@ async def request(channel):
9591
msg = request.SerializeToString()
9692
return await channel.unary_unary(rpc_call)(msg)
9793

98-
await run_with_test_server(request, add_interceptor=False)
94+
await run_with_test_server(request)
9995

10096
spans_list = self.memory_exporter.get_finished_spans()
10197
self.assertEqual(len(spans_list), 1)
@@ -140,7 +136,7 @@ async def request(channel):
140136
msg = request.SerializeToString()
141137
return await channel.unary_unary(rpc_call)(msg)
142138

143-
await run_with_test_server(request, add_interceptor=False)
139+
await run_with_test_server(request)
144140

145141
spans_list = self.memory_exporter.get_finished_spans()
146142
self.assertEqual(len(spans_list), 0)
@@ -154,7 +150,9 @@ async def request(channel):
154150
msg = request.SerializeToString()
155151
return await channel.unary_unary(rpc_call)(msg)
156152

157-
await run_with_test_server(request)
153+
await run_with_test_server(
154+
request, interceptors=[aio_server_interceptor()]
155+
)
158156

159157
spans_list = self.memory_exporter.get_finished_spans()
160158
self.assertEqual(len(spans_list), 1)
@@ -206,7 +204,11 @@ async def request(channel):
206204
msg = request.SerializeToString()
207205
return await channel.unary_unary(rpc_call)(msg)
208206

209-
await run_with_test_server(request, servicer=TwoSpanServicer())
207+
await run_with_test_server(
208+
request,
209+
servicer=TwoSpanServicer(),
210+
interceptors=[aio_server_interceptor()],
211+
)
210212

211213
spans_list = self.memory_exporter.get_finished_spans()
212214
self.assertEqual(len(spans_list), 2)
@@ -253,7 +255,9 @@ async def request(channel):
253255
async for response in channel.unary_stream(rpc_call)(msg):
254256
print(response)
255257

256-
await run_with_test_server(request)
258+
await run_with_test_server(
259+
request, interceptors=[aio_server_interceptor()]
260+
)
257261

258262
spans_list = self.memory_exporter.get_finished_spans()
259263
self.assertEqual(len(spans_list), 1)
@@ -307,7 +311,11 @@ async def request(channel):
307311
async for response in channel.unary_stream(rpc_call)(msg):
308312
print(response)
309313

310-
await run_with_test_server(request, servicer=TwoSpanServicer())
314+
await run_with_test_server(
315+
request,
316+
servicer=TwoSpanServicer(),
317+
interceptors=[aio_server_interceptor()],
318+
)
311319

312320
spans_list = self.memory_exporter.get_finished_spans()
313321
self.assertEqual(len(spans_list), 2)
@@ -367,7 +375,11 @@ async def request(channel):
367375
lifetime_servicer = SpanLifetimeServicer()
368376
active_span_before_call = trace.get_current_span()
369377

370-
await run_with_test_server(request, servicer=lifetime_servicer)
378+
await run_with_test_server(
379+
request,
380+
servicer=lifetime_servicer,
381+
interceptors=[aio_server_interceptor()],
382+
)
371383

372384
active_span_in_handler = lifetime_servicer.span
373385
active_span_after_call = trace.get_current_span()
@@ -390,7 +402,9 @@ async def sequential_requests(channel):
390402
await request(channel)
391403
await request(channel)
392404

393-
await run_with_test_server(sequential_requests)
405+
await run_with_test_server(
406+
sequential_requests, interceptors=[aio_server_interceptor()]
407+
)
394408

395409
spans_list = self.memory_exporter.get_finished_spans()
396410
self.assertEqual(len(spans_list), 2)
@@ -450,7 +464,9 @@ async def concurrent_requests(channel):
450464
await asyncio.gather(request(channel), request(channel))
451465

452466
await run_with_test_server(
453-
concurrent_requests, servicer=LatchedServicer()
467+
concurrent_requests,
468+
servicer=LatchedServicer(),
469+
interceptors=[aio_server_interceptor()],
454470
)
455471

456472
spans_list = self.memory_exporter.get_finished_spans()
@@ -504,7 +520,11 @@ async def request(channel):
504520
self.assertEqual(cm.exception.code(), grpc.StatusCode.INTERNAL)
505521
self.assertEqual(cm.exception.details(), failure_message)
506522

507-
await run_with_test_server(request, servicer=AbortServicer())
523+
await run_with_test_server(
524+
request,
525+
servicer=AbortServicer(),
526+
interceptors=[aio_server_interceptor()],
527+
)
508528

509529
spans_list = self.memory_exporter.get_finished_spans()
510530
self.assertEqual(len(spans_list), 1)
@@ -569,7 +589,11 @@ async def request(channel):
569589
)
570590
self.assertEqual(cm.exception.details(), failure_message)
571591

572-
await run_with_test_server(request, servicer=AbortServicer())
592+
await run_with_test_server(
593+
request,
594+
servicer=AbortServicer(),
595+
interceptors=[aio_server_interceptor()],
596+
)
573597

574598
spans_list = self.memory_exporter.get_finished_spans()
575599
self.assertEqual(len(spans_list), 1)
@@ -602,6 +626,60 @@ async def request(channel):
602626
},
603627
)
604628

629+
async def test_non_list_interceptors(self):
630+
"""Check that we handle non-list interceptors correctly."""
631+
632+
grpc_server_instrumentor = GrpcAioInstrumentorServer()
633+
grpc_server_instrumentor.instrument()
634+
635+
try:
636+
rpc_call = "/GRPCTestServer/SimpleMethod"
637+
638+
async def request(channel):
639+
request = Request(client_id=1, request_data="test")
640+
msg = request.SerializeToString()
641+
return await channel.unary_unary(rpc_call)(msg)
642+
643+
class MockInterceptor(grpc.aio.ServerInterceptor):
644+
async def intercept_service(
645+
self, continuation, handler_call_details
646+
):
647+
return await continuation(handler_call_details)
648+
649+
await run_with_test_server(
650+
request, interceptors=(MockInterceptor(),)
651+
)
652+
653+
finally:
654+
grpc_server_instrumentor.uninstrument()
655+
656+
spans_list = self.memory_exporter.get_finished_spans()
657+
self.assertEqual(len(spans_list), 1)
658+
span = spans_list[0]
659+
660+
self.assertEqual(span.name, rpc_call)
661+
self.assertIs(span.kind, trace.SpanKind.SERVER)
662+
663+
# Check version and name in span's instrumentation info
664+
self.assertEqualSpanInstrumentationScope(
665+
span, opentelemetry.instrumentation.grpc
666+
)
667+
668+
# Check attributes
669+
self.assertSpanHasAttributes(
670+
span,
671+
{
672+
SpanAttributes.NET_PEER_IP: "[::1]",
673+
SpanAttributes.NET_PEER_NAME: "localhost",
674+
SpanAttributes.RPC_METHOD: "SimpleMethod",
675+
SpanAttributes.RPC_SERVICE: "GRPCTestServer",
676+
SpanAttributes.RPC_SYSTEM: "grpc",
677+
SpanAttributes.RPC_GRPC_STATUS_CODE: grpc.StatusCode.OK.value[
678+
0
679+
],
680+
},
681+
)
682+
605683

606684
def get_latch(num):
607685
"""Get a countdown latch function for use in n threads."""

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

Lines changed: 86 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import tempfile
2020
import threading
2121
from concurrent import futures
22+
from unittest import mock
2223

2324
import grpc
2425

@@ -104,41 +105,46 @@ def handler(request, context):
104105

105106
grpc_server_instrumentor = GrpcInstrumentorServer()
106107
grpc_server_instrumentor.instrument()
107-
with self.server(max_workers=1) as (server, channel):
108-
server.add_generic_rpc_handlers((UnaryUnaryRpcHandler(handler),))
109-
rpc_call = "TestServicer/handler"
110-
try:
111-
server.start()
112-
channel.unary_unary(rpc_call)(b"test")
113-
finally:
114-
server.stop(None)
115108

116-
spans_list = self.memory_exporter.get_finished_spans()
117-
self.assertEqual(len(spans_list), 1)
118-
span = spans_list[0]
119-
self.assertEqual(span.name, rpc_call)
120-
self.assertIs(span.kind, trace.SpanKind.SERVER)
109+
try:
110+
with self.server(max_workers=1) as (server, channel):
111+
server.add_generic_rpc_handlers(
112+
(UnaryUnaryRpcHandler(handler),)
113+
)
114+
rpc_call = "TestServicer/handler"
115+
try:
116+
server.start()
117+
channel.unary_unary(rpc_call)(b"test")
118+
finally:
119+
server.stop(None)
120+
121+
finally:
122+
grpc_server_instrumentor.uninstrument()
121123

122-
# Check version and name in span's instrumentation info
123-
self.assertEqualSpanInstrumentationScope(
124-
span, opentelemetry.instrumentation.grpc
125-
)
124+
spans_list = self.memory_exporter.get_finished_spans()
125+
self.assertEqual(len(spans_list), 1)
126+
span = spans_list[0]
127+
self.assertEqual(span.name, rpc_call)
128+
self.assertIs(span.kind, trace.SpanKind.SERVER)
126129

127-
# Check attributes
128-
self.assertSpanHasAttributes(
129-
span,
130-
{
131-
**self.net_peer_span_attributes,
132-
SpanAttributes.RPC_METHOD: "handler",
133-
SpanAttributes.RPC_SERVICE: "TestServicer",
134-
SpanAttributes.RPC_SYSTEM: "grpc",
135-
SpanAttributes.RPC_GRPC_STATUS_CODE: grpc.StatusCode.OK.value[
136-
0
137-
],
138-
},
139-
)
130+
# Check version and name in span's instrumentation info
131+
self.assertEqualSpanInstrumentationScope(
132+
span, opentelemetry.instrumentation.grpc
133+
)
140134

141-
grpc_server_instrumentor.uninstrument()
135+
# Check attributes
136+
self.assertSpanHasAttributes(
137+
span,
138+
{
139+
**self.net_peer_span_attributes,
140+
SpanAttributes.RPC_METHOD: "handler",
141+
SpanAttributes.RPC_SERVICE: "TestServicer",
142+
SpanAttributes.RPC_SYSTEM: "grpc",
143+
SpanAttributes.RPC_GRPC_STATUS_CODE: grpc.StatusCode.OK.value[
144+
0
145+
],
146+
},
147+
)
142148

143149
def test_uninstrument(self):
144150
def handler(request, context):
@@ -647,6 +653,55 @@ def unset_status_handler(request, context):
647653
},
648654
)
649655

656+
def test_non_list_interceptors(self):
657+
"""Check that we handle non-list interceptors correctly."""
658+
grpc_server_instrumentor = GrpcInstrumentorServer()
659+
grpc_server_instrumentor.instrument()
660+
661+
try:
662+
with self.server(
663+
max_workers=1,
664+
interceptors=(mock.MagicMock(),),
665+
) as (server, channel):
666+
add_GRPCTestServerServicer_to_server(Servicer(), server)
667+
668+
rpc_call = "/GRPCTestServer/SimpleMethod"
669+
request = Request(client_id=1, request_data="test")
670+
msg = request.SerializeToString()
671+
try:
672+
server.start()
673+
channel.unary_unary(rpc_call)(msg)
674+
finally:
675+
server.stop(None)
676+
finally:
677+
grpc_server_instrumentor.uninstrument()
678+
679+
spans_list = self.memory_exporter.get_finished_spans()
680+
self.assertEqual(len(spans_list), 1)
681+
span = spans_list[0]
682+
683+
self.assertEqual(span.name, rpc_call)
684+
self.assertIs(span.kind, trace.SpanKind.SERVER)
685+
686+
# Check version and name in span's instrumentation info
687+
self.assertEqualSpanInstrumentationScope(
688+
span, opentelemetry.instrumentation.grpc
689+
)
690+
691+
# Check attributes
692+
self.assertSpanHasAttributes(
693+
span,
694+
{
695+
**self.net_peer_span_attributes,
696+
SpanAttributes.RPC_METHOD: "SimpleMethod",
697+
SpanAttributes.RPC_SERVICE: "GRPCTestServer",
698+
SpanAttributes.RPC_SYSTEM: "grpc",
699+
SpanAttributes.RPC_GRPC_STATUS_CODE: grpc.StatusCode.OK.value[
700+
0
701+
],
702+
},
703+
)
704+
650705

651706
class TestOpenTelemetryServerInterceptorUnix(
652707
TestOpenTelemetryServerInterceptor,

0 commit comments

Comments
 (0)