diff --git a/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/utils.py b/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/utils.py index 338550693b..cf0f92d9f5 100644 --- a/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/utils.py +++ b/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/utils.py @@ -129,6 +129,24 @@ def wrapper(wrapped, instance, args, kwargs): return _with_chat_telemetry +def _with_responses_telemetry_wrapper(func): + def _with_responses_telemetry(tracer, token_histogram=None, duration_histogram=None): + def wrapper(wrapped, instance, args, kwargs): + return func( + tracer, + token_histogram, + duration_histogram, + wrapped, + instance, + args, + kwargs, + ) + + return wrapper + + return _with_responses_telemetry + + def _with_tracer_wrapper(func): def _with_tracer(tracer): def wrapper(wrapped, instance, args, kwargs): diff --git a/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/v1/__init__.py b/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/v1/__init__.py index e7e3deb56b..7c14fa7d92 100644 --- a/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/v1/__init__.py +++ b/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/v1/__init__.py @@ -305,17 +305,17 @@ def _instrument(self, **kwargs): self._try_wrap( "openai.resources.responses", "Responses.create", - responses_get_or_create_wrapper(tracer), + responses_get_or_create_wrapper(tracer, tokens_histogram, duration_histogram), ) self._try_wrap( "openai.resources.responses", "Responses.retrieve", - responses_get_or_create_wrapper(tracer), + responses_get_or_create_wrapper(tracer, tokens_histogram, duration_histogram), ) self._try_wrap( "openai.resources.responses", "Responses.parse", - responses_get_or_create_wrapper(tracer), + responses_get_or_create_wrapper(tracer, tokens_histogram, duration_histogram), ) self._try_wrap( "openai.resources.responses", @@ -325,17 +325,17 @@ def _instrument(self, **kwargs): self._try_wrap( "openai.resources.responses", "AsyncResponses.create", - async_responses_get_or_create_wrapper(tracer), + async_responses_get_or_create_wrapper(tracer, tokens_histogram, duration_histogram), ) self._try_wrap( "openai.resources.responses", "AsyncResponses.retrieve", - async_responses_get_or_create_wrapper(tracer), + async_responses_get_or_create_wrapper(tracer, tokens_histogram, duration_histogram), ) self._try_wrap( "openai.resources.responses", "AsyncResponses.parse", - async_responses_get_or_create_wrapper(tracer), + async_responses_get_or_create_wrapper(tracer, tokens_histogram, duration_histogram), ) self._try_wrap( "openai.resources.responses", diff --git a/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/v1/responses_wrappers.py b/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/v1/responses_wrappers.py index 340402ea9b..12caf2bfc5 100644 --- a/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/v1/responses_wrappers.py +++ b/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/v1/responses_wrappers.py @@ -23,14 +23,16 @@ from opentelemetry.instrumentation.openai.shared import ( _build_tool_def_dict, _extract_model_name_from_provider_format, + _get_openai_base_url, _parse_arguments, _set_request_attributes, _set_span_attribute, _set_tool_definitions_json, + metric_shared_attributes, model_as_dict, ) - from opentelemetry.instrumentation.openai.utils import ( + _with_responses_telemetry_wrapper, _with_tracer_wrapper, dont_throw, json_dumps, @@ -534,9 +536,44 @@ def set_data_attributes(traced_response: TracedData, span: Span): _set_responses_json_messages(traced_response, span) +def _responses_metric_attributes(instance, response_model, is_streaming=False): + return metric_shared_attributes( + response_model=_extract_model_name_from_provider_format(response_model) or None, + operation="chat", + server_address=_get_openai_base_url(instance), + is_streaming=is_streaming, + ) + + +def _error_metric_attributes(instance, request_model, exception, is_streaming=False): + attributes = _responses_metric_attributes(instance, None, is_streaming) + del attributes[GenAIAttributes.GEN_AI_RESPONSE_MODEL] + if request_model: + attributes[GenAIAttributes.GEN_AI_REQUEST_MODEL] = request_model + attributes[ERROR_TYPE] = exception.__class__.__name__ + return attributes + + @dont_throw -@_with_tracer_wrapper -def responses_get_or_create_wrapper(tracer: Tracer, wrapped, instance, args, kwargs): +def _record_duration(duration_histogram, start_time, attributes): + if duration_histogram: + duration_histogram.record((time.time_ns() - start_time) / 1e9, attributes=attributes) + + +@dont_throw +def _record_token_usage(token_histogram, usage, attributes): + if token_histogram and usage: + for token_type, count in (("input", usage.input_tokens), ("output", usage.output_tokens)): + token_histogram.record( + count, attributes={**attributes, GenAIAttributes.GEN_AI_TOKEN_TYPE: token_type} + ) + + +@dont_throw +@_with_responses_telemetry_wrapper +def responses_get_or_create_wrapper( + tracer: Tracer, token_histogram, duration_histogram, wrapped, instance, args, kwargs +): if context_api.get_value(_SUPPRESS_INSTRUMENTATION_KEY): return wrapped(*args, **kwargs) start_time = time.time_ns() @@ -563,8 +600,18 @@ def responses_get_or_create_wrapper(tracer: Tracer, wrapped, instance, args, kwa start_time=start_time, request_kwargs=non_sentinel_kwargs, tracer=tracer, + instance=instance, + token_histogram=token_histogram, + duration_histogram=duration_histogram, ) except Exception as e: + _record_duration( + duration_histogram, + start_time, + _error_metric_attributes( + instance, non_sentinel_kwargs.get("model"), e, bool(non_sentinel_kwargs.get("stream")) + ), + ) response_id = non_sentinel_kwargs.get("response_id") existing_data = {} if response_id and response_id in responses: @@ -708,14 +755,18 @@ def responses_get_or_create_wrapper(tracer: Tracer, wrapped, instance, args, kwa _set_request_attributes(span, prepare_kwargs_for_shared_attributes(non_sentinel_kwargs), instance) set_data_attributes(traced_data, span) span.end() + # One point per emitted span, so background polls and repeat retrieves don't add operations + metric_attributes = _responses_metric_attributes(instance, parsed_response.model) + _record_duration(duration_histogram, traced_data.start_time, metric_attributes) + _record_token_usage(token_histogram, traced_data.usage, metric_attributes) return response @dont_throw -@_with_tracer_wrapper +@_with_responses_telemetry_wrapper async def async_responses_get_or_create_wrapper( - tracer: Tracer, wrapped, instance, args, kwargs + tracer: Tracer, token_histogram, duration_histogram, wrapped, instance, args, kwargs ): if context_api.get_value(_SUPPRESS_INSTRUMENTATION_KEY): return await wrapped(*args, **kwargs) @@ -743,8 +794,18 @@ async def async_responses_get_or_create_wrapper( start_time=start_time, request_kwargs=non_sentinel_kwargs, tracer=tracer, + instance=instance, + token_histogram=token_histogram, + duration_histogram=duration_histogram, ) except Exception as e: + _record_duration( + duration_histogram, + start_time, + _error_metric_attributes( + instance, non_sentinel_kwargs.get("model"), e, bool(non_sentinel_kwargs.get("stream")) + ), + ) response_id = non_sentinel_kwargs.get("response_id") existing_data = {} if response_id and response_id in responses: @@ -886,6 +947,10 @@ async def async_responses_get_or_create_wrapper( _set_request_attributes(span, prepare_kwargs_for_shared_attributes(non_sentinel_kwargs), instance) set_data_attributes(traced_data, span) span.end() + # One point per emitted span, so background polls and repeat retrieves don't add operations + metric_attributes = _responses_metric_attributes(instance, parsed_response.model) + _record_duration(duration_histogram, traced_data.start_time, metric_attributes) + _record_token_usage(token_histogram, traced_data.usage, metric_attributes) return response @@ -960,6 +1025,9 @@ class ResponseStream(ObjectProxy): _request_kwargs = None _tracer = None _traced_data = None + _instance = None + _token_histogram = None + _duration_histogram = None def __init__( self, @@ -969,10 +1037,16 @@ def __init__( request_kwargs=None, tracer=None, traced_data=None, + instance=None, + token_histogram=None, + duration_histogram=None, ): super().__init__(response) self._span = span self._start_time = start_time + self._instance = instance + self._token_histogram = token_histogram + self._duration_histogram = duration_histogram # Filter sentinel values (defensive, in case called directly without prior filtering) self._request_kwargs = _sanitize_sentinel_values(request_kwargs or {}) self._tracer = tracer @@ -1109,6 +1183,20 @@ def _process_chunk(self, chunk): if hasattr(chunk, "response") and chunk.response: self._complete_response_data = chunk.response + def _metric_attributes(self): + return _responses_metric_attributes( + self._instance, + self._traced_data.response_model or self._request_kwargs.get("model"), + is_streaming=True, + ) + + def _record_error_duration(self, exception): + _record_duration( + self._duration_histogram, + self._start_time, + _error_metric_attributes(self._instance, self._request_kwargs.get("model"), exception, True), + ) + @dont_throw def _process_complete_response(self): """Process the complete response and emit span""" @@ -1149,12 +1237,16 @@ def _process_complete_response(self): self._span.set_status(StatusCode.OK) self._span.end() self._cleanup_completed = True + metric_attributes = self._metric_attributes() + _record_duration(self._duration_histogram, self._start_time, metric_attributes) + _record_token_usage(self._token_histogram, self._traced_data.usage, metric_attributes) except Exception as e: if self._span and self._span.is_recording(): self._span.set_attribute(ERROR_TYPE, e.__class__.__name__) self._span.set_status(StatusCode.ERROR, str(e)) self._span.end() + self._record_error_duration(e) self._cleanup_completed = True @dont_throw @@ -1170,6 +1262,7 @@ def _handle_exception(self, exception): self._span.set_status(StatusCode.ERROR, str(exception)) self._span.end() + self._record_error_duration(exception) self._cleanup_completed = True @dont_throw @@ -1185,6 +1278,7 @@ def _ensure_cleanup(self): self._span.set_status(StatusCode.OK) self._span.end() + _record_duration(self._duration_histogram, self._start_time, self._metric_attributes()) self._cleanup_completed = True except Exception: diff --git a/packages/opentelemetry-instrumentation-openai/tests/metrics/test_openai_metrics.py b/packages/opentelemetry-instrumentation-openai/tests/metrics/test_openai_metrics.py index efa7f7f018..0a063b27e6 100644 --- a/packages/opentelemetry-instrumentation-openai/tests/metrics/test_openai_metrics.py +++ b/packages/opentelemetry-instrumentation-openai/tests/metrics/test_openai_metrics.py @@ -1,11 +1,16 @@ +import inspect +import json +import time +import httpx import pytest -from openai import OpenAI +from openai import AsyncOpenAI, InternalServerError, OpenAI from opentelemetry.semconv._incubating.attributes import ( gen_ai_attributes as GenAIAttributes, ) from opentelemetry.semconv._incubating.metrics import ( gen_ai_metrics as GenAIMetrics, ) +from opentelemetry.semconv.attributes.error_attributes import ERROR_TYPE from opentelemetry.semconv_ai import Meters from pydantic import BaseModel @@ -309,3 +314,186 @@ def test_image_gen_metrics(instrument_legacy, reader, openai_client): assert len(data_point.attributes["server.address"]) > 0 assert found_duration_metric + + +_RESPONSES_BODY = { + "object": "response", + "created_at": 0, + "status": "completed", + "model": "gpt-4o-mini", + "output": [{ + "type": "message", "id": "msg_1", "status": "completed", "role": "assistant", + "content": [{"type": "output_text", "text": "hi", "annotations": []}], + }], + "parallel_tool_calls": True, + "tool_choice": "auto", + "tools": [], + "usage": { + "input_tokens": 7, "output_tokens": 3, "total_tokens": 10, + "input_tokens_details": {"cached_tokens": 0}, "output_tokens_details": {"reasoning_tokens": 0}, + }, +} + + +def _responses_ok(request): + # A unique id per call, as already-completed ids are not traced again + body = {**_RESPONSES_BODY, "id": f"resp_metrics_{time.time_ns()}"} + if json.loads(request.content).get("stream"): + event = {"type": "response.completed", "sequence_number": 0, "response": body} + return httpx.Response( + 200, + headers={"content-type": "text/event-stream"}, + content=f"event: response.completed\ndata: {json.dumps(event)}\n\n", + ) + return httpx.Response(200, json=body) + + +def _responses_server_error(request): + return httpx.Response(500, json={"error": {"message": "boom"}}) + + +class _BrokenStream(httpx.SyncByteStream, httpx.AsyncByteStream): + def __iter__(self): + yield from () + raise httpx.ReadError("stream broke") + + async def __aiter__(self): + for chunk in self: + yield chunk + + +def _responses_broken_stream(request): + return httpx.Response(200, headers={"content-type": "text/event-stream"}, stream=_BrokenStream()) + + +def _background_handler(response_id): + def handler(request): + status = "queued" if request.method == "POST" else "completed" + return httpx.Response(200, json={**_RESPONSES_BODY, "id": response_id, "status": status}) + + return handler + + +def _client(is_async, handler): + transport = httpx.MockTransport(handler) + if is_async: + return AsyncOpenAI(api_key="test", max_retries=0, http_client=httpx.AsyncClient(transport=transport)) + return OpenAI(api_key="test", max_retries=0, http_client=httpx.Client(transport=transport)) + + +async def _result(value): + return await value if inspect.isawaitable(value) else value + + +async def _call_responses(is_async, handler, method="create", stream=False): + client = _client(is_async, handler) + kwargs = {"stream": True} if stream else {} + response = await _result(getattr(client.responses, method)(model="gpt-4o-mini", input="hello", **kwargs)) + if stream and is_async: + async for _ in response: + pass + elif stream: + for _ in response: + pass + return client + + +def _assert_shared_attributes(attributes, client, stream): + assert attributes[GenAIAttributes.GEN_AI_OPERATION_NAME] == "chat" + assert attributes[GenAIAttributes.GEN_AI_PROVIDER_NAME] == "openai" + assert attributes["server.address"] == str(client.base_url) + assert attributes["stream"] is stream + + +def _collect_points(reader): + data = reader.get_metrics_data() + points = {} + for rm in data.resource_metrics if data else []: + for sm in rm.scope_metrics: + for metric in sm.metrics: + points.setdefault(metric.name, []).extend(metric.data.data_points) + return points + + +@pytest.mark.asyncio +@pytest.mark.parametrize("method,stream", [("create", False), ("create", True), ("parse", False)]) +@pytest.mark.parametrize("is_async", [False, True]) +async def test_responses_metrics(instrument_legacy, reader, is_async, method, stream): + _collect_points(reader) + + client = await _call_responses(is_async, _responses_ok, method, stream) + + points = _collect_points(reader) + durations = points.get(Meters.LLM_OPERATION_DURATION, []) + assert [p.count for p in durations] == [1] + assert durations[0].sum > 0 + assert durations[0].attributes[GenAIAttributes.GEN_AI_RESPONSE_MODEL] == "gpt-4o-mini" + _assert_shared_attributes(durations[0].attributes, client, stream) + tokens = points.get(Meters.LLM_TOKEN_USAGE, []) + assert {p.attributes[GenAIAttributes.GEN_AI_TOKEN_TYPE]: p.sum for p in tokens} == {"input": 7, "output": 3} + for point in tokens: + assert point.attributes[GenAIAttributes.GEN_AI_RESPONSE_MODEL] == "gpt-4o-mini" + _assert_shared_attributes(point.attributes, client, stream) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "handler,stream,error", + [ + (_responses_server_error, False, InternalServerError), + (_responses_server_error, True, InternalServerError), + (_responses_broken_stream, True, httpx.ReadError), + ], +) +@pytest.mark.parametrize("is_async", [False, True]) +async def test_responses_error_metrics(instrument_legacy, reader, is_async, handler, stream, error): + _collect_points(reader) + + with pytest.raises(error): + await _call_responses(is_async, handler, stream=stream) + + points = _collect_points(reader) + durations = points.get(Meters.LLM_OPERATION_DURATION, []) + assert [p.count for p in durations] == [1] + attributes = durations[0].attributes + assert attributes[ERROR_TYPE] == error.__name__ + assert attributes[GenAIAttributes.GEN_AI_REQUEST_MODEL] == "gpt-4o-mini" + _assert_shared_attributes(attributes, _client(is_async, handler), stream) + assert Meters.LLM_TOKEN_USAGE not in points + + +@pytest.mark.asyncio +@pytest.mark.parametrize("is_async", [False, True]) +async def test_responses_stream_closed_early_records_duration(instrument_legacy, reader, is_async): + _collect_points(reader) + client = _client(is_async, _responses_ok) + + response = await _result(client.responses.create(model="gpt-4o-mini", input="hello", stream=True)) + await _result(response.aclose() if is_async else response.close()) + + durations = _collect_points(reader).get(Meters.LLM_OPERATION_DURATION, []) + assert [p.count for p in durations] == [1] + assert ERROR_TYPE not in durations[0].attributes + _assert_shared_attributes(durations[0].attributes, client, True) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("is_async", [False, True]) +async def test_responses_background_retrieve_metrics(instrument_legacy, reader, is_async): + _collect_points(reader) + response_id = f"resp_background_{time.time_ns()}" + client = _client(is_async, _background_handler(response_id)) + + await _result(client.responses.create(model="gpt-4o-mini", input="hello", background=True)) + assert Meters.LLM_OPERATION_DURATION not in _collect_points(reader) + + await _result(client.responses.retrieve(response_id)) + points = _collect_points(reader) + durations = points.get(Meters.LLM_OPERATION_DURATION, []) + assert [p.count for p in durations] == [1] + _assert_shared_attributes(durations[0].attributes, client, False) + assert sorted(p.sum for p in points.get(Meters.LLM_TOKEN_USAGE, [])) == [3, 7] + + # Retrieving an already-completed response emits no span, so no new operation either + await _result(client.responses.retrieve(response_id)) + assert _collect_points(reader) == {}