diff --git a/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/__init__.py b/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/__init__.py index 4a275051f2..2924b5de56 100644 --- a/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/__init__.py +++ b/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/__init__.py @@ -23,9 +23,11 @@ set_streaming_response_attributes, ) from opentelemetry.instrumentation.groq.utils import ( + dont_throw, 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,12 +192,96 @@ 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): +@dont_throw +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. + + 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 + 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(response_model or 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: + # 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={ + **token_attributes, + GenAIAttributes.GEN_AI_TOKEN_TYPE: "input", + }, + ) + if usage.completion_tokens is not None: + token_histogram.record( + usage.completion_tokens, + attributes={ + **token_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 = {} 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: @@ -207,8 +293,11 @@ def _create_stream_processor(response, span, event_logger): 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: + _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))) @@ -218,18 +307,31 @@ 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, response_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 = {} 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: @@ -241,8 +343,11 @@ async def _create_async_stream_processor(response, span, event_logger): 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: + _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))) @@ -252,6 +357,9 @@ 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, response_model + ) if span.is_recording(): span.set_status(Status(StatusCode.OK)) finally: @@ -310,24 +418,24 @@ 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(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", @@ -398,24 +506,24 @@ 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(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..e1f01b60ec 100644 --- a/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/utils.py +++ b/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/utils.py @@ -63,6 +63,20 @@ def shared_metrics_attributes(response): } +@dont_throw +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: response_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..dceea85d93 --- /dev/null +++ b/packages/opentelemetry-instrumentation-groq/tests/metrics/test_groq_streaming_metrics.py @@ -0,0 +1,478 @@ +"""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, + _record_streaming_metrics, +) +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, model=None): + return SimpleNamespace( + model=model, + 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 + + +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: + 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_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) + + 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 + + +# --------------------------------------------------------------------------- +# _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" + + +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 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"