From c340a86160997ad4ff15480116c4a3622d088d93 Mon Sep 17 00:00:00 2001 From: CJstate <1507965754@qq.com> Date: Sat, 19 Sep 2026 09:14:39 +0800 Subject: [PATCH] fix(groq): record metrics for streaming responses Streaming (stream=True) responses only updated span attributes but never recorded token usage or operation duration to the histograms, so streaming calls were invisible on metric dashboards. Pass the token/duration histograms into the stream processors and record the accumulated usage (from chunk.x_groq.usage) and elapsed duration once the stream is drained. Non-streaming behavior is unchanged. Fixes #4419 --- .../instrumentation/groq/__init__.py | 79 +++++++++++++++++-- .../tests/traces/test_streaming_metrics.py | 57 +++++++++++++ 2 files changed, 130 insertions(+), 6 deletions(-) create mode 100644 packages/opentelemetry-instrumentation-groq/tests/traces/test_streaming_metrics.py diff --git a/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/__init__.py b/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/__init__.py index 4a275051f2..9d2f720db3 100644 --- a/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/__init__.py +++ b/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,7 +256,15 @@ 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)) @@ -224,7 +272,14 @@ def _create_stream_processor(response, span, event_logger): 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", diff --git a/packages/opentelemetry-instrumentation-groq/tests/traces/test_streaming_metrics.py b/packages/opentelemetry-instrumentation-groq/tests/traces/test_streaming_metrics.py new file mode 100644 index 0000000000..d7842225c0 --- /dev/null +++ b/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}"