Thread-ConsumeBidirectionalStream caught unexpected exception 'generator' object has no attribute 'add_done_callback' and will exit.

[mharoonkwentra]

Notes
We are trying to use Opentelemetry to instrument grpc to get traces of functions. We observed this error on local once, and we couldn't reproduce it again, but it's always on production server. Furthermore, we are using GCP monitoring suite (stack drive) to check traces.

Environment:
Django==3.2.6
Python==3.7

grpcio-tools==1.32.0
grpcio==1.33.2

opentelemetry-api==1.4.1
opentelemetry-sdk==1.4.1
opentelemetry-instrumentation-grpc==0.23b2

Instrumentation code:
set_global_textmap(CloudTraceFormatPropagator())
# sampling sends a portion of generated spans based on a 1/X population size
sample_rate = int(os.getenv('SAMPLE_RATE', '1'))
sampler = TraceIdRatioBased(1 / sample_rate)
tracer_provider = TracerProvider(sampler=sampler)
cloud_trace_exporter = CloudTraceSpanExporter()
tracer_provider.add_span_processor(SimpleSpanProcessor(cloud_trace_exporter))
trace.set_tracer_provider(tracer_provider)
tracer = trace.get_tracer(name)
GrpcInstrumentorClient().instrument() # when commented error isn't reproduced
except (ValueError, NotImplementedError, DefaultCredentialsError) as exc:
print(exc) # Handle errors here

The error
Traceback (most recent call last):
File "/usr/local/lib/python3.7/site-packages/google/api_core/bidi.py", line 637, in _thread_main
self._bidi_rpc.open()
File "/usr/local/lib/python3.7/site-packages/google/api_core/bidi.py", line 287, in open
call._wrapped.add_done_callback(self._on_call_done)
AttributeError: 'generator' object has no attribute 'add_done_callback'

[AkselAllas]

I have the same issue. Not using django. Maybe a grpcio issue?

[drice]

I've found this occurs when using StreamingPullManager from google-cloud-pubsub, as in this sample:

import google.cloud.pubsub_v1 as pubsub_v1
from opentelemetry.instrumentation.grpc import GrpcInstrumentorClient


def sample_streaming_pull():
    # Create a client
    client = pubsub_v1.SubscriberClient()
    future = client.subscribe(
        subscription="<your pubsub subscription ID>",
        callback=print
    )
    with client:
        future.result()


if __name__ == "__main__":
    client = GrpcInstrumentorClient()
    client.instrument()

    sample_streaming_pull()

I think the issue is potentially here

opentelemetry-python-contrib/instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_client.py

Line 228 in 65b7035

if client_info.is_server_stream:

Instead of checking for only is_server_stream:

        if client_info.is_server_stream:
            return self._intercept_server_stream(
                request_or_iterator, metadata, client_info, invoker
            )

should this check that it's not bidirectional?

if client_info.is_server_stream and not client_info.is_client_stream:

It looks like from the referenced PRs that #1583 would potentially resolve.

[jaydeerossi]

I'm having this same issue right now.

[pickfire]

Same issue with google ads

[arjunraman-ant]

@drice I had the same issue with Pub/Sub recently. The issue was SubscriberClient is NOT thread safe, and I was sharing the client. Creating a SubscriberClient per thread fixed it. There is no documentation of this in any of the GCP docs unfortunately.

[ElliotSteene-Ki]

Also having this issue:

Environment:
FastAPI==0.109.2
Python==3.11

grpcio==1.70.0

opentelemetry-api==1.30.0
opentelemetry-sdk==1.30.0
opentelemetry-instrumentation-grpc==0.51b0

@arjunraman-ant - do you have any details about how these were run in separate threads?

[robert-lore]

Having same issue.

[bartek-kossak-flyr]

me too

[robert-lore]

@drice I had the same issue with Pub/Sub recently. The issue was SubscriberClient is NOT thread safe, and I was sharing the client. Creating a SubscriberClient per thread fixed it. There is no documentation of this in any of the GCP docs unfortunately.

Could you post your solution? It may help us.