🐛 Bug Report: failed calls are missing from gen_ai.client.operation.duration or recorded without error.type (Google GenAI, Bedrock, Anthropic, OpenAI streams, LiteLLM) #4590

import json
import httpx, boto3
from botocore.awsrequest import AWSResponse
from botocore.config import Config
from google import genai
from google.genai import types
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import InMemoryMetricReader
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.instrumentation.google_generativeai import GoogleGenerativeAiInstrumentor
from opentelemetry.instrumentation.bedrock import BedrockInstrumentor

reader = InMemoryMetricReader()
tp, mp = TracerProvider(), MeterProvider(metric_readers=[reader])
GoogleGenerativeAiInstrumentor().instrument(tracer_provider=tp, meter_provider=mp)
BedrockInstrumentor().instrument(tracer_provider=tp, meter_provider=mp)

def duration_points():
    data = reader.get_metrics_data()
    return sum(p.count for rm in (data.resource_metrics if data else []) for sm in rm.scope_metrics
               for m in sm.metrics if m.name == "gen_ai.client.operation.duration" for p in m.data.data_points)

def check(label, call):
    before = duration_points()
    try:
        call(); outcome = "ok"
    except Exception as e:
        outcome = type(e).__name__
    print(f"{label:28} {outcome:24} new duration points: {duration_points() - before}")

gclient = genai.Client(api_key="x", http_options=types.HttpOptions(httpx_client=httpx.Client(
    transport=httpx.MockTransport(lambda r: httpx.Response(
        500, json={"error": {"code": 500, "message": "boom", "status": "INTERNAL"}})))))
check("google generate_content", lambda: gclient.models.generate_content(model="gemini-2.0-flash", contents="hello"))

class Raw:
    def __init__(self, d): self.d = d
    def stream(self, *a, **k): yield self.d
    def read(self, amt=None): return self.d
    def close(self): pass
    release_conn = close

bclient = boto3.client("bedrock-runtime", region_name="us-east-1", aws_access_key_id="x",
                       aws_secret_access_key="x", config=Config(retries={"total_max_attempts": 1}))
bclient.meta.events.register("before-send.bedrock-runtime.*", lambda request, **kw: AWSResponse(
    request.url, 500, {"content-type": "application/json", "x-amzn-errortype": "InternalServerException"},
    Raw(json.dumps({"message": "boom"}).encode())))
model = "anthropic.claude-3-haiku-20240307-v1:0"
check("bedrock invoke_model", lambda: bclient.invoke_model(modelId=model, body=json.dumps(
    {"anthropic_version": "bedrock-2023-05-31", "max_tokens": 50, "messages": [{"role": "user", "content": "hi"}]})))
check("bedrock converse", lambda: bclient.converse(modelId=model,
                                                   messages=[{"role": "user", "content": [{"text": "hi"}]}]))

import json
import httpx, openai, anthropic
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import InMemoryMetricReader
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.instrumentation.openai import OpenAIInstrumentor
from opentelemetry.instrumentation.anthropic import AnthropicInstrumentor

reader = InMemoryMetricReader()
tp, mp = TracerProvider(), MeterProvider(metric_readers=[reader])
OpenAIInstrumentor().instrument(tracer_provider=tp, meter_provider=mp)
AnthropicInstrumentor().instrument(tracer_provider=tp, meter_provider=mp)

def duration_points():
    out = {}
    data = reader.get_metrics_data()
    for rm in (data.resource_metrics if data else []):
        for sm in rm.scope_metrics:
            for m in sm.metrics:
                if m.name == "gen_ai.client.operation.duration":
                    for p in m.data.data_points:
                        k = p.attributes.get("error.type") or "no error.type"
                        out[k] = out.get(k, 0) + p.count
    return out

def broken(hx, first):        # sends the first SSE event, then the connection breaks
    class S(hx.SyncByteStream):
        def __iter__(self):
            yield first
            raise hx.ReadError("connection broken")
    return hx.MockTransport(lambda r: hx.Response(200, headers={"content-type": "text/event-stream"}, stream=S()))

def check(label, call):
    before = duration_points()
    try:
        call(); outcome = "ok"
    except Exception as e:
        outcome = type(e).__name__
    after = duration_points()
    print(label, outcome, {k: c - before.get(k, 0) for k, c in after.items() if c != before.get(k, 0)} or "NONE")

chunk = {"id": "c1", "object": "chat.completion.chunk", "created": 0, "model": "gpt-4o-mini",
         "choices": [{"index": 0, "delta": {"role": "assistant", "content": "hi"}, "finish_reason": None}]}
oc = openai.OpenAI(api_key="x", max_retries=0, http_client=httpx.Client(
    transport=broken(httpx, f"data: {json.dumps(chunk)}\n\n".encode())))
check("openai chat stream fails mid-way:", lambda: list(oc.chat.completions.create(
    model="gpt-4o-mini", messages=[{"role": "user", "content": "hello"}], stream=True)))

import anthropic._base_client as bc
hx = getattr(bc, "httpx2", None) or httpx          # newer anthropic SDKs use httpx2
start = {"type": "message_start", "message": {"id": "m1", "type": "message", "role": "assistant", "model": "claude-x",
         "content": [], "stop_reason": None, "stop_sequence": None, "usage": {"input_tokens": 7, "output_tokens": 0}}}
ac = anthropic.Anthropic(api_key="x", max_retries=0, http_client=hx.Client(
    transport=broken(hx, f"event: message_start\ndata: {json.dumps(start)}\n\n".encode())))
check("anthropic stream fails mid-way:", lambda: list(ac.messages.create(
    model="claude-x", max_tokens=50, messages=[{"role": "user", "content": "hello"}], stream=True)))

import os
os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True"
import httpx, litellm, openai
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import InMemoryMetricReader
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.instrumentation.litellm import LiteLLMInstrumentor

reader = InMemoryMetricReader()
LiteLLMInstrumentor().instrument(tracer_provider=TracerProvider(), meter_provider=MeterProvider(metric_readers=[reader]))

def duration_points():
    data = reader.get_metrics_data()
    return sum(p.count for rm in (data.resource_metrics if data else []) for sm in rm.scope_metrics
               for m in sm.metrics if m.name == "gen_ai.client.operation.duration" for p in m.data.data_points)

client = openai.OpenAI(api_key="x", max_retries=0, http_client=httpx.Client(transport=httpx.MockTransport(
    lambda r: httpx.Response(500, json={"error": {"message": "boom", "type": "server_error"}}))))
for label, call in (("litellm completion", lambda: litellm.completion(model="openai/gpt-4o-mini", client=client, api_key="x",
                                                                       messages=[{"role": "user", "content": "hello"}])),
                    ("litellm embedding", lambda: litellm.embedding(model="openai/text-embedding-3-small", client=client,
                                                                     api_key="x", input=["hello"]))):
    before = duration_points()
    try:
        call(); outcome = "ok"
    except Exception as e:
        outcome = type(e).__name__
    print(f"{label:28} {outcome:24} new duration points: {duration_points() - before}")