diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/.changelog/269.added b/instrumentation/opentelemetry-instrumentation-genai-anthropic/.changelog/269.added new file mode 100644 index 00000000..4f72152f --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/.changelog/269.added @@ -0,0 +1 @@ +Emit streaming timing metrics (time-to-first-chunk and time-per-output-chunk) for streaming messages. diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/pyproject.toml b/instrumentation/opentelemetry-instrumentation-genai-anthropic/pyproject.toml index 064a89f9..d46358cc 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-anthropic/pyproject.toml +++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/pyproject.toml @@ -28,7 +28,7 @@ dependencies = [ "opentelemetry-api ~= 1.43", "opentelemetry-instrumentation >= 0.64b0, <1", "opentelemetry-semantic-conventions >= 0.64b0, <1", - "opentelemetry-util-genai >= 1.0b0, <2", + "opentelemetry-util-genai >= 1.1b0.dev, <2", ] [project.optional-dependencies] diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/wrappers.py b/instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/wrappers.py index 668a1911..a66468d7 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/wrappers.py +++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/wrappers.py @@ -183,7 +183,7 @@ def __init__( invocation: InferenceInvocation, capture_content: bool, ): - super().__init__(stream) + super().__init__(stream, invocation=invocation) self._self_invocation = invocation self._self_message = None self._self_capture_content = capture_content @@ -225,7 +225,7 @@ def __init__( invocation: InferenceInvocation, capture_content: bool, ): - super().__init__(stream) + super().__init__(stream, invocation=invocation) self._self_invocation = invocation self._self_message = None self._self_capture_content = capture_content diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/conformance/inference_streaming.py b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/conformance/inference_streaming.py new file mode 100644 index 00000000..b665c4b3 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/conformance/inference_streaming.py @@ -0,0 +1,87 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +"""Conformance scenario: anthropic streaming chat (inference). + +Exercises the streaming messages path so the streaming timing metrics +(time-to-first-chunk and time-per-output-chunk) are emitted and validated in +addition to the duration and token usage metrics. +""" + +from __future__ import annotations + +import os +from typing import Any +from unittest import mock + +from anthropic import Anthropic + +from opentelemetry.instrumentation.genai.anthropic import AnthropicInstrumentor +from opentelemetry.sdk._logs import LoggerProvider +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.test.weaver_live_check import LiveCheckReport +from opentelemetry.test_util_genai.conformance import Scenario +from opentelemetry.test_util_genai.instrumentor import instrument + + +class InferenceStreamingScenario(Scenario): + expected_spans = {"chat": 1} + expected_metrics = ( + "gen_ai.client.operation.duration", + "gen_ai.client.token.usage", + "gen_ai.client.operation.time_to_first_chunk", + "gen_ai.client.operation.time_per_output_chunk", + ) + + def validate(self, report: LiveCheckReport) -> None: + super().validate(report) + stream_values = [ + attr["value"] + for entry in report["samples"] + if "span" in entry + for attr in entry["span"]["attributes"] + if attr["name"] == "gen_ai.request.stream" + ] + assert stream_values == [True], ( + "streaming messages should set gen_ai.request.stream=true on the " + f"chat span; saw {stream_values}" + ) + + def run( + self, + *, + tracer_provider: TracerProvider, + meter_provider: MeterProvider, + logger_provider: LoggerProvider, + vcr: Any, + ) -> None: + key_override = ( + {} + if os.getenv("ANTHROPIC_API_KEY") + else {"ANTHROPIC_API_KEY": "test_anthropic_api_key"} + ) + with mock.patch.dict(os.environ, key_override): + with instrument( + AnthropicInstrumentor(), + tracer_provider=tracer_provider, + logger_provider=logger_provider, + meter_provider=meter_provider, + content_capture="SPAN_ONLY", + ): + with vcr.use_cassette( + "test_sync_messages_create_streaming.yaml" + ): + with Anthropic().messages.create( + model="claude-sonnet-4-20250514", + max_tokens=100, + messages=[ + { + "role": "user", + "content": "Say hello in one word.", + } + ], + stream=True, + ) as stream: + for _ in stream: + pass diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/requirements.oldest.txt b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/requirements.oldest.txt index 91ee6363..60421083 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/requirements.oldest.txt +++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/requirements.oldest.txt @@ -21,5 +21,5 @@ # OpenTelemetry SDK and test utilities come transitively from opentelemetry-test-util-genai, which # every oldest env installs. Pin here only test-only deps that nothing else already provides. # -# There is currently nothing to pin: anthropic has no test-only dependency that isn't already -# provided by a declared bound or by the shared test fixtures. +# Drop this once opentelemetry-util-genai 1.1b0 is published. +-e util/opentelemetry-util-genai diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_async_wrappers.py b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_async_wrappers.py index 5aa6d996..d1907fe8 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_async_wrappers.py +++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_async_wrappers.py @@ -19,6 +19,7 @@ def _make_invocation(): request_model=None, stop=lambda: None, fail=lambda error: None, + _on_stream_chunk=lambda chunk_at: None, ) diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_conformance.py b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_conformance.py index 31e58b74..7dbc78a9 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_conformance.py +++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_conformance.py @@ -19,6 +19,7 @@ ) from .conformance.inference import InferenceScenario +from .conformance.inference_streaming import InferenceStreamingScenario from .conformance.tool_calling import ToolCallingScenario @@ -26,6 +27,7 @@ "scenario", [ InferenceScenario(), + InferenceStreamingScenario(), ToolCallingScenario(), ], ids=lambda s: type(s).__name__, diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_metrics.py b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_metrics.py new file mode 100644 index 00000000..dddea8f5 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_metrics.py @@ -0,0 +1,90 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +"""Metric-recording tests for the Anthropic instrumentation.""" + +import pytest + +from opentelemetry.semconv._incubating.attributes import ( + gen_ai_attributes as GenAIAttributes, +) +from opentelemetry.semconv._incubating.metrics import gen_ai_metrics + +MODEL = "claude-sonnet-4-20250514" +MESSAGES = [{"role": "user", "content": "Say hello in one word."}] + + +def _metrics_by_name(metric_reader): + by_name = {} + for rm in metric_reader.get_metrics_data().resource_metrics: + for scope in rm.scope_metrics: + for metric in scope.metrics: + by_name[metric.name] = metric + return by_name + + +def _assert_streaming_timing_metrics(metric_reader): + """Assert the streaming timing metrics are emitted through the real + Anthropic stream wrapper path. + + Regression coverage for the ``invocation=invocation`` wiring in + ``wrappers.py``: dropping it would keep every span/attribute test green but + silently stop emitting TTFC and per-output-chunk metrics for Anthropic + streaming. + """ + metrics = _metrics_by_name(metric_reader) + + ttfc = metrics.get( + gen_ai_metrics.GEN_AI_CLIENT_OPERATION_TIME_TO_FIRST_CHUNK + ) + assert ttfc is not None + ttfc_point = ttfc.data.data_points[0] + assert ttfc_point.count == 1 + assert ttfc_point.sum >= 0 + assert ( + ttfc_point.attributes[GenAIAttributes.GEN_AI_OPERATION_NAME] + == GenAIAttributes.GenAiOperationNameValues.CHAT.value + ) + assert ttfc_point.attributes[GenAIAttributes.GEN_AI_REQUEST_MODEL] == MODEL + + per_chunk = metrics.get( + gen_ai_metrics.GEN_AI_CLIENT_OPERATION_TIME_PER_OUTPUT_CHUNK + ) + assert per_chunk is not None + per_chunk_point = per_chunk.data.data_points[0] + # One record per inter-chunk gap; the streaming cassette has several events. + assert per_chunk_point.count >= 1 + assert per_chunk_point.sum >= 0 + + +def test_sync_messages_streaming_timing_metrics( + metric_reader, anthropic_client, instrument_with_content, vcr +): + with vcr.use_cassette("test_sync_messages_create_streaming.yaml"): + with anthropic_client.messages.create( + model=MODEL, + max_tokens=100, + messages=MESSAGES, + stream=True, + ) as stream: + for _ in stream: + pass + + _assert_streaming_timing_metrics(metric_reader) + + +@pytest.mark.asyncio() +async def test_async_messages_streaming_timing_metrics( + metric_reader, async_anthropic_client, instrument_with_content, vcr +): + with vcr.use_cassette("test_async_messages_create_streaming.yaml"): + async with await async_anthropic_client.messages.create( + model=MODEL, + max_tokens=100, + messages=MESSAGES, + stream=True, + ) as stream: + async for _ in stream: + pass + + _assert_streaming_timing_metrics(metric_reader) diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/.changelog/269.added b/instrumentation/opentelemetry-instrumentation-genai-openai/.changelog/269.added new file mode 100644 index 00000000..9daa5928 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-openai/.changelog/269.added @@ -0,0 +1 @@ +Emit streaming timing metrics (time-to-first-chunk and time-per-output-chunk) for streaming chat completions. diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/README.rst b/instrumentation/opentelemetry-instrumentation-genai-openai/README.rst index 91604da7..adc9b01e 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-openai/README.rst +++ b/instrumentation/opentelemetry-instrumentation-genai-openai/README.rst @@ -8,7 +8,9 @@ OpenTelemetry OpenAI Instrumentation This library allows tracing LLM requests and logging of messages made by the `OpenAI Python API library `_. It also captures -the duration of the operations and the number of tokens used as metrics. +the duration of the operations and the number of tokens used as metrics, and for +streaming chat completions the time to the first chunk and the time between output +chunks. .. note:: This package continues the project previously published as diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/pyproject.toml b/instrumentation/opentelemetry-instrumentation-genai-openai/pyproject.toml index 7b7f2469..117f596a 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-openai/pyproject.toml +++ b/instrumentation/opentelemetry-instrumentation-genai-openai/pyproject.toml @@ -28,7 +28,7 @@ dependencies = [ "opentelemetry-api ~= 1.43", "opentelemetry-instrumentation >= 0.64b0, <1", "opentelemetry-semantic-conventions >= 0.64b0, <1", - "opentelemetry-util-genai >= 1.0b0, <2", + "opentelemetry-util-genai >= 1.1b0.dev, <2", ] [project.optional-dependencies] diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/chat_wrappers.py b/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/chat_wrappers.py index 4c5dfa32..40c1bdd1 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/chat_wrappers.py +++ b/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/chat_wrappers.py @@ -33,17 +33,18 @@ class _ChatStreamMixin: _self_capture_content: bool _self_choice_buffers: list[ChoiceBuffer] _self_response_id: Optional[str] - _self_response_model: Optional[str] _self_service_tier: Optional[str] _self_prompt_tokens: Optional[int] _self_completion_tokens: Optional[int] def _set_response_model(self, chunk: ChatCompletionChunk) -> None: - if self._self_response_model: + # Set eagerly so the per-chunk streaming timing metrics carry + # gen_ai.response.model. + if self._self_invocation.response_model_name: return if chunk.model: - self._self_response_model = chunk.model + self._self_invocation.response_model_name = chunk.model def _set_response_id(self, chunk: ChatCompletionChunk) -> None: if self._self_response_id: @@ -146,7 +147,6 @@ def parse(self) -> _ChatStreamMixin: return self def _cleanup(self, error: Optional[BaseException] = None) -> None: - self._self_invocation.response_model_name = self._self_response_model self._self_invocation.response_id = self._self_response_id self._self_invocation.input_tokens = self._self_prompt_tokens self._self_invocation.output_tokens = self._self_completion_tokens @@ -182,12 +182,11 @@ def __init__( invocation: InferenceInvocation, capture_content: bool, ) -> None: - super().__init__(stream) + super().__init__(stream, invocation=invocation) self._self_invocation = invocation self._self_choice_buffers = [] self._self_capture_content = capture_content self._self_response_id = None - self._self_response_model = None self._self_service_tier = None self._self_prompt_tokens = None self._self_completion_tokens = None @@ -203,12 +202,11 @@ def __init__( invocation: InferenceInvocation, capture_content: bool, ) -> None: - super().__init__(stream) + super().__init__(stream, invocation=invocation) self._self_invocation = invocation self._self_choice_buffers = [] self._self_capture_content = capture_content self._self_response_id = None - self._self_response_model = None self._self_service_tier = None self._self_prompt_tokens = None self._self_completion_tokens = None diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/response_wrappers.py b/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/response_wrappers.py index 8a3f0f11..cd3fd80f 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/response_wrappers.py +++ b/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/response_wrappers.py @@ -224,7 +224,7 @@ def __init__( invocation: "GenAIInvocation", capture_content: bool, ): - SyncStreamWrapper.__init__(self, stream) + SyncStreamWrapper.__init__(self, stream, invocation=invocation) # Marks streams already wrapped by the inner Responses.create path so # Responses.stream manager entry can return them without wrapping twice. self._self_is_response_stream_wrapper = True @@ -334,7 +334,7 @@ def __init__( invocation: "GenAIInvocation", capture_content: bool, ): - AsyncStreamWrapper.__init__(self, stream) + AsyncStreamWrapper.__init__(self, stream, invocation=invocation) # Marks streams already wrapped by the inner AsyncResponses.create path # so AsyncResponses.stream manager entry avoids wrapping twice. self._self_is_response_stream_wrapper = True diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/conformance/inference_streaming.py b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/conformance/inference_streaming.py new file mode 100644 index 00000000..83cf2b52 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/conformance/inference_streaming.py @@ -0,0 +1,74 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +"""Conformance scenario: openai streaming chat completion (inference). + +Exercises the streaming chat path so the streaming timing metrics +(time-to-first-chunk and time-per-output-chunk) are emitted and validated in +addition to the duration and token usage metrics. +""" + +from __future__ import annotations + +from typing import Any + +from openai import OpenAI + +from opentelemetry.instrumentation.genai.openai import OpenAIInstrumentor +from opentelemetry.sdk._logs import LoggerProvider +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.test.weaver_live_check import LiveCheckReport +from opentelemetry.test_util_genai.conformance import Scenario +from opentelemetry.test_util_genai.instrumentor import instrument + + +class InferenceStreamingScenario(Scenario): + expected_spans = {"chat": 1} + expected_metrics = ( + "gen_ai.client.operation.duration", + "gen_ai.client.token.usage", + "gen_ai.client.operation.time_to_first_chunk", + "gen_ai.client.operation.time_per_output_chunk", + ) + + def validate(self, report: LiveCheckReport) -> None: + super().validate(report) + stream_values = [ + attr["value"] + for entry in report["samples"] + if "span" in entry + for attr in entry["span"]["attributes"] + if attr["name"] == "gen_ai.request.stream" + ] + assert stream_values == [True], ( + "streaming chat should set gen_ai.request.stream=true on the chat " + f"span; saw {stream_values}" + ) + + def run( + self, + *, + tracer_provider: TracerProvider, + meter_provider: MeterProvider, + logger_provider: LoggerProvider, + vcr: Any, + ) -> None: + with instrument( + OpenAIInstrumentor(), + tracer_provider=tracer_provider, + logger_provider=logger_provider, + meter_provider=meter_provider, + content_capture="SPAN_ONLY", + ): + with vcr.use_cassette("test_chat_completion_streaming.yaml"): + stream = OpenAI().chat.completions.create( + messages=[ + {"role": "user", "content": "Say this is a test"} + ], + model="gpt-4", + stream=True, + stream_options={"include_usage": True}, + ) + for _ in stream: + pass diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/conformance/responses_streaming.py b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/conformance/responses_streaming.py new file mode 100644 index 00000000..76c93f78 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/conformance/responses_streaming.py @@ -0,0 +1,76 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +"""Conformance scenario: OpenAI Responses API streaming. + +Exercises the streaming Responses path so the streaming timing metrics +(time-to-first-chunk and time-per-output-chunk) are emitted and validated in +addition to the duration and token usage metrics. +""" + +from __future__ import annotations + +from typing import Any + +from openai import OpenAI + +from opentelemetry.instrumentation.genai.openai import OpenAIInstrumentor +from opentelemetry.sdk._logs import LoggerProvider +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.test.weaver_live_check import LiveCheckReport +from opentelemetry.test_util_genai.conformance import Scenario +from opentelemetry.test_util_genai.instrumentor import instrument + +DEFAULT_MODEL = "gpt-4o-mini" + + +class ResponsesStreamingScenario(Scenario): + expected_spans = {"chat": 1} + expected_metrics = ( + "gen_ai.client.operation.duration", + "gen_ai.client.token.usage", + "gen_ai.client.operation.time_to_first_chunk", + "gen_ai.client.operation.time_per_output_chunk", + ) + + def validate(self, report: LiveCheckReport) -> None: + super().validate(report) + stream_values = [ + attr["value"] + for entry in report["samples"] + if "span" in entry + for attr in entry["span"]["attributes"] + if attr["name"] == "gen_ai.request.stream" + ] + assert stream_values == [True], ( + "streaming responses should set gen_ai.request.stream=true on the " + f"chat span; saw {stream_values}" + ) + + def run( + self, + *, + tracer_provider: TracerProvider, + meter_provider: MeterProvider, + logger_provider: LoggerProvider, + vcr: Any, + ) -> None: + with instrument( + OpenAIInstrumentor(), + tracer_provider=tracer_provider, + logger_provider=logger_provider, + meter_provider=meter_provider, + content_capture="SPAN_ONLY", + ): + with vcr.use_cassette( + "test_responses_create_streaming[content_mode0].yaml" + ): + with OpenAI().responses.create( + model=DEFAULT_MODEL, + instructions="You are a helpful assistant.", + input="Say this is a test", + stream=True, + ) as stream: + for _ in stream: + pass diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/requirements.oldest.txt b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/requirements.oldest.txt index af091b44..2a6856ef 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/requirements.oldest.txt +++ b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/requirements.oldest.txt @@ -20,6 +20,9 @@ # factor, so they are NOT pinned here — pyproject.toml is the single source of truth. The # OpenTelemetry SDK and test utilities come transitively from opentelemetry-test-util-genai, which # every oldest env installs. Pin here only test-only deps that nothing else already provides. +# +# Drop this once opentelemetry-util-genai 1.1b0 is published. +-e util/opentelemetry-util-genai numpy==2.3.1 ; python_version >= "3.11" numpy==2.1.0 ; python_version == "3.10" diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_async_responses.py b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_async_responses.py index 4409048a..1dc2fa1e 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_async_responses.py +++ b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_async_responses.py @@ -28,6 +28,7 @@ ) from opentelemetry.util.genai.utils import is_experimental_mode +from .test_responses import assert_responses_streaming_timing_metrics from .test_utils import ( DEFAULT_MODEL, USER_ONLY_EXPECTED_INPUT_MESSAGES, @@ -385,6 +386,27 @@ async def test_async_responses_create_api_error( ) +@pytest.mark.asyncio() +async def test_async_responses_create_streaming_timing_metrics( + metric_reader, async_openai_client, instrument_no_content, vcr +): + _skip_if_not_latest() + + with vcr.use_cassette( + "test_async_responses_create_streaming[content_mode0].yaml" + ): + stream = await async_openai_client.responses.create( + model=DEFAULT_MODEL, + instructions=SYSTEM_INSTRUCTIONS, + input=USER_ONLY_PROMPT[0]["content"], + service_tier="default", + stream=True, + ) + await _collect_completed_response(stream) + + assert_responses_streaming_timing_metrics(metric_reader) + + @pytest.mark.asyncio() async def test_async_responses_create_streaming( span_exporter, async_openai_client, instrument_no_content, vcr diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_chat_metrics.py b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_chat_metrics.py index 15826ac4..362b7a7b 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_chat_metrics.py +++ b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_chat_metrics.py @@ -254,3 +254,97 @@ async def test_async_chat_completion_metrics( assert_all_metric_attributes( output_token_usage, latest_experimental_enabled ) + + +def _metrics_by_name(metric_reader): + resource_metrics = metric_reader.get_metrics_data().resource_metrics + by_name = {} + for rm in resource_metrics: + for scope in rm.scope_metrics: + for metric in scope.metrics: + by_name[metric.name] = metric + return by_name + + +def _assert_streaming_timing_metrics(metric_reader): + """Assert the streaming timing metrics are emitted through the real + ChatStreamWrapper path. + + Regression coverage for the ``invocation=invocation`` wiring in + chat_wrappers.py: dropping it would still pass every util-layer test but + would silently stop emitting TTFC and per-output-chunk metrics for OpenAI + streaming. + """ + metrics = _metrics_by_name(metric_reader) + + ttfc = metrics.get( + gen_ai_metrics.GEN_AI_CLIENT_OPERATION_TIME_TO_FIRST_CHUNK + ) + assert ttfc is not None + ttfc_point = ttfc.data.data_points[0] + # Exactly one time-to-first-chunk record per stream. + assert ttfc_point.count == 1 + assert ttfc_point.sum >= 0 + assert ( + ttfc_point.attributes[GenAIAttributes.GEN_AI_OPERATION_NAME] + == GenAIAttributes.GenAiOperationNameValues.CHAT.value + ) + # response.model is set eagerly during streaming, so it lands on the + # first-chunk metric too. + assert ( + ttfc_point.attributes[GenAIAttributes.GEN_AI_RESPONSE_MODEL] + == "gpt-4-0613" + ) + + per_chunk = metrics.get( + gen_ai_metrics.GEN_AI_CLIENT_OPERATION_TIME_PER_OUTPUT_CHUNK + ) + assert per_chunk is not None + per_chunk_point = per_chunk.data.data_points[0] + # One record per inter-chunk gap; the streaming cassette has several + # chunks, so at least one gap is recorded. + assert per_chunk_point.count >= 1 + assert per_chunk_point.sum >= 0 + assert ( + per_chunk_point.attributes[GenAIAttributes.GEN_AI_RESPONSE_MODEL] + == "gpt-4-0613" + ) + + +def test_chat_completion_streaming_timing_metrics( + metric_reader, openai_client, instrument_with_content, vcr +): + if not is_experimental_mode(): + pytest.skip("streaming timing metrics require experimental semconv") + kwargs = { + "model": "gpt-4", + "messages": USER_ONLY_PROMPT, + "stream": True, + "stream_options": {"include_usage": True}, + } + with vcr.use_cassette("test_chat_completion_streaming.yaml"): + response = openai_client.chat.completions.create(**kwargs) + for _ in response: + pass + + _assert_streaming_timing_metrics(metric_reader) + + +@pytest.mark.asyncio() +async def test_async_chat_completion_streaming_timing_metrics( + metric_reader, async_openai_client, instrument_with_content, vcr +): + if not is_experimental_mode(): + pytest.skip("streaming timing metrics require experimental semconv") + kwargs = { + "model": "gpt-4", + "messages": USER_ONLY_PROMPT, + "stream": True, + "stream_options": {"include_usage": True}, + } + with vcr.use_cassette("test_async_chat_completion_streaming.yaml"): + response = await async_openai_client.chat.completions.create(**kwargs) + async for _ in response: + pass + + _assert_streaming_timing_metrics(metric_reader) diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_conformance.py b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_conformance.py index a54da03a..a34cbc8f 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_conformance.py +++ b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_conformance.py @@ -25,8 +25,10 @@ from .conformance.embedding import EmbeddingScenario from .conformance.inference import InferenceScenario +from .conformance.inference_streaming import InferenceStreamingScenario from .conformance.responses_conversation import ResponsesConversationScenario from .conformance.responses_stream import ResponsesStreamScenario +from .conformance.responses_streaming import ResponsesStreamingScenario from .conformance.tool_calling import ToolCallingScenario @@ -34,10 +36,12 @@ "scenario", [ InferenceScenario(), + InferenceStreamingScenario(), EmbeddingScenario(), ToolCallingScenario(), ResponsesConversationScenario(), ResponsesStreamScenario(), + ResponsesStreamingScenario(), ], ids=lambda s: type(s).__name__, ) diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_response_wrappers.py b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_response_wrappers.py index b85da731..de369c01 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_response_wrappers.py +++ b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_response_wrappers.py @@ -65,10 +65,15 @@ def _noop_fail(error): del error +def _noop_on_stream_chunk(chunk_at): + del chunk_at + + def _make_wrapper(manager): invocation = SimpleNamespace( request_model=None, stop=_noop_stop, + _on_stream_chunk=_noop_on_stream_chunk, fail=_noop_fail, ) return ResponseStreamManagerWrapper( @@ -84,6 +89,7 @@ def _make_stream_wrapper(stream, invocation=None): request_model=None, stop=_noop_stop, fail=_noop_fail, + _on_stream_chunk=_noop_on_stream_chunk, ) return ResponseStreamWrapper( stream=stream, @@ -96,6 +102,7 @@ def _make_async_manager_wrapper(manager): invocation = SimpleNamespace( request_model=None, stop=_noop_stop, + _on_stream_chunk=_noop_on_stream_chunk, fail=_noop_fail, ) return AsyncResponseStreamManagerWrapper( @@ -111,6 +118,7 @@ def _make_async_stream_wrapper(stream, invocation=None): request_model=None, stop=_noop_stop, fail=_noop_fail, + _on_stream_chunk=_noop_on_stream_chunk, ) return AsyncResponseStreamWrapper( stream=stream, @@ -223,6 +231,7 @@ def test_manager_enter_failure_fails_invocation_and_reraises(): invocation = SimpleNamespace( request_model=None, stop=_noop_stop, + _on_stream_chunk=_noop_on_stream_chunk, fail=failures.append, ) wrapper = ResponseStreamManagerWrapper( @@ -321,6 +330,7 @@ async def test_async_manager_enter_failure_fails_invocation_and_reraises(): invocation = SimpleNamespace( request_model=None, stop=_noop_stop, + _on_stream_chunk=_noop_on_stream_chunk, fail=failures.append, ) wrapper = AsyncResponseStreamManagerWrapper( diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_responses.py b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_responses.py index d28ed973..77190b15 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_responses.py +++ b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_responses.py @@ -20,6 +20,7 @@ from opentelemetry.semconv._incubating.attributes import ( server_attributes as ServerAttributes, ) +from opentelemetry.semconv._incubating.metrics import gen_ai_metrics from opentelemetry.util.genai.utils import is_experimental_mode from .test_utils import ( @@ -140,6 +141,42 @@ def _collect_completed_response(stream): return response +def assert_responses_streaming_timing_metrics(metric_reader): + """Assert the streaming timing metrics are emitted through the real + Responses stream wrapper path. + + Regression coverage for the ``invocation=invocation`` wiring in + ``response_wrappers.py``: dropping it would keep every span/attribute test + green but silently stop emitting TTFC and per-output-chunk metrics for the + Responses streaming path. + """ + metrics = {} + for rm in metric_reader.get_metrics_data().resource_metrics: + for scope in rm.scope_metrics: + for metric in scope.metrics: + metrics[metric.name] = metric + + ttfc = metrics.get( + gen_ai_metrics.GEN_AI_CLIENT_OPERATION_TIME_TO_FIRST_CHUNK + ) + assert ttfc is not None + ttfc_point = ttfc.data.data_points[0] + assert ttfc_point.count == 1 + assert ttfc_point.sum >= 0 + assert ( + ttfc_point.attributes[GenAIAttributes.GEN_AI_OPERATION_NAME] + == GenAIAttributes.GenAiOperationNameValues.CHAT.value + ) + + per_chunk = metrics.get( + gen_ai_metrics.GEN_AI_CLIENT_OPERATION_TIME_PER_OUTPUT_CHUNK + ) + assert per_chunk is not None + per_chunk_point = per_chunk.data.data_points[0] + assert per_chunk_point.count >= 1 + assert per_chunk_point.sum >= 0 + + def test_responses_uninstrument_removes_patching( span_exporter, tracer_provider, logger_provider, meter_provider ): @@ -384,6 +421,26 @@ def test_responses_create_api_error( ) +def test_responses_create_streaming_timing_metrics( + metric_reader, openai_client, instrument_no_content, vcr +): + _skip_if_not_latest() + + with vcr.use_cassette( + "test_responses_create_streaming[content_mode0].yaml" + ): + with openai_client.responses.create( + model=DEFAULT_MODEL, + instructions=SYSTEM_INSTRUCTIONS, + input=USER_ONLY_PROMPT[0]["content"], + service_tier="default", + stream=True, + ) as stream: + _collect_completed_response(stream) + + assert_responses_streaming_timing_metrics(metric_reader) + + @pytest.mark.vcr() def test_responses_create_streaming( request, span_exporter, openai_client, instrument_no_content diff --git a/tox.ini b/tox.ini index ab363522..aeaa95af 100644 --- a/tox.ini +++ b/tox.ini @@ -280,7 +280,7 @@ deps = -c {toxinidir}/dev-requirements.txt pyright {[testenv]test_deps} - {toxinidir}/util/opentelemetry-util-genai[upload] + -e {toxinidir}/util/opentelemetry-util-genai[upload] {toxinidir}/instrumentation/opentelemetry-instrumentation-google-genai[instruments] {toxinidir}/instrumentation/opentelemetry-instrumentation-genai-anthropic[instruments] {toxinidir}/instrumentation/opentelemetry-instrumentation-genai-langchain[instruments] diff --git a/util/opentelemetry-util-genai/.changelog/269.added b/util/opentelemetry-util-genai/.changelog/269.added new file mode 100644 index 00000000..66364940 --- /dev/null +++ b/util/opentelemetry-util-genai/.changelog/269.added @@ -0,0 +1 @@ +Add streaming timing metrics (``gen_ai.client.operation.time_to_first_chunk``, ``gen_ai.client.operation.time_per_output_chunk``), the ``gen_ai.response.time_to_first_chunk`` span attribute, and the ``gen_ai.request.stream`` span attribute, set by the shared stream wrappers. diff --git a/util/opentelemetry-util-genai/.changelog/269.fixed b/util/opentelemetry-util-genai/.changelog/269.fixed new file mode 100644 index 00000000..bc0c322e --- /dev/null +++ b/util/opentelemetry-util-genai/.changelog/269.fixed @@ -0,0 +1 @@ +Raise the minimum ``wrapt`` to 1.14.0, the first release with the async ``ObjectProxy`` support the stream wrappers require. diff --git a/util/opentelemetry-util-genai/README.rst b/util/opentelemetry-util-genai/README.rst index b9cdaa7c..0a9f8f97 100644 --- a/util/opentelemetry-util-genai/README.rst +++ b/util/opentelemetry-util-genai/README.rst @@ -12,7 +12,9 @@ Key Components - ``TelemetryHandler`` -- manages LLM invocation lifecycles (spans, metrics, events) - ``InferenceInvocation`` and message types (``Text``, ``Reasoning``, ``Blob``, etc.) -- structured data model for GenAI interactions - ``CompletionHook`` -- protocol for uploading content to external storage (built-in ``fsspec`` support) -- Metrics -- ``gen_ai.client.operation.duration`` and ``gen_ai.client.token.usage`` histograms +- Metrics -- ``gen_ai.client.operation.duration`` and ``gen_ai.client.token.usage`` histograms, plus + the streaming timing histograms ``gen_ai.client.operation.time_to_first_chunk`` and + ``gen_ai.client.operation.time_per_output_chunk`` Usage diff --git a/util/opentelemetry-util-genai/pyproject.toml b/util/opentelemetry-util-genai/pyproject.toml index 2dd40308..64d95697 100644 --- a/util/opentelemetry-util-genai/pyproject.toml +++ b/util/opentelemetry-util-genai/pyproject.toml @@ -28,7 +28,7 @@ dependencies = [ "opentelemetry-instrumentation >= 0.64b0, <1", "opentelemetry-semantic-conventions >= 0.64b0, <1", "opentelemetry-api ~= 1.43", - "wrapt >= 1.0.0, < 3.0.0", + "wrapt >= 1.14.0, < 3.0.0", ] [project.entry-points.opentelemetry_genai_completion_hook] diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_inference_invocation.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_inference_invocation.py index 2647a39f..2e920cb0 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_inference_invocation.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_inference_invocation.py @@ -72,7 +72,7 @@ def __init__( self.input_messages: list[InputMessage] = [] self.output_messages: list[OutputMessage] = [] self.system_instruction: list[MessagePart] = [] - self.response_model_name: str | None = None + self._response_model_name: str | None = None self.response_id: str | None = None self.finish_reasons: list[str] | None = None self.input_tokens: int | None = None @@ -92,8 +92,19 @@ def __init__( self.top_k: float | None = None self.request_choice_count: int | None = None self.output_type: str | None = None + self._cached_metric_attributes: dict[str, AttributeValue] | None = None self._start(self._get_base_attributes()) + @property + def response_model_name(self) -> str | None: + return self._response_model_name + + @response_model_name.setter + def response_model_name(self, value: str | None) -> None: + if value != self._response_model_name: + self._response_model_name = value + self._cached_metric_attributes = None + def _get_message_attributes( self, *, for_span: bool ) -> dict[str, AttributeValue]: @@ -138,6 +149,7 @@ def _get_attributes(self) -> dict[str, AttributeValue]: self.thinking_tokens or 0 ) optional_attrs = ( + (GenAI.GEN_AI_REQUEST_STREAM, self._request_stream), (GenAI.GEN_AI_REQUEST_TEMPERATURE, self.temperature), (GenAI.GEN_AI_REQUEST_TOP_P, self.top_p), (GenAI.GEN_AI_REQUEST_TOP_K, self.top_k), @@ -165,16 +177,29 @@ def _get_attributes(self) -> dict[str, AttributeValue]: GenAI.GEN_AI_USAGE_REASONING_OUTPUT_TOKENS, self.thinking_tokens, ), + ( + GenAI.GEN_AI_RESPONSE_TIME_TO_FIRST_CHUNK, + self._ttfc_seconds, + ), ) attrs.update({k: v for k, v in optional_attrs if v is not None}) return attrs def _get_metric_attributes(self) -> dict[str, AttributeValue]: - attrs = self._get_base_attributes() - if self.response_model_name is not None: - attrs[GenAI.GEN_AI_RESPONSE_MODEL] = self.response_model_name - attrs.update(self.metric_attributes) - return attrs + # Cached because this is called once per streaming chunk; invalidated + # by the ``response_model_name`` setter and ``_apply_error_attributes``. + if self._cached_metric_attributes is None: + attrs = self._get_base_attributes() + if self._response_model_name is not None: + attrs[GenAI.GEN_AI_RESPONSE_MODEL] = self._response_model_name + attrs.update(self.metric_attributes) + self._cached_metric_attributes = attrs + return self._cached_metric_attributes + + def _apply_error_attributes(self, error: Error) -> None: + super()._apply_error_attributes(error) + # error.type was added to metric_attributes; invalidate the cache. + self._cached_metric_attributes = None def _get_metric_token_counts(self) -> dict[str, int]: counts: dict[str, int] = {} diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py index 5fd921fc..4e8e5ad2 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py @@ -83,7 +83,14 @@ def __init__( self._span_name: str = span_name self._span_kind: SpanKind = span_kind self._context_token: ContextToken | None = None - self._monotonic_start_s: float | None = None + self._monotonic_start_s: float + # Streaming state, set when the invocation is handed to a stream + # wrapper. ``_request_stream`` marks the request as streamed + # (gen_ai.request.stream); the timing fields are populated by + # ``_on_stream_chunk`` as each chunk arrives. + self._request_stream: bool | None = None + self._ttfc_seconds: float | None = None + self._stream_last_chunk_at: float | None = None def _start( self, attributes: dict[str, AttributeValue] | None = None @@ -110,6 +117,37 @@ def _get_metric_token_counts(self) -> dict[str, int]: # pylint: disable=no-self """Return {token_type: count} for token histogram recording.""" return {} + def _on_stream_chunk(self, chunk_at: float) -> None: + """Record streaming timing for one output chunk as it arrives. + + The first chunk's delta from the invocation start is the + time-to-first-chunk; each later chunk's delta from the previous one is + the inter-chunk gap. Called by the stream wrapper for any invocation + type handed to it. + """ + last_chunk_at = ( + self._stream_last_chunk_at + if self._stream_last_chunk_at is not None + else self._monotonic_start_s + ) + + self._stream_last_chunk_at = chunk_at + delta = max(chunk_at - last_chunk_at, 0.0) + attributes = self._get_metric_attributes() + if self._ttfc_seconds is None: + self._ttfc_seconds = delta + self._metrics_recorder.record_time_to_first_chunk( + delta, + attributes=attributes, + context=self._span_context, + ) + else: + self._metrics_recorder.record_time_per_chunk( + delta, + attributes=attributes, + context=self._span_context, + ) + def _apply_error_attributes(self, error: Error) -> None: """Apply error status and error.type attribute to the span, events, and metrics.""" error_type = error.type.__qualname__ diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/instruments.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/instruments.py index b79d4472..c697e89d 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/instruments.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/instruments.py @@ -55,3 +55,21 @@ def create_token_histogram(meter: Meter) -> Histogram: unit="{token}", explicit_bucket_boundaries_advisory=_GEN_AI_CLIENT_TOKEN_USAGE_BUCKETS, ) + + +def create_time_to_first_chunk_histogram(meter: Meter) -> Histogram: + return meter.create_histogram( + name=gen_ai_metrics.GEN_AI_CLIENT_OPERATION_TIME_TO_FIRST_CHUNK, + description="Time to receive the first chunk, measured from when the client issues the generation request to when the first chunk is received in the response stream.", + unit="s", + explicit_bucket_boundaries_advisory=_GEN_AI_CLIENT_OPERATION_DURATION_BUCKETS, + ) + + +def create_time_per_output_chunk_histogram(meter: Meter) -> Histogram: + return meter.create_histogram( + name=gen_ai_metrics.GEN_AI_CLIENT_OPERATION_TIME_PER_OUTPUT_CHUNK, + description="Time per output chunk, recorded for each chunk received after the first one, measured as the time elapsed from the end of the previous chunk to the end of the current chunk.", + unit="s", + explicit_bucket_boundaries_advisory=_GEN_AI_CLIENT_OPERATION_DURATION_BUCKETS, + ) diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/metrics.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/metrics.py index 7d0ad944..57796c77 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/metrics.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/metrics.py @@ -6,45 +6,50 @@ from __future__ import annotations import timeit -from typing import Optional +from opentelemetry.context import Context from opentelemetry.metrics import Histogram, Meter from opentelemetry.semconv._incubating.attributes import ( gen_ai_attributes as GenAI, ) from opentelemetry.util.genai.instruments import ( create_duration_histogram, + create_time_per_output_chunk_histogram, + create_time_to_first_chunk_histogram, create_token_histogram, ) +from opentelemetry.util.types import Attributes from ._invocation import GenAIInvocation class InvocationMetricsRecorder: - """Records duration and token usage histograms for GenAI invocations.""" + """Records duration, token usage, and streaming timing histograms for GenAI invocations.""" def __init__(self, meter: Meter): self._duration_histogram: Histogram = create_duration_histogram(meter) self._token_histogram: Histogram = create_token_histogram(meter) + self._time_to_first_chunk_histogram: Histogram = ( + create_time_to_first_chunk_histogram(meter) + ) + self._time_per_output_chunk_histogram: Histogram = ( + create_time_per_output_chunk_histogram(meter) + ) def record(self, invocation: GenAIInvocation) -> None: """Record duration and token metrics for an invocation if possible.""" attributes = invocation._get_metric_attributes() token_counts = invocation._get_metric_token_counts() - duration_seconds: Optional[float] = None - if invocation._monotonic_start_s is not None: - duration_seconds = max( - timeit.default_timer() - invocation._monotonic_start_s, - 0.0, - ) - - if duration_seconds is not None: - self._duration_histogram.record( - duration_seconds, - attributes=attributes, - context=invocation._span_context, - ) + duration_seconds = max( + timeit.default_timer() - invocation._monotonic_start_s, + 0.0, + ) + self._duration_histogram.record( + duration_seconds, + attributes=attributes, + context=invocation._span_context, + ) for token_type, token_count in token_counts.items(): self._token_histogram.record( @@ -53,5 +58,33 @@ def record(self, invocation: GenAIInvocation) -> None: context=invocation._span_context, ) + def record_time_to_first_chunk( + self, + ttfc_seconds: float, + *, + attributes: Attributes = None, + context: Context | None = None, + ) -> None: + """Record the streaming time-to-first-chunk.""" + self._time_to_first_chunk_histogram.record( + ttfc_seconds, + attributes=attributes, + context=context, + ) + + def record_time_per_chunk( + self, + gap_seconds: float, + *, + attributes: Attributes = None, + context: Context | None = None, + ) -> None: + """Record one streaming inter-chunk gap.""" + self._time_per_output_chunk_histogram.record( + gap_seconds, + attributes=attributes, + context=context, + ) + __all__ = ["InvocationMetricsRecorder"] diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/stream.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/stream.py index 83bb0a49..ecef4101 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/stream.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/stream.py @@ -4,6 +4,7 @@ from __future__ import annotations import logging +import timeit from abc import ABCMeta, abstractmethod from types import TracebackType from typing import ( @@ -17,6 +18,7 @@ ) if TYPE_CHECKING: + from opentelemetry.util.genai._invocation import GenAIInvocation class _ObjectProxy: def __init__(self, wrapped: object) -> None: ... @@ -64,11 +66,20 @@ class SyncStreamWrapper( internally by the wrapper lifecycle and are not part of the public API. """ - def __init__(self, stream: _SyncStream[ChunkT]): + def __init__( + self, + stream: _SyncStream[ChunkT], + invocation: GenAIInvocation | None = None, + ): super().__init__(stream) self._self_stream = stream self._self_iterator = iter(stream) self._self_finalized = False + # Marks the request as streamed (gen_ai.request.stream) and receives + # per-chunk timing via _on_stream_chunk. + self._self_invocation = invocation + if invocation is not None: + invocation._request_stream = True def __enter__(self): return self @@ -116,7 +127,12 @@ def __next__(self) -> ChunkT: except Exception as error: self._finalize_failure(error) raise + invocation = self._self_invocation + chunk_at = timeit.default_timer() if invocation is not None else None self._process_chunk(chunk) + # Record after _process_chunk so response.model is on the metrics. + if invocation is not None and chunk_at is not None: + invocation._on_stream_chunk(chunk_at) return chunk def _finalize_success(self) -> None: @@ -163,11 +179,20 @@ class AsyncStreamWrapper( are owned by this base class. """ - def __init__(self, stream: _AsyncStream[ChunkT]): + def __init__( + self, + stream: _AsyncStream[ChunkT], + invocation: GenAIInvocation | None = None, + ): super().__init__(stream) self._self_stream = stream self._self_aiter = aiter(stream) self._self_finalized = False + # Marks the request as streamed (gen_ai.request.stream) and receives + # per-chunk timing via _on_stream_chunk. + self._self_invocation = invocation + if invocation is not None: + invocation._request_stream = True async def __aenter__(self): return self @@ -222,7 +247,12 @@ async def __anext__(self) -> ChunkT: self._finalize_failure(error) raise + invocation = self._self_invocation + chunk_at = timeit.default_timer() if invocation is not None else None self._process_chunk(chunk) + # Record after _process_chunk so response.model is on the metrics. + if invocation is not None and chunk_at is not None: + invocation._on_stream_chunk(chunk_at) return chunk def _finalize_success(self) -> None: diff --git a/util/opentelemetry-util-genai/tests/test_handler_metrics.py b/util/opentelemetry-util-genai/tests/test_handler_metrics.py index d5eaa9b0..b528e1eb 100644 --- a/util/opentelemetry-util-genai/tests/test_handler_metrics.py +++ b/util/opentelemetry-util-genai/tests/test_handler_metrics.py @@ -156,6 +156,119 @@ def test_fail_llm_records_error_and_available_tokens(self) -> None: ) self.assertAlmostEqual(token_point.sum, 11.0, places=3) + def test_streaming_records_ttfc_and_per_output_chunk(self) -> None: + handler = TelemetryHandler( + tracer_provider=self.tracer_provider, + meter_provider=self.meter_provider, + ) + with patch("timeit.default_timer", return_value=1000.0): + invocation = handler.inference("prov", request_model="model") + invocation.response_model_name = "model-2025" + + # First chunk 0.35s after start -> TTFC. Later chunks -> inter-chunk + # gaps of 0.05, 0.08, 0.12. + for chunk_at in (1000.35, 1000.40, 1000.48, 1000.60): + invocation._on_stream_chunk(chunk_at) + + with patch("timeit.default_timer", return_value=1002.0): + invocation.stop() + + metrics = self._harvest_metrics() + + self.assertIn("gen_ai.client.operation.time_to_first_chunk", metrics) + ttfc_points = metrics["gen_ai.client.operation.time_to_first_chunk"] + self.assertEqual(len(ttfc_points), 1) + ttfc_point = ttfc_points[0] + self.assertEqual(ttfc_point.count, 1) + self.assertAlmostEqual(ttfc_point.sum, 0.35, places=6) + self.assertEqual( + ttfc_point.attributes[GenAI.GEN_AI_RESPONSE_MODEL], "model-2025" + ) + self.assertEqual( + ttfc_point.attributes[GenAI.GEN_AI_REQUEST_MODEL], "model" + ) + self.assertEqual( + ttfc_point.attributes[GenAI.GEN_AI_PROVIDER_NAME], "prov" + ) + + self.assertIn("gen_ai.client.operation.time_per_output_chunk", metrics) + chunk_points = metrics["gen_ai.client.operation.time_per_output_chunk"] + self.assertEqual(len(chunk_points), 1) + chunk_point = chunk_points[0] + # One data point per inter-chunk gap (3 gaps for 4 chunks). + self.assertEqual(chunk_point.count, 3) + self.assertAlmostEqual(chunk_point.sum, 0.25, places=6) + self.assertEqual( + chunk_point.attributes[GenAI.GEN_AI_RESPONSE_MODEL], "model-2025" + ) + + def test_streaming_records_ttfc_span_attribute(self) -> None: + handler = TelemetryHandler( + tracer_provider=self.tracer_provider, + meter_provider=self.meter_provider, + ) + with patch("timeit.default_timer", return_value=1000.0): + invocation = handler.inference("prov", request_model="model") + + # First chunk 0.42s after the start (1000.0) yields TTFC = 0.42. + invocation._on_stream_chunk(1000.42) + + with patch("timeit.default_timer", return_value=1002.0): + invocation.stop() + + (span,) = self.get_finished_spans() + self.assertAlmostEqual( + span.attributes[GenAI.GEN_AI_RESPONSE_TIME_TO_FIRST_CHUNK], + 0.42, + places=6, + ) + + def test_streamed_request_sets_stream_span_attribute(self) -> None: + handler = TelemetryHandler( + tracer_provider=self.tracer_provider, + meter_provider=self.meter_provider, + ) + invocation = handler.inference("prov", request_model="model") + # Set by the stream wrapper when the invocation is streamed. + invocation._request_stream = True + invocation.stop() + + (span,) = self.get_finished_spans() + self.assertIs(span.attributes[GenAI.GEN_AI_REQUEST_STREAM], True) + + def test_non_streamed_request_omits_stream_span_attribute(self) -> None: + handler = TelemetryHandler( + tracer_provider=self.tracer_provider, + meter_provider=self.meter_provider, + ) + invocation = handler.inference("prov", request_model="model") + invocation.stop() + + (span,) = self.get_finished_spans() + self.assertNotIn(GenAI.GEN_AI_REQUEST_STREAM, span.attributes) + + def test_no_streaming_timing_metrics_without_chunks(self) -> None: + handler = TelemetryHandler( + tracer_provider=self.tracer_provider, + meter_provider=self.meter_provider, + ) + with patch("timeit.default_timer", return_value=1000.0): + invocation = handler.inference("prov", request_model="model") + with patch("timeit.default_timer", return_value=1002.0): + invocation.stop() + + metrics = self._harvest_metrics() + self.assertNotIn( + "gen_ai.client.operation.time_to_first_chunk", metrics + ) + self.assertNotIn( + "gen_ai.client.operation.time_per_output_chunk", metrics + ) + (span,) = self.get_finished_spans() + self.assertNotIn( + GenAI.GEN_AI_RESPONSE_TIME_TO_FIRST_CHUNK, span.attributes + ) + def _harvest_metrics( self, ) -> Dict[str, List[Any]]: diff --git a/util/opentelemetry-util-genai/tests/test_stream.py b/util/opentelemetry-util-genai/tests/test_stream.py index 31ac9a1d..93140f61 100644 --- a/util/opentelemetry-util-genai/tests/test_stream.py +++ b/util/opentelemetry-util-genai/tests/test_stream.py @@ -5,6 +5,8 @@ import asyncio import inspect +import timeit +from unittest.mock import patch import pytest @@ -386,3 +388,159 @@ async def exercise(): assert not wrapper._self_failures asyncio.run(exercise()) + + +# --- Streaming timing seam tests --- +# +# The wrapper does not compute TTFC/gaps itself: when an invocation is passed +# it reports each chunk's arrival time via invocation._on_stream_chunk, and the +# invocation turns those timestamps into metrics (covered by +# test_handler_metrics). These tests cover the wrapper's side of that seam. + + +class _FakeTimingInvocation: + """Minimal stand-in for InferenceInvocation's timing seam.""" + + def __init__(self): + self.chunk_times = [] + self._request_stream = None + + def _on_stream_chunk(self, chunk_at): + self.chunk_times.append(chunk_at) + + +class _TimingSyncWrapper(SyncStreamWrapper): + def __init__(self, stream, invocation=None, process_hook=None): + super().__init__(stream, invocation=invocation) + self._self_processed = [] + self._self_process_hook = process_hook + + def _process_chunk(self, chunk): + self._self_processed.append(chunk) + if self._self_process_hook is not None: + self._self_process_hook() + + def _on_stream_end(self): + pass + + def _on_stream_error(self, error): + pass + + +class _TimingAsyncWrapper(AsyncStreamWrapper): + def __init__(self, stream, invocation=None): + super().__init__(stream, invocation=invocation) + self._self_processed = [] + + def _process_chunk(self, chunk): + self._self_processed.append(chunk) + + def _on_stream_end(self): + pass + + def _on_stream_error(self, error): + pass + + +def test_sync_wrapper_reports_each_chunk_arrival(): + invocation = _FakeTimingInvocation() + stream = _FakeSyncStream(chunks=["a", "b", "c"]) + wrapper = _TimingSyncWrapper(stream, invocation=invocation) + + with patch( + "timeit.default_timer", side_effect=iter([101.2, 101.8, 102.1]) + ): + assert list(wrapper) == ["a", "b", "c"] + + assert invocation.chunk_times == pytest.approx([101.2, 101.8, 102.1]) + + +def test_sync_wrapper_marks_request_stream(): + invocation = _FakeTimingInvocation() + # The wrapper marks the request as streamed at construction, before any + # chunk is read. + _TimingSyncWrapper(_FakeSyncStream(chunks=["a"]), invocation=invocation) + assert invocation._request_stream is True + + +def test_sync_wrapper_single_chunk_one_report(): + invocation = _FakeTimingInvocation() + stream = _FakeSyncStream(chunks=["only"]) + wrapper = _TimingSyncWrapper(stream, invocation=invocation) + + with patch("timeit.default_timer", side_effect=iter([60.5])): + assert list(wrapper) == ["only"] + + assert invocation.chunk_times == pytest.approx([60.5]) + + +def test_sync_wrapper_without_invocation_skips_timing(): + stream = _FakeSyncStream(chunks=["a", "b"]) + wrapper = _TimingSyncWrapper(stream) + + with patch("timeit.default_timer") as timer: + assert list(wrapper) == ["a", "b"] + + timer.assert_not_called() + + +def test_sync_wrapper_error_before_first_chunk_no_report(): + invocation = _FakeTimingInvocation() + stream = _FakeSyncStream(error=RuntimeError("network")) + wrapper = _TimingSyncWrapper(stream, invocation=invocation) + + with pytest.raises(RuntimeError, match="network"): + next(wrapper) + + assert invocation.chunk_times == [] + + +def test_sync_wrapper_captures_arrival_before_processing(): + """Arrival time is taken before _process_chunk, so per-chunk timing + excludes the instrumentation's own processing of the chunk.""" + invocation = _FakeTimingInvocation() + stream = _FakeSyncStream(chunks=["a"]) + # First clock read is the chunk arrival (100.0); the second is consumed + # inside _process_chunk to simulate processing taking time. + times = iter([100.0, 100.9]) + wrapper = _TimingSyncWrapper( + stream, + invocation=invocation, + process_hook=lambda: timeit.default_timer(), + ) + + with patch("timeit.default_timer", side_effect=times): + list(wrapper) + + assert invocation.chunk_times == pytest.approx([100.0]) + + +def test_async_wrapper_reports_each_chunk_arrival(): + async def exercise(): + invocation = _FakeTimingInvocation() + stream = _FakeAsyncStream(chunks=["x", "y", "z"]) + wrapper = _TimingAsyncWrapper(stream, invocation=invocation) + + with patch( + "timeit.default_timer", side_effect=iter([201.3, 202.0, 202.2]) + ): + chunks = [chunk async for chunk in wrapper] + + assert chunks == ["x", "y", "z"] + assert invocation.chunk_times == pytest.approx([201.3, 202.0, 202.2]) + + asyncio.run(exercise()) + + +def test_async_wrapper_without_invocation_skips_timing(): + async def exercise(): + stream = _FakeAsyncStream(chunks=["a", "b"]) + wrapper = _TimingAsyncWrapper(stream) + + with patch("timeit.default_timer") as timer: + chunks = [chunk async for chunk in wrapper] + + assert chunks == ["a", "b"] + timer.assert_not_called() + + asyncio.run(exercise()) diff --git a/uv.lock b/uv.lock index 94f9b8f1..d34ddd3d 100644 --- a/uv.lock +++ b/uv.lock @@ -1465,7 +1465,7 @@ requires-dist = [ { name = "opentelemetry-instrumentation", specifier = ">=0.64b0,<1" }, { name = "opentelemetry-semantic-conventions", specifier = ">=0.64b0,<1" }, { name = "pytest", marker = "extra == 'test'", specifier = ">=7.0.0" }, - { name = "wrapt", specifier = ">=1.0.0,<3.0.0" }, + { name = "wrapt", specifier = ">=1.14.0,<3.0.0" }, ] provides-extras = ["test", "upload"]