From ed3adfb878522174d5a1093163918ca818582b97 Mon Sep 17 00:00:00 2001 From: chengwudi1 Date: Thu, 20 Aug 2026 22:53:57 +0800 Subject: [PATCH 1/4] fix(groq): record token usage and duration metrics for streaming responses --- .../instrumentation/groq/__init__.py | 81 ++++++- .../instrumentation/groq/utils.py | 10 + .../tests/{traces => }/conftest.py | 0 .../tests/metrics/__init__.py | 0 .../metrics/test_groq_streaming_metrics.py | 227 ++++++++++++++++++ 5 files changed, 314 insertions(+), 4 deletions(-) rename packages/opentelemetry-instrumentation-groq/tests/{traces => }/conftest.py (100%) create mode 100644 packages/opentelemetry-instrumentation-groq/tests/metrics/__init__.py create mode 100644 packages/opentelemetry-instrumentation-groq/tests/metrics/test_groq_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..666e551d84 100644 --- a/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/__init__.py +++ b/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/__init__.py @@ -26,6 +26,7 @@ error_metrics_attributes, shared_metrics_attributes, should_emit_events, + streaming_metrics_attributes, ) from opentelemetry.instrumentation.groq.version import __version__ from opentelemetry.instrumentation.instrumentor import BaseInstrumentor @@ -190,7 +191,53 @@ def _handle_streaming_response( set_streaming_response_attributes(span, accumulated_content, finish_reason, tool_calls=tool_calls) -def _create_stream_processor(response, span, event_logger): +def _record_streaming_metrics( + usage, + token_histogram, + duration_histogram, + start_time, + llm_model, +) -> None: + """Record token usage and duration metrics for a consumed streaming response. + + Called once the stream is fully drained; the final chunk carries usage data. + """ + metric_attributes = streaming_metrics_attributes(llm_model) + + if duration_histogram and start_time is not None: + duration_histogram.record( + time.time() - start_time, + attributes=metric_attributes, + ) + + if usage and token_histogram: + if usage.prompt_tokens is not None: + token_histogram.record( + usage.prompt_tokens, + attributes={ + **metric_attributes, + GenAIAttributes.GEN_AI_TOKEN_TYPE: "input", + }, + ) + if usage.completion_tokens is not None: + token_histogram.record( + usage.completion_tokens, + attributes={ + **metric_attributes, + GenAIAttributes.GEN_AI_TOKEN_TYPE: "output", + }, + ) + + +def _create_stream_processor( + response, + span, + event_logger, + token_histogram=None, + duration_histogram=None, + start_time=None, + llm_model=None, +): """Create a generator that processes a stream while collecting telemetry.""" accumulated_content = "" accumulated_tool_calls: dict = {} @@ -218,13 +265,22 @@ def _create_stream_processor(response, span, event_logger): _handle_streaming_response( span, accumulated_content, tool_calls, accumulated_finish_reasons, usage, event_logger ) + _record_streaming_metrics(usage, token_histogram, duration_histogram, start_time, llm_model) 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=None, + duration_histogram=None, + start_time=None, + llm_model=None, +): """Create an async generator that processes a stream while collecting telemetry.""" accumulated_content = "" accumulated_tool_calls: dict = {} @@ -252,6 +308,7 @@ async def _create_async_stream_processor(response, span, event_logger): _handle_streaming_response( span, accumulated_content, tool_calls, accumulated_finish_reasons, usage, event_logger ) + _record_streaming_metrics(usage, token_histogram, duration_histogram, start_time, llm_model) if span.is_recording(): span.set_status(Status(StatusCode.OK)) finally: @@ -327,7 +384,15 @@ 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, + llm_model, + ) except Exception as ex: logger.warning( "Failed to process streaming response for groq span, error: %s", @@ -415,7 +480,15 @@ 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, + llm_model, + ) except Exception as ex: logger.warning( "Failed to process streaming response for groq span, error: %s", diff --git a/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/utils.py b/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/utils.py index 9d105d5242..07aab456a1 100644 --- a/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/utils.py +++ b/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/utils.py @@ -63,6 +63,16 @@ def shared_metrics_attributes(response): } +@dont_throw +def streaming_metrics_attributes(model): + """Metrics attributes for streaming responses, where no full response object exists yet.""" + return { + **Config.get_common_metrics_attributes(), + GenAIAttributes.GEN_AI_PROVIDER_NAME: GenAIAttributes.GenAiProviderNameValues.GROQ.value, + GenAIAttributes.GEN_AI_RESPONSE_MODEL: model, + } + + @dont_throw def error_metrics_attributes(exception): return { diff --git a/packages/opentelemetry-instrumentation-groq/tests/traces/conftest.py b/packages/opentelemetry-instrumentation-groq/tests/conftest.py similarity index 100% rename from packages/opentelemetry-instrumentation-groq/tests/traces/conftest.py rename to packages/opentelemetry-instrumentation-groq/tests/conftest.py diff --git a/packages/opentelemetry-instrumentation-groq/tests/metrics/__init__.py b/packages/opentelemetry-instrumentation-groq/tests/metrics/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/packages/opentelemetry-instrumentation-groq/tests/metrics/test_groq_streaming_metrics.py b/packages/opentelemetry-instrumentation-groq/tests/metrics/test_groq_streaming_metrics.py new file mode 100644 index 0000000000..1d0354e15b --- /dev/null +++ b/packages/opentelemetry-instrumentation-groq/tests/metrics/test_groq_streaming_metrics.py @@ -0,0 +1,227 @@ +"""Unit tests for streaming metrics in the Groq instrumentation. + +Covers the fix for #4419: streaming calls now record token usage and +duration metrics once the stream is fully drained. +All tests use fake streams — no network calls, no cassettes. +""" + +from types import SimpleNamespace + +import pytest + +from opentelemetry.instrumentation.groq import ( + _create_async_stream_processor, + _create_stream_processor, +) +from opentelemetry.instrumentation.groq.utils import streaming_metrics_attributes +from opentelemetry.semconv._incubating.attributes import gen_ai_attributes as GenAIAttributes +from opentelemetry.semconv_ai import Meters +from opentelemetry.trace import get_tracer + + +def _chunk(content="", finish_reason=None, usage=None): + return SimpleNamespace( + choices=[ + SimpleNamespace( + delta=SimpleNamespace(content=content, tool_calls=None), + finish_reason=finish_reason, + ) + ], + x_groq=SimpleNamespace(usage=usage) if usage else None, + ) + + +def _usage(prompt=10, completion=5): + return SimpleNamespace( + prompt_tokens=prompt, + completion_tokens=completion, + total_tokens=prompt + completion, + ) + + +class _FakeStream: + def __init__(self, chunks): + self._chunks = chunks + + def __iter__(self): + return iter(self._chunks) + + +class _FakeAsyncStream: + def __init__(self, chunks): + self._chunks = chunks + + async def __aiter__(self): + for chunk in self._chunks: + yield chunk + + +def _find_metric(metrics_data, name): + """Return the first metric whose instrument name matches, or None.""" + if metrics_data is None: + return None + for resource_metric in metrics_data.resource_metrics: + for scope_metric in resource_metric.scope_metrics: + for metric in scope_metric.metrics: + if metric.name == name: + return metric + return None + + +def _find_data_point(metric, attributes_subset): + """Return the first data point whose attributes contain the given subset.""" + for dp in metric.data.data_points: + dp_attrs = dict(dp.attributes) + if all(dp_attrs.get(k) == v for k, v in attributes_subset.items()): + return dp + return None + + +# --------------------------------------------------------------------------- +# streaming_metrics_attributes +# --------------------------------------------------------------------------- + + +class TestStreamingMetricsAttributes: + def test_returns_provider_and_model(self): + attrs = streaming_metrics_attributes("llama-3.3-70b-versatile") + assert attrs[GenAIAttributes.GEN_AI_PROVIDER_NAME] == "groq" + assert attrs[GenAIAttributes.GEN_AI_RESPONSE_MODEL] == "llama-3.3-70b-versatile" + + +# --------------------------------------------------------------------------- +# _create_stream_processor (sync) +# --------------------------------------------------------------------------- + + +class TestStreamProcessorMetrics: + def test_records_token_and_duration_metrics(self, reader, tracer_provider, meter_provider): + span = get_tracer("test", tracer_provider=tracer_provider).start_span("chat llama-3.3-70b-versatile") + token_histogram = meter_provider.get_meter("test").create_histogram(name=Meters.LLM_TOKEN_USAGE) + duration_histogram = meter_provider.get_meter("test").create_histogram(name=Meters.LLM_OPERATION_DURATION) + + usage = _usage(prompt=10, completion=5) + stream = _FakeStream( + [ + _chunk(content="hello"), + _chunk(content=" world", finish_reason="stop", usage=usage), + ] + ) + + processor = _create_stream_processor( + stream, + span, + None, + token_histogram, + duration_histogram, + start_time=0, + llm_model="llama-3.3-70b-versatile", + ) + for _ in processor: + pass + + metrics_data = reader.get_metrics_data() + + token_metric = _find_metric(metrics_data, Meters.LLM_TOKEN_USAGE) + assert token_metric is not None + input_dp = _find_data_point(token_metric, {GenAIAttributes.GEN_AI_TOKEN_TYPE: "input"}) + assert input_dp is not None and input_dp.sum == 10 + output_dp = _find_data_point(token_metric, {GenAIAttributes.GEN_AI_TOKEN_TYPE: "output"}) + assert output_dp is not None and output_dp.sum == 5 + + duration_metric = _find_metric(metrics_data, Meters.LLM_OPERATION_DURATION) + assert duration_metric is not None + assert duration_metric.data.data_points[0].sum > 0 + + def test_records_duration_even_without_usage(self, reader, tracer_provider, meter_provider): + span = get_tracer("test", tracer_provider=tracer_provider).start_span("chat llama-3.3-70b-versatile") + duration_histogram = meter_provider.get_meter("test").create_histogram(name=Meters.LLM_OPERATION_DURATION) + + stream = _FakeStream([_chunk(content="hello", finish_reason="stop")]) + + processor = _create_stream_processor( + stream, + span, + None, + token_histogram=None, + duration_histogram=duration_histogram, + start_time=0, + llm_model="llama-3.3-70b-versatile", + ) + for _ in processor: + pass + + metrics_data = reader.get_metrics_data() + duration_metric = _find_metric(metrics_data, Meters.LLM_OPERATION_DURATION) + assert duration_metric is not None + assert duration_metric.data.data_points[0].sum > 0 + + token_metric = _find_metric(metrics_data, Meters.LLM_TOKEN_USAGE) + assert token_metric is None + + def test_metrics_disabled_skips_recording(self, reader, tracer_provider, meter_provider): + """Histograms are None when metrics are disabled; recording must be skipped.""" + span = get_tracer("test", tracer_provider=tracer_provider).start_span("chat llama-3.3-70b-versatile") + stream = _FakeStream([_chunk(content="hello", finish_reason="stop", usage=_usage())]) + + processor = _create_stream_processor( + stream, + span, + None, + token_histogram=None, + duration_histogram=None, + start_time=0, + llm_model="llama-3.3-70b-versatile", + ) + for _ in processor: + pass + + metrics_data = reader.get_metrics_data() + assert _find_metric(metrics_data, Meters.LLM_TOKEN_USAGE) is None + assert _find_metric(metrics_data, Meters.LLM_OPERATION_DURATION) is None + + +# --------------------------------------------------------------------------- +# _create_async_stream_processor +# --------------------------------------------------------------------------- + + +class TestAsyncStreamProcessorMetrics: + @pytest.mark.asyncio + async def test_async_records_token_and_duration_metrics(self, reader, tracer_provider, meter_provider): + span = get_tracer("test", tracer_provider=tracer_provider).start_span("chat llama-3.3-70b-versatile") + token_histogram = meter_provider.get_meter("test").create_histogram(name=Meters.LLM_TOKEN_USAGE) + duration_histogram = meter_provider.get_meter("test").create_histogram(name=Meters.LLM_OPERATION_DURATION) + + usage = _usage(prompt=7, completion=3) + stream = _FakeAsyncStream( + [ + _chunk(content="hello"), + _chunk(content=" world", finish_reason="stop", usage=usage), + ] + ) + + processor = _create_async_stream_processor( + stream, + span, + None, + token_histogram, + duration_histogram, + start_time=0, + llm_model="llama-3.3-70b-versatile", + ) + async for _ in processor: + pass + + metrics_data = reader.get_metrics_data() + + token_metric = _find_metric(metrics_data, Meters.LLM_TOKEN_USAGE) + assert token_metric is not None + input_dp = _find_data_point(token_metric, {GenAIAttributes.GEN_AI_TOKEN_TYPE: "input"}) + assert input_dp is not None and input_dp.sum == 7 + output_dp = _find_data_point(token_metric, {GenAIAttributes.GEN_AI_TOKEN_TYPE: "output"}) + assert output_dp is not None and output_dp.sum == 3 + + duration_metric = _find_metric(metrics_data, Meters.LLM_OPERATION_DURATION) + assert duration_metric is not None + assert duration_metric.data.data_points[0].sum > 0 From 2f712bdeba019e5b511ff3aef6c361e0a76178d3 Mon Sep 17 00:00:00 2001 From: chengwudi1 Date: Mon, 31 Aug 2026 09:51:59 +0800 Subject: [PATCH 2/4] fix(groq): record duration metric when streaming iteration fails Addresses reviewer feedback: a stream that yields chunks and then raises now emits an operation-duration data point with error attributes in both the sync and async stream processors, matching the error-path behavior in _wrap/_awrap. --- .../instrumentation/groq/__init__.py | 10 ++ .../metrics/test_groq_streaming_metrics.py | 97 +++++++++++++++++++ 2 files changed, 107 insertions(+) diff --git a/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/__init__.py b/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/__init__.py index 666e551d84..d8a8761fda 100644 --- a/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/__init__.py +++ b/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/__init__.py @@ -256,6 +256,11 @@ def _create_stream_processor( usage = chunk_usage yield chunk except Exception as e: + if duration_histogram and start_time is not None: + duration_histogram.record( + time.time() - start_time, + attributes=error_metrics_attributes(e), + ) span.set_attribute(ERROR_TYPE, e.__class__.__name__) span.record_exception(e) span.set_status(Status(StatusCode.ERROR, str(e))) @@ -299,6 +304,11 @@ async def _create_async_stream_processor( usage = chunk_usage yield chunk except Exception as e: + if duration_histogram and start_time is not None: + duration_histogram.record( + time.time() - start_time, + attributes=error_metrics_attributes(e), + ) span.set_attribute(ERROR_TYPE, e.__class__.__name__) span.record_exception(e) span.set_status(Status(StatusCode.ERROR, str(e))) diff --git a/packages/opentelemetry-instrumentation-groq/tests/metrics/test_groq_streaming_metrics.py b/packages/opentelemetry-instrumentation-groq/tests/metrics/test_groq_streaming_metrics.py index 1d0354e15b..403b31b8b6 100644 --- a/packages/opentelemetry-instrumentation-groq/tests/metrics/test_groq_streaming_metrics.py +++ b/packages/opentelemetry-instrumentation-groq/tests/metrics/test_groq_streaming_metrics.py @@ -56,6 +56,14 @@ async def __aiter__(self): yield chunk +class _FailingAsyncStream: + """Async stream that yields one chunk then raises.""" + + async def __aiter__(self): + yield _chunk(content="hello") + raise RuntimeError("async stream exploded") + + def _find_metric(metrics_data, name): """Return the first metric whose instrument name matches, or None.""" if metrics_data is None: @@ -225,3 +233,92 @@ async def test_async_records_token_and_duration_metrics(self, reader, tracer_pro duration_metric = _find_metric(metrics_data, Meters.LLM_OPERATION_DURATION) assert duration_metric is not None assert duration_metric.data.data_points[0].sum > 0 + + +# --------------------------------------------------------------------------- +# _create_stream_processor failure paths (review: record duration on error) +# --------------------------------------------------------------------------- + + +class TestStreamProcessorErrorMetrics: + def test_sync_records_duration_metric_on_stream_failure(self, reader, tracer_provider, meter_provider): + """A stream that raises mid-iteration must still emit an operation-duration + metric (with error attributes) before re-raising.""" + span = get_tracer("test", tracer_provider=tracer_provider).start_span("chat llama-3.3-70b-versatile") + duration_histogram = meter_provider.get_meter("test").create_histogram(name=Meters.LLM_OPERATION_DURATION) + + def _failing(): + yield _chunk(content="hello") + raise RuntimeError("stream exploded") + + processor = _create_stream_processor( + _FakeStream(_failing()), + span, + None, + token_histogram=None, + duration_histogram=duration_histogram, + start_time=0, + llm_model="llama-3.3-70b-versatile", + ) + with pytest.raises(RuntimeError): + for _ in processor: + pass + + metrics_data = reader.get_metrics_data() + duration_metric = _find_metric(metrics_data, Meters.LLM_OPERATION_DURATION) + assert duration_metric is not None + assert duration_metric.data.data_points[0].sum > 0 + dp = duration_metric.data.data_points[0] + assert dict(dp.attributes)["error.type"] == "RuntimeError" + + def test_sync_skips_duration_metric_when_disabled_on_failure(self, reader, tracer_provider, meter_provider): + """With histograms disabled (None), a failing stream must not crash and + must still propagate the exception.""" + span = get_tracer("test", tracer_provider=tracer_provider).start_span("chat llama-3.3-70b-versatile") + + def _failing(): + yield _chunk(content="hello") + raise ValueError("boom") + + processor = _create_stream_processor( + _FakeStream(_failing()), + span, + None, + token_histogram=None, + duration_histogram=None, + start_time=0, + llm_model="llama-3.3-70b-versatile", + ) + with pytest.raises(ValueError): + for _ in processor: + pass + + metrics_data = reader.get_metrics_data() + assert _find_metric(metrics_data, Meters.LLM_OPERATION_DURATION) is None + + @pytest.mark.asyncio + async def test_async_records_duration_metric_on_stream_failure(self, reader, tracer_provider, meter_provider): + """The async stream processor must emit the duration metric with error + attributes when iteration fails.""" + span = get_tracer("test", tracer_provider=tracer_provider).start_span("chat llama-3.3-70b-versatile") + duration_histogram = meter_provider.get_meter("test").create_histogram(name=Meters.LLM_OPERATION_DURATION) + + processor = _create_async_stream_processor( + _FailingAsyncStream(), + span, + None, + token_histogram=None, + duration_histogram=duration_histogram, + start_time=0, + llm_model="llama-3.3-70b-versatile", + ) + with pytest.raises(RuntimeError): + async for _ in processor: + pass + + metrics_data = reader.get_metrics_data() + duration_metric = _find_metric(metrics_data, Meters.LLM_OPERATION_DURATION) + assert duration_metric is not None + assert duration_metric.data.data_points[0].sum > 0 + dp = duration_metric.data.data_points[0] + assert dict(dp.attributes)["error.type"] == "RuntimeError" From 91be7b30f1f97f3137753c2e70f73e538099288a Mon Sep 17 00:00:00 2001 From: chengwudi1 Date: Mon, 28 Sep 2026 17:42:22 +0800 Subject: [PATCH 3/4] fix(groq): align streaming metric attributes with the non-streaming path Addresses reviewer feedback on #4439 (4 inline comments + 2 requested tests). Attribute parity: streaming token points only carried provider + response model + token type, while the non-streaming path in span_utils also records gen_ai.operation.name and gen_ai.request.model. semconv marks operation.name required on this metric, and any dashboard grouping or filtering by operation or request model silently left out streaming calls. The two missing keys are now added to the token points; duration points keep the attribute set they already shared with the non-streaming path. Response model: gen_ai.response.model was populated from the request kwargs. The processors now keep the last model reported on the chunks and fall back to the requested model when the chunks carry none, matching how the non-streaming path reads response.model. The duplicated error-duration block (four copies: both stream processors and _wrap/_awrap) moves into _record_error_duration, and _record_streaming_metrics is guarded with @dont_throw since it runs in the generator else block, where a raise would reach the caller for loop after they had consumed the whole stream. Dropping the now-dead end_time assignments surfaces ruff F841. Tests: test_chat_streaming_legacy and test_chat_legacy now go through instrument_legacy and assert on the reader metrics, so a regression in the _wrap histogram wiring (the actual #4419 bug) fails the test instead of being masked by processors the test builds itself. Both assert the same expected attribute-key set, which pins streaming/non-streaming parity. Unit tests cover chunk.model precedence, the request-model fallback, and the exception guard. --- .../instrumentation/groq/__init__.py | 86 +++++++++------ .../instrumentation/groq/utils.py | 10 +- .../metrics/test_groq_streaming_metrics.py | 102 +++++++++++++++++- .../tests/traces/test_chat_tracing.py | 56 +++++++++- 4 files changed, 214 insertions(+), 40 deletions(-) diff --git a/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/__init__.py b/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/__init__.py index d8a8761fda..84906f4e07 100644 --- a/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/__init__.py +++ b/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/__init__.py @@ -23,6 +23,7 @@ set_streaming_response_attributes, ) from opentelemetry.instrumentation.groq.utils import ( + dont_throw, error_metrics_attributes, shared_metrics_attributes, should_emit_events, @@ -191,18 +192,42 @@ def _handle_streaming_response( set_streaming_response_attributes(span, accumulated_content, finish_reason, tool_calls=tool_calls) +def _record_error_duration( + duration_histogram: Optional[Histogram], + start_time: Optional[float], + exception: Exception, +) -> None: + """Record how long a failed call took on the duration histogram. + + Shared by the stream processors and ``_wrap``/``_awrap`` so the four copies + of this block can't drift apart. + """ + if not duration_histogram or start_time is None: + return + duration_histogram.record( + time.time() - start_time, + attributes=error_metrics_attributes(exception), + ) + + +@dont_throw def _record_streaming_metrics( usage, token_histogram, duration_histogram, start_time, llm_model, + response_model=None, ) -> None: """Record token usage and duration metrics for a consumed streaming response. Called once the stream is fully drained; the final chunk carries usage data. + + Runs in the generator's ``else:`` block, so an exception here would surface + in the caller's ``for`` loop *after* they consumed the whole stream - hence + the guard, like every other telemetry helper in this package. """ - metric_attributes = streaming_metrics_attributes(llm_model) + metric_attributes = streaming_metrics_attributes(response_model or llm_model) if duration_histogram and start_time is not None: duration_histogram.record( @@ -211,11 +236,18 @@ def _record_streaming_metrics( ) if usage and token_histogram: + # Token points carry the same attribute set as the non-streaming path so + # dashboards grouping by operation/request model don't drop streaming calls. + token_attributes = { + **metric_attributes, + GenAIAttributes.GEN_AI_OPERATION_NAME: GenAIAttributes.GenAiOperationNameValues.CHAT.value, + GenAIAttributes.GEN_AI_REQUEST_MODEL: llm_model, + } if usage.prompt_tokens is not None: token_histogram.record( usage.prompt_tokens, attributes={ - **metric_attributes, + **token_attributes, GenAIAttributes.GEN_AI_TOKEN_TYPE: "input", }, ) @@ -223,7 +255,7 @@ def _record_streaming_metrics( token_histogram.record( usage.completion_tokens, attributes={ - **metric_attributes, + **token_attributes, GenAIAttributes.GEN_AI_TOKEN_TYPE: "output", }, ) @@ -243,6 +275,8 @@ def _create_stream_processor( accumulated_tool_calls: dict = {} accumulated_finish_reasons: list = [] usage = None + # Last model reported by the server; falls back to the requested model. + response_model = llm_model try: for chunk in response: @@ -254,13 +288,11 @@ def _create_stream_processor( accumulated_finish_reasons.extend(chunk_finish_reasons) if chunk_usage: usage = chunk_usage + if getattr(chunk, "model", None): + response_model = chunk.model yield chunk except Exception as e: - if duration_histogram and start_time is not None: - duration_histogram.record( - time.time() - start_time, - attributes=error_metrics_attributes(e), - ) + _record_error_duration(duration_histogram, start_time, e) span.set_attribute(ERROR_TYPE, e.__class__.__name__) span.record_exception(e) span.set_status(Status(StatusCode.ERROR, str(e))) @@ -270,7 +302,9 @@ def _create_stream_processor( _handle_streaming_response( span, accumulated_content, tool_calls, accumulated_finish_reasons, usage, event_logger ) - _record_streaming_metrics(usage, token_histogram, duration_histogram, start_time, llm_model) + _record_streaming_metrics( + usage, token_histogram, duration_histogram, start_time, llm_model, response_model + ) if span.is_recording(): span.set_status(Status(StatusCode.OK)) finally: @@ -291,6 +325,8 @@ async def _create_async_stream_processor( accumulated_tool_calls: dict = {} accumulated_finish_reasons: list = [] usage = None + # Last model reported by the server; falls back to the requested model. + response_model = llm_model try: async for chunk in response: @@ -302,13 +338,11 @@ async def _create_async_stream_processor( accumulated_finish_reasons.extend(chunk_finish_reasons) if chunk_usage: usage = chunk_usage + if getattr(chunk, "model", None): + response_model = chunk.model yield chunk except Exception as e: - if duration_histogram and start_time is not None: - duration_histogram.record( - time.time() - start_time, - attributes=error_metrics_attributes(e), - ) + _record_error_duration(duration_histogram, start_time, e) span.set_attribute(ERROR_TYPE, e.__class__.__name__) span.record_exception(e) span.set_status(Status(StatusCode.ERROR, str(e))) @@ -318,7 +352,9 @@ async def _create_async_stream_processor( _handle_streaming_response( span, accumulated_content, tool_calls, accumulated_finish_reasons, usage, event_logger ) - _record_streaming_metrics(usage, token_histogram, duration_histogram, start_time, llm_model) + _record_streaming_metrics( + usage, token_histogram, duration_histogram, start_time, llm_model, response_model + ) if span.is_recording(): span.set_status(Status(StatusCode.OK)) finally: @@ -377,21 +413,13 @@ def _wrap( try: response = wrapped(*args, **kwargs) except Exception as e: - end_time = time.time() - attributes = error_metrics_attributes(e) - - if duration_histogram: - duration = end_time - start_time - duration_histogram.record(duration, attributes=attributes) - + _record_error_duration(duration_histogram, start_time, e) span.set_attribute(ERROR_TYPE, e.__class__.__name__) span.record_exception(e) span.set_status(Status(StatusCode.ERROR, str(e))) span.end() raise - end_time = time.time() - if is_streaming_response(response): try: return _create_stream_processor( @@ -473,21 +501,13 @@ async def _awrap( try: response = await wrapped(*args, **kwargs) except Exception as e: - end_time = time.time() - attributes = error_metrics_attributes(e) - - if duration_histogram: - duration = end_time - start_time - duration_histogram.record(duration, attributes=attributes) - + _record_error_duration(duration_histogram, start_time, e) span.set_attribute(ERROR_TYPE, e.__class__.__name__) span.record_exception(e) span.set_status(Status(StatusCode.ERROR, str(e))) span.end() raise - end_time = time.time() - if is_streaming_response(response): try: return _create_async_stream_processor( diff --git a/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/utils.py b/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/utils.py index 07aab456a1..e1f01b60ec 100644 --- a/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/utils.py +++ b/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/utils.py @@ -64,12 +64,16 @@ def shared_metrics_attributes(response): @dont_throw -def streaming_metrics_attributes(model): - """Metrics attributes for streaming responses, where no full response object exists yet.""" +def streaming_metrics_attributes(response_model): + """Metrics attributes for streaming responses, where no full response object exists yet. + + response_model is the model the server reported on the chunks, falling + back to the requested model when the chunks carry none. + """ return { **Config.get_common_metrics_attributes(), GenAIAttributes.GEN_AI_PROVIDER_NAME: GenAIAttributes.GenAiProviderNameValues.GROQ.value, - GenAIAttributes.GEN_AI_RESPONSE_MODEL: model, + GenAIAttributes.GEN_AI_RESPONSE_MODEL: response_model, } diff --git a/packages/opentelemetry-instrumentation-groq/tests/metrics/test_groq_streaming_metrics.py b/packages/opentelemetry-instrumentation-groq/tests/metrics/test_groq_streaming_metrics.py index 403b31b8b6..b7dcdde2fc 100644 --- a/packages/opentelemetry-instrumentation-groq/tests/metrics/test_groq_streaming_metrics.py +++ b/packages/opentelemetry-instrumentation-groq/tests/metrics/test_groq_streaming_metrics.py @@ -12,6 +12,7 @@ from opentelemetry.instrumentation.groq import ( _create_async_stream_processor, _create_stream_processor, + _record_streaming_metrics, ) from opentelemetry.instrumentation.groq.utils import streaming_metrics_attributes from opentelemetry.semconv._incubating.attributes import gen_ai_attributes as GenAIAttributes @@ -19,8 +20,9 @@ from opentelemetry.trace import get_tracer -def _chunk(content="", finish_reason=None, usage=None): +def _chunk(content="", finish_reason=None, usage=None, model=None): return SimpleNamespace( + model=model, choices=[ SimpleNamespace( delta=SimpleNamespace(content=content, tool_calls=None), @@ -141,6 +143,104 @@ def test_records_token_and_duration_metrics(self, reader, tracer_provider, meter assert duration_metric is not None assert duration_metric.data.data_points[0].sum > 0 + def test_token_points_carry_non_streaming_attribute_set(self, reader, tracer_provider, meter_provider): + """Token points must match the non-streaming attribute keys (semconv requires operation.name).""" + span = get_tracer("test", tracer_provider=tracer_provider).start_span("chat llama-3.3-70b-versatile") + token_histogram = meter_provider.get_meter("test").create_histogram(name=Meters.LLM_TOKEN_USAGE) + + stream = _FakeStream([_chunk(content="hello", finish_reason="stop", usage=_usage())]) + for _ in _create_stream_processor( + stream, + span, + None, + token_histogram, + None, + start_time=0, + llm_model="llama-3.3-70b-versatile", + ): + pass + + metrics_data = reader.get_metrics_data() + token_metric = _find_metric(metrics_data, Meters.LLM_TOKEN_USAGE) + assert token_metric is not None + assert {frozenset(dict(dp.attributes)) for dp in token_metric.data.data_points} == { + frozenset( + { + GenAIAttributes.GEN_AI_PROVIDER_NAME, + GenAIAttributes.GEN_AI_OPERATION_NAME, + GenAIAttributes.GEN_AI_REQUEST_MODEL, + GenAIAttributes.GEN_AI_RESPONSE_MODEL, + GenAIAttributes.GEN_AI_TOKEN_TYPE, + } + ) + } + + def test_uses_last_chunk_model_as_response_model(self, reader, tracer_provider, meter_provider): + """gen_ai.response.model comes from the server's chunks, not the request kwargs.""" + span = get_tracer("test", tracer_provider=tracer_provider).start_span("chat llama-3.3-70b-versatile") + token_histogram = meter_provider.get_meter("test").create_histogram(name=Meters.LLM_TOKEN_USAGE) + duration_histogram = meter_provider.get_meter("test").create_histogram(name=Meters.LLM_OPERATION_DURATION) + + stream = _FakeStream( + [ + _chunk(content="hello", model="server-reported-model"), + _chunk(content=" world", finish_reason="stop", usage=_usage(), model="server-reported-model"), + ] + ) + for _ in _create_stream_processor( + stream, + span, + None, + token_histogram, + duration_histogram, + start_time=0, + llm_model="llama-3.3-70b-versatile", + ): + pass + + metrics_data = reader.get_metrics_data() + token_dp = _find_metric(metrics_data, Meters.LLM_TOKEN_USAGE).data.data_points[0] + duration_dp = _find_metric(metrics_data, Meters.LLM_OPERATION_DURATION).data.data_points[0] + # Duration points keep the non-streaming attribute set (provider + response + # model only); the request model is added to token points alone. + assert dict(duration_dp.attributes)[GenAIAttributes.GEN_AI_RESPONSE_MODEL] == "server-reported-model" + assert GenAIAttributes.GEN_AI_REQUEST_MODEL not in dict(duration_dp.attributes) + + token_attrs = dict(token_dp.attributes) + assert token_attrs[GenAIAttributes.GEN_AI_RESPONSE_MODEL] == "server-reported-model" + assert token_attrs[GenAIAttributes.GEN_AI_REQUEST_MODEL] == "llama-3.3-70b-versatile" + + def test_falls_back_to_request_model_when_chunks_carry_none( + self, reader, tracer_provider, meter_provider + ): + span = get_tracer("test", tracer_provider=tracer_provider).start_span("chat llama-3.3-70b-versatile") + token_histogram = meter_provider.get_meter("test").create_histogram(name=Meters.LLM_TOKEN_USAGE) + + stream = _FakeStream([_chunk(content="hello", finish_reason="stop", usage=_usage())]) + for _ in _create_stream_processor( + stream, + span, + None, + token_histogram, + None, + start_time=0, + llm_model="llama-3.3-70b-versatile", + ): + pass + + metrics_data = reader.get_metrics_data() + token_dp = _find_metric(metrics_data, Meters.LLM_TOKEN_USAGE).data.data_points[0] + assert dict(token_dp.attributes)[GenAIAttributes.GEN_AI_RESPONSE_MODEL] == "llama-3.3-70b-versatile" + + def test_recording_failure_never_reaches_the_caller(self): + """_record_streaming_metrics runs in the generator's else block — it must not raise.""" + + class _Boom: + def record(self, *args, **kwargs): + raise RuntimeError("histogram exploded") + + _record_streaming_metrics(_usage(), _Boom(), _Boom(), 0, "llama-3.3-70b-versatile") + def test_records_duration_even_without_usage(self, reader, tracer_provider, meter_provider): span = get_tracer("test", tracer_provider=tracer_provider).start_span("chat llama-3.3-70b-versatile") duration_histogram = meter_provider.get_meter("test").create_histogram(name=Meters.LLM_OPERATION_DURATION) diff --git a/packages/opentelemetry-instrumentation-groq/tests/traces/test_chat_tracing.py b/packages/opentelemetry-instrumentation-groq/tests/traces/test_chat_tracing.py index fdf7daacf5..f7993ca963 100644 --- a/packages/opentelemetry-instrumentation-groq/tests/traces/test_chat_tracing.py +++ b/packages/opentelemetry-instrumentation-groq/tests/traces/test_chat_tracing.py @@ -4,7 +4,7 @@ from opentelemetry.semconv._incubating.attributes import ( gen_ai_attributes as GenAIAttributes, ) -from opentelemetry.semconv_ai import SpanAttributes +from opentelemetry.semconv_ai import Meters, SpanAttributes GEN_AI_IS_STREAMING = SpanAttributes.GEN_AI_IS_STREAMING GEN_AI_USAGE_TOTAL_TOKENS = SpanAttributes.GEN_AI_USAGE_TOTAL_TOKENS @@ -31,6 +31,39 @@ def _assert_otel_v2_span_attributes(span): assert span.attributes[GenAIAttributes.GEN_AI_REQUEST_MODEL] == MODEL +# Attribute keys every gen_ai.client.token.usage data point must carry, whether +# the call was streamed or not. semconv marks gen_ai.operation.name required, and +# dashboards grouping by operation/request model silently drop streaming calls +# when those two keys are missing (the bug behind #4419's follow-up review). +EXPECTED_TOKEN_ATTRIBUTE_KEYS = frozenset( + { + GenAIAttributes.GEN_AI_PROVIDER_NAME, + GenAIAttributes.GEN_AI_OPERATION_NAME, + GenAIAttributes.GEN_AI_REQUEST_MODEL, + GenAIAttributes.GEN_AI_RESPONSE_MODEL, + GenAIAttributes.GEN_AI_TOKEN_TYPE, + } +) + + +def _data_points(metrics_data, name): + """All data points recorded so far for the named metric.""" + if metrics_data is None: + return [] + for resource_metric in metrics_data.resource_metrics: + for scope_metric in resource_metric.scope_metrics: + for metric in scope_metric.metrics: + if metric.name == name: + return list(metric.data.data_points) + return [] + + +def _assert_token_usage_attribute_parity(token_points): + """Streaming and non-streaming token points must expose the same attribute keys.""" + assert token_points, "expected token usage data points" + assert {frozenset(dict(p.attributes)) for p in token_points} == {EXPECTED_TOKEN_ATTRIBUTE_KEYS} + + def assert_message_in_logs(log: ReadableLogRecord, event_name: str, expected_content: dict): assert log.log_record.event_name == event_name assert ( @@ -51,7 +84,7 @@ def assert_message_in_logs(log: ReadableLogRecord, event_name: str, expected_con @pytest.mark.vcr -def test_chat_legacy(instrument_legacy, groq_client, span_exporter, log_exporter): +def test_chat_legacy(instrument_legacy, groq_client, span_exporter, log_exporter, reader): groq_client.chat.completions.create( model=MODEL, messages=[{"role": "user", "content": "Tell me a joke about opentelemetry"}], @@ -81,6 +114,11 @@ def test_chat_legacy(instrument_legacy, groq_client, span_exporter, log_exporter assert groq_span.attributes.get(GEN_AI_USAGE_TOTAL_TOKENS) > 0 assert groq_span.attributes.get("gen_ai.response.id") == "chatcmpl-645691ff-34af-4d0f-a1c1-fe888f8685cc" + metrics_data = reader.get_metrics_data() + token_points = _data_points(metrics_data, Meters.LLM_TOKEN_USAGE) + _assert_token_usage_attribute_parity(token_points) + assert all(p.sum > 0 for p in token_points) + logs = log_exporter.get_finished_logs() assert len(logs) == 0, "Assert that it doesn't emit logs when use_legacy_attributes is True" @@ -305,7 +343,7 @@ async def test_async_chat_with_events_with_no_content( @pytest.mark.vcr -def test_chat_streaming_legacy(instrument_legacy, groq_client, span_exporter, log_exporter): +def test_chat_streaming_legacy(instrument_legacy, groq_client, span_exporter, log_exporter, reader): response = groq_client.chat.completions.create( model=MODEL, messages=[{"role": "user", "content": "Tell me a joke about opentelemetry"}], @@ -337,6 +375,18 @@ def test_chat_streaming_legacy(instrument_legacy, groq_client, span_exporter, lo assert groq_span.attributes.get(GenAIAttributes.GEN_AI_USAGE_OUTPUT_TOKENS) == 73 assert groq_span.attributes.get(GEN_AI_USAGE_TOTAL_TOKENS) == 91 + # End-to-end regression for #4419: the instrumentor must actually hand the + # histograms to the stream processors, so metrics come out of a real + # instrumented call rather than out of processors built by the test itself. + metrics_data = reader.get_metrics_data() + token_points = _data_points(metrics_data, Meters.LLM_TOKEN_USAGE) + _assert_token_usage_attribute_parity(token_points) + assert {p.attributes[GenAIAttributes.GEN_AI_TOKEN_TYPE]: p.sum for p in token_points} == { + "input": 18, + "output": 73, + } + assert _data_points(metrics_data, Meters.LLM_OPERATION_DURATION), "expected a duration data point" + logs = log_exporter.get_finished_logs() assert len(logs) == 0, "Assert that it doesn't emit logs when use_legacy_attributes is True" From b20e3cd1d71e3dce68294f3dba90e4570115d72c Mon Sep 17 00:00:00 2001 From: chengwudi1 Date: Mon, 28 Sep 2026 17:56:49 +0800 Subject: [PATCH 4/4] fix(groq): keep the caller error when recording the failure duration _record_error_duration runs inside the except block of both stream processors, so if duration_histogram.record() raised, that new exception would replace the one the caller actually hit - they would see a telemetry failure instead of their own stream error. Guard it with @dont_throw, the same way _record_streaming_metrics and error_metrics_attributes are guarded. Adds a regression test for each processor: a histogram whose record() raises must not change which exception escapes, asserted with pytest.raises on the original message. --- .../instrumentation/groq/__init__.py | 5 ++ .../metrics/test_groq_streaming_metrics.py | 54 +++++++++++++++++++ 2 files changed, 59 insertions(+) diff --git a/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/__init__.py b/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/__init__.py index 84906f4e07..2924b5de56 100644 --- a/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/__init__.py +++ b/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/__init__.py @@ -192,6 +192,7 @@ def _handle_streaming_response( set_streaming_response_attributes(span, accumulated_content, finish_reason, tool_calls=tool_calls) +@dont_throw def _record_error_duration( duration_histogram: Optional[Histogram], start_time: Optional[float], @@ -201,6 +202,10 @@ def _record_error_duration( Shared by the stream processors and ``_wrap``/``_awrap`` so the four copies of this block can't drift apart. + + Guarded because it runs inside an ``except`` block: if ``record()`` raised, + that new exception would replace the one the caller actually hit and they + would see a telemetry failure instead of their own error. """ if not duration_histogram or start_time is None: return diff --git a/packages/opentelemetry-instrumentation-groq/tests/metrics/test_groq_streaming_metrics.py b/packages/opentelemetry-instrumentation-groq/tests/metrics/test_groq_streaming_metrics.py index b7dcdde2fc..dceea85d93 100644 --- a/packages/opentelemetry-instrumentation-groq/tests/metrics/test_groq_streaming_metrics.py +++ b/packages/opentelemetry-instrumentation-groq/tests/metrics/test_groq_streaming_metrics.py @@ -422,3 +422,57 @@ async def test_async_records_duration_metric_on_stream_failure(self, reader, tra assert duration_metric.data.data_points[0].sum > 0 dp = duration_metric.data.data_points[0] assert dict(dp.attributes)["error.type"] == "RuntimeError" + + +class _RaisingHistogram: + """A histogram whose record() blows up, as the SDK's can under a broken exporter.""" + + def record(self, *args, **kwargs): + raise RuntimeError("histogram exploded") + + +class TestTelemetryFailureDoesNotMaskCallerError: + """Recording a metric must never change which exception the caller sees.""" + + def test_sync_stream_error_survives_a_broken_histogram(self, tracer_provider): + span = get_tracer("test", tracer_provider=tracer_provider).start_span("chat llama-3.3-70b-versatile") + + def _failing(): + yield _chunk(content="hello") + raise RuntimeError("stream exploded") + + processor = _create_stream_processor( + _FakeStream(_failing()), + span, + None, + token_histogram=None, + duration_histogram=_RaisingHistogram(), + start_time=0, + llm_model="llama-3.3-70b-versatile", + ) + + with pytest.raises(RuntimeError, match="stream exploded"): + for _ in processor: + pass + + @pytest.mark.asyncio + async def test_async_stream_error_survives_a_broken_histogram(self, tracer_provider): + span = get_tracer("test", tracer_provider=tracer_provider).start_span("chat llama-3.3-70b-versatile") + + async def _failing(): + yield _chunk(content="hello") + raise RuntimeError("async stream exploded") + + processor = _create_async_stream_processor( + _failing(), + span, + None, + token_histogram=None, + duration_histogram=_RaisingHistogram(), + start_time=0, + llm_model="llama-3.3-70b-versatile", + ) + + with pytest.raises(RuntimeError, match="async stream exploded"): + async for _ in processor: + pass