fix(groq): record metrics for streaming responses

packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/__init__.py

@@ -180,6 +180,9 @@ def _handle_streaming_response(
     finish_reasons: list[str],
     usage: Union[CompletionUsage, None],
     event_logger: Union[Logger, None],
+    token_histogram: Histogram,
+    duration_histogram: Histogram,
+    duration: float,
 ) -> None:
     # finish_reasons is a list; use first entry for message-level finish_reason
     finish_reason = finish_reasons[0] if finish_reasons else None
@@ -189,8 +192,45 @@ def _handle_streaming_response(
     else:
         set_streaming_response_attributes(span, accumulated_content, finish_reason, tool_calls=tool_calls)
 
+    # Streaming responses never recorded metrics: the usage and duration are
+    # collected here but were only ever set on the span. Emit the same token and
+    # duration histograms as the non-streaming path so streaming calls show up
+    # on dashboards. Attribute shape mirrors span_utils.set_model_response_attributes.
+    if duration_histogram:
+        duration_histogram.record(
+            duration,
+            attributes={
+                GenAIAttributes.GEN_AI_PROVIDER_NAME: GenAIAttributes.GenAiProviderNameValues.GROQ.value,
+                GenAIAttributes.GEN_AI_OPERATION_NAME: GenAIAttributes.GenAiOperationNameValues.CHAT.value,
+            },
+        )
+    if usage and token_histogram:
+        token_histogram.record(
+            usage.prompt_tokens,
+            attributes={
+                GenAIAttributes.GEN_AI_PROVIDER_NAME: GenAIAttributes.GenAiProviderNameValues.GROQ.value,
+                GenAIAttributes.GEN_AI_OPERATION_NAME: GenAIAttributes.GenAiOperationNameValues.CHAT.value,
+                GenAIAttributes.GEN_AI_TOKEN_TYPE: "input",
+            },
+        )
+        token_histogram.record(
+            usage.completion_tokens,
+            attributes={
+                GenAIAttributes.GEN_AI_PROVIDER_NAME: GenAIAttributes.GenAiProviderNameValues.GROQ.value,
+                GenAIAttributes.GEN_AI_OPERATION_NAME: GenAIAttributes.GenAiOperationNameValues.CHAT.value,
+                GenAIAttributes.GEN_AI_TOKEN_TYPE: "output",
+            },
+        )
+
 
-def _create_stream_processor(response, span, event_logger):
+def _create_stream_processor(
+    response,
+    span,
+    event_logger,
+    token_histogram: Histogram = None,
+    duration_histogram: Histogram = None,
+    start_time: float = None,
+):
     """Create a generator that processes a stream while collecting telemetry."""
     accumulated_content = ""
     accumulated_tool_calls: dict = {}
@@ -216,15 +256,30 @@ def _create_stream_processor(response, span, event_logger):
     else:
         tool_calls = [accumulated_tool_calls[i] for i in sorted(accumulated_tool_calls)] or None
         _handle_streaming_response(
-            span, accumulated_content, tool_calls, accumulated_finish_reasons, usage, event_logger
+            span,
+            accumulated_content,
+            tool_calls,
+            accumulated_finish_reasons,
+            usage,
+            event_logger,
+            token_histogram,
+            duration_histogram,
+            time.time() - start_time if start_time is not None else None,
         )
         if span.is_recording():
             span.set_status(Status(StatusCode.OK))
     finally:
         span.end()
 
 
-async def _create_async_stream_processor(response, span, event_logger):
+async def _create_async_stream_processor(
+    response,
+    span,
+    event_logger,
+    token_histogram: Histogram = None,
+    duration_histogram: Histogram = None,
+    start_time: float = None,
+):
     """Create an async generator that processes a stream while collecting telemetry."""
     accumulated_content = ""
     accumulated_tool_calls: dict = {}
@@ -250,7 +305,15 @@ async def _create_async_stream_processor(response, span, event_logger):
     else:
         tool_calls = [accumulated_tool_calls[i] for i in sorted(accumulated_tool_calls)] or None
         _handle_streaming_response(
-            span, accumulated_content, tool_calls, accumulated_finish_reasons, usage, event_logger
+            span,
+            accumulated_content,
+            tool_calls,
+            accumulated_finish_reasons,
+            usage,
+            event_logger,
+            token_histogram,
+            duration_histogram,
+            time.time() - start_time if start_time is not None else None,
         )
         if span.is_recording():
             span.set_status(Status(StatusCode.OK))
@@ -327,7 +390,9 @@ def _wrap(
 
     if is_streaming_response(response):
         try:
-            return _create_stream_processor(response, span, event_logger)
+            return _create_stream_processor(
+                response, span, event_logger, token_histogram, duration_histogram, start_time
+            )
         except Exception as ex:
             logger.warning(
                 "Failed to process streaming response for groq span, error: %s",
@@ -415,7 +480,9 @@ async def _awrap(
 
     if is_streaming_response(response):
         try:
-            return _create_async_stream_processor(response, span, event_logger)
+            return _create_async_stream_processor(
+                response, span, event_logger, token_histogram, duration_histogram, start_time
+            )
         except Exception as ex:
             logger.warning(
                 "Failed to process streaming response for groq span, error: %s",

packages/opentelemetry-instrumentation-groq/tests/traces/test_streaming_metrics.py

@@ -0,0 +1,57 @@
+"""Regression tests for #4419: streaming responses must record token and
+duration metrics.
+
+The streaming path (_create_stream_processor) collected usage from each chunk
+but never recorded it to the token/duration histograms, so streaming calls were
+invisible on metric dashboards. These tests drive the processor with mocked
+chunks and assert the metrics land.
+"""
+
+from unittest.mock import MagicMock
+
+from opentelemetry.instrumentation.groq import _create_stream_processor
+from opentelemetry.semconv_ai import Meters
+
+
+def _chunk(*, prompt_tokens=0, completion_tokens=0):
+    """Mock a Groq streaming chunk carrying x_groq.usage (as the final chunk does)."""
+    chunk = MagicMock()
+    chunk.choices = [MagicMock()]
+    chunk.choices[0].delta.content = ""
+    chunk.choices[0].delta.tool_calls = None
+    chunk.choices[0].finish_reason = None
+    chunk.x_groq = MagicMock()
+    chunk.x_groq.usage.prompt_tokens = prompt_tokens
+    chunk.x_groq.usage.completion_tokens = completion_tokens
+    chunk.x_groq.usage.total_tokens = prompt_tokens + completion_tokens
+    return chunk
+
+
+def _collect_metric_names(reader):
+    data = reader.get_metrics_data()
+    names = set()
+    for rm in data.resource_metrics:
+        for sm in rm.scope_metrics:
+            for metric in sm.metrics:
+                names.add(metric.name)
+    return names
+
+
+def test_streaming_records_token_and_duration_metrics(reader, meter_provider):
+    """Streaming calls record LLM_TOKEN_USAGE and LLM_OPERATION_DURATION (fixes #4419)."""
+
+    meter = meter_provider.get_meter("groq-streaming")
+    token_histogram = meter.create_histogram(name=Meters.LLM_TOKEN_USAGE, unit="token")
+    duration_histogram = meter.create_histogram(name=Meters.LLM_OPERATION_DURATION, unit="s")
+
+    span = MagicMock()
+    span.is_recording.return_value = False
+
+    resp = [_chunk(), _chunk(prompt_tokens=9, completion_tokens=4)]
+    gen = _create_stream_processor(resp, span, None, token_histogram, duration_histogram, 0.0)
+    for _ in gen:
+        pass
+
+    names = _collect_metric_names(reader)
+    assert Meters.LLM_TOKEN_USAGE in names, f"token usage not recorded, got {names}"
+    assert Meters.LLM_OPERATION_DURATION in names, f"operation duration not recorded, got {names}"