🐛 Bug Report: openai, anthropic, google-generativeai: a call cancelled while waiting for the response (asyncio timeout/cancel, Ctrl-C) never ends its span #4576

import asyncio, gc, logging, sys
from importlib.metadata import version

import httpx
from opentelemetry.sdk.trace import SpanProcessor, TracerProvider

logging.disable(logging.WARNING)
SDK = sys.argv[1] if len(sys.argv) > 1 else "openai"


class Spans(SpanProcessor):
    def __init__(self): self.open, self.ended = set(), []
    def on_start(self, span, parent_context=None): self.open.add(span.context.span_id)
    def on_end(self, span):
        self.open.discard(span.context.span_id)
        self.ended.append(f"{span.name} [{span.status.status_code.name}]")


spans = Spans()
provider = TracerProvider()
provider.add_span_processor(spans)


def slow_http(hx=httpx):                           # a server that answers after 5 s
    async def slow_server(request):
        await asyncio.sleep(5)
        return hx.Response(200, json={})
    return hx.AsyncClient(transport=hx.MockTransport(slow_server))


M = [{"role": "user", "content": "hi"}]
if SDK == "openai":
    import openai
    from opentelemetry.instrumentation.openai import OpenAIInstrumentor
    OpenAIInstrumentor().instrument(tracer_provider=provider)
    client = openai.AsyncOpenAI(api_key="x", base_url="http://mock/v1", max_retries=0, http_client=slow_http())
    calls = {"create": lambda: client.chat.completions.create(model="m", messages=M),
             "create(stream=True)": lambda: client.chat.completions.create(model="m", messages=M, stream=True)}
    pkg = ("openai", "opentelemetry-instrumentation-openai")
elif SDK == "anthropic":
    import anthropic, anthropic._base_client as base
    from opentelemetry.instrumentation.anthropic import AnthropicInstrumentor
    AnthropicInstrumentor().instrument(tracer_provider=provider)
    hx = getattr(base, "httpx2", None) or httpx            # newer anthropic releases use httpx2
    client = anthropic.AsyncAnthropic(api_key="x", base_url="http://mock", max_retries=0, http_client=slow_http(hx))
    calls = {"create": lambda: client.messages.create(model="m", max_tokens=50, messages=M),
             "create(stream=True)": lambda: client.messages.create(model="m", max_tokens=50, messages=M, stream=True)}
    pkg = ("anthropic", "opentelemetry-instrumentation-anthropic")
else:
    from google import genai
    from google.genai import types
    from opentelemetry.instrumentation.google_generativeai import GoogleGenerativeAiInstrumentor
    GoogleGenerativeAiInstrumentor().instrument(tracer_provider=provider)
    client = genai.Client(api_key="x", http_options=types.HttpOptions(base_url="http://mock", httpx_async_client=slow_http()))
    calls = {"generate_content": lambda: client.aio.models.generate_content(model="m", contents="hi"),
             "generate_content_stream": lambda: client.aio.models.generate_content_stream(model="m", contents="hi")}
    pkg = ("google-genai", "opentelemetry-instrumentation-google-generativeai")


async def main():
    print(" | ".join(f"{p} {version(p)}" for p in pkg))
    for name, call in calls.items():
        spans.ended.clear()
        try:
            await asyncio.wait_for(call(), timeout=1.0)    # the app gives up after 1 s
        except TimeoutError:
            pass
        gc.collect()
        await asyncio.sleep(0.1)
        print(f"{name:25} timed out -> ended spans: {spans.ended or 'none'}; spans never ended: {len(spans.open)}")
        spans.open.clear()

asyncio.run(main())