Add streaming metrics and attribute

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.

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]

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

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

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

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,
     )
 
 

instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_conformance.py

@@ -19,13 +19,15 @@
 )
 
 from .conformance.inference import InferenceScenario
+from .conformance.inference_streaming import InferenceStreamingScenario
 from .conformance.tool_calling import ToolCallingScenario
 
 
 @pytest.mark.parametrize(
     "scenario",
     [
         InferenceScenario(),
+        InferenceStreamingScenario(),
         ToolCallingScenario(),
     ],
     ids=lambda s: type(s).__name__,

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)

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.

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 <https://pypi.org/project/openai/>`_. 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

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]

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

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