open-telemetry / open-telemetry/opentelemetry-python-contrib
grpc aio server interceptor error with stream_stream calls
Nobody has claimed this yet.
- Dominant language
- Python
- Stars
- 1.1k
- Forks
- 1.1k
- Avg merge
- 4d 15h
- Merged PRs (30d)
- 16
Description
Describe your environment
python = "^3.10"
grpcio = "1.64.3"
opentelemetry-distro = "0.48b0"
opentelemetry-instrumentation-grpc = "0.48b0"
protobuf = "5.28.1"
What happened?
Using aio server interceptor with a stream_stream call results in an error:
File "[PYTHON]/lib/python3.10/site-packages/opentelemetry/instrumentation/grpc/_aio_server.py", line 144, in _stream_interceptor
raise error
File "[PYTHON]/lib/python3.10/site-packages/opentelemetry/instrumentation/grpc/_aio_server.py", line 135, in _stream_interceptor
async for response in behavior(
TypeError: 'async for' requires an object with __aiter__ method, got coroutine
Steps to Reproduce
Use the grpc classes from the example: https://github.com/grpc/grpc/tree/e9c16ace65211b0f0d07ff2fdc7e657865fe8350/examples/python/async_streaming
import asyncio
import logging
from typing import AsyncIterable
import grpc
from opentelemetry.instrumentation.grpc import aio_server_interceptor
import phone_pb2
import phone_pb2_grpc
def create_state_response(
call_state: phone_pb2.CallState.State,
) -> phone_pb2.StreamCallResponse:
response = phone_pb2.StreamCallResponse()
response.call_state.state = call_state
return response
class Phone(phone_pb2_grpc.PhoneServicer):
async def StreamCall(
self,
request_iterator: AsyncIterable[phone_pb2.StreamCallRequest],
context: grpc.aio.ServicerContext,
):
async for _ in request_iterator:
await context.write(
phone_pb2.StreamCallResponse()
)
async def serve(address: str) -> None:
server = grpc.aio.server(interceptors=[aio_server_interceptor()])
phone_pb2_grpc.add_PhoneServicer_to_server(Phone(), server)
server.add_insecure_port(address)
await server.start()
# Make a call to the server
async def call_request_stream():
yield phone_pb2.StreamCallRequest()
async with grpc.aio.insecure_channel(address) as channel:
stub = phone_pb2_grpc.PhoneStub(channel)
async for _ in stub.StreamCall(call_request_stream()):
pass
await server.stop(grace=None)
await server.wait_for_termination()
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
asyncio.run(serve("[::]:50051"))
Expected Result
No error
Actual Result
Error
Additional context
No response
Would you like to implement a fix?
None
Contributor guide
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
Start with the async streaming example linked in the issue and inspect opentelemetry/instrumentation/grpc/_aio_server.py, especially _stream_interceptor around lines 135-144. Run the provided StreamCall reproduction with grpcio 1.64.3 and verify that the aio server interceptor handles the stream_stream call without the async-for TypeError.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- grpc, python
- Domain
- backend-api-design
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100