From 28a93e372184ebac071fd18cc41a32410d0a6f4e Mon Sep 17 00:00:00 2001 From: sayonfortify Date: Sat, 20 Jun 2026 15:24:17 +0530 Subject: [PATCH] fix(streaming): use monotonic clamped latency timings --- .../instrumentation/anthropic/__init__.py | 18 ++--- .../instrumentation/anthropic/streaming.py | 14 ++-- .../tests/test_safety_unit.py | 11 ++- .../instrumentation/bedrock/__init__.py | 4 +- .../bedrock/streaming_safety.py | 20 +++--- .../tests/test_safety_hooks.py | 9 ++- .../langchain/callback_handler.py | 10 +-- .../instrumentation/langchain/span_utils.py | 2 +- .../test_streaming_latency_span_attrs.py | 36 +++++++++- .../litellm/streaming_safety.py | 20 +++--- .../tests/test_safety_hooks.py | 11 ++- .../openai/shared/chat_wrappers.py | 67 +++++++++++-------- .../test_streaming_latency_span_attrs.py | 10 ++- 13 files changed, 150 insertions(+), 82 deletions(-) diff --git a/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/__init__.py b/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/__init__.py index 6269354785..aed9eac54a 100644 --- a/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/__init__.py +++ b/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/__init__.py @@ -556,7 +556,7 @@ def _wrap( kwargs = _apply_prompt_safety(span, kwargs, name) _handle_input(span, event_logger, kwargs) - start_time = time.time() + start_time = time.perf_counter() # ST-10.4 (review-driven 2026-05-17): make the anthropic.chat span # the AMBIENT OTel context for the duration of the SDK call so the # FortifyRoot retry handler (wrapping @@ -575,11 +575,11 @@ def _wrap( with trace.use_span(span, end_on_exit=False): response = wrapped(*args, **kwargs) except Exception as e: # pylint: disable=broad-except - end_time = time.time() + end_time = time.perf_counter() attributes = error_metrics_attributes(e) if duration_histogram: - duration = end_time - start_time + duration = max(0, end_time - start_time) duration_histogram.record(duration, attributes=attributes) if exception_counter: @@ -587,7 +587,7 @@ def _wrap( raise e - end_time = time.time() + end_time = time.perf_counter() if is_streaming_response(response): return AnthropicStream( @@ -636,7 +636,7 @@ def _wrap( ) if duration_histogram: - duration = time.time() - start_time + duration = max(0, time.perf_counter() - start_time) duration_histogram.record( duration, attributes=metric_attributes, @@ -697,18 +697,18 @@ async def _awrap( ) kwargs = await asyncio.to_thread(_apply_prompt_safety, span, kwargs, name) # FR: async safety await _ahandle_input(span, event_logger, kwargs) - start_time = time.time() + start_time = time.perf_counter() # ST-10.4 (review-driven 2026-05-17): see sync _wrap above for # rationale on use_span(end_on_exit=False) around the wrapped call. try: with trace.use_span(span, end_on_exit=False): response = await wrapped(*args, **kwargs) except Exception as e: # pylint: disable=broad-except - end_time = time.time() + end_time = time.perf_counter() attributes = error_metrics_attributes(e) if duration_histogram: - duration = end_time - start_time + duration = max(0, end_time - start_time) duration_histogram.record(duration, attributes=attributes) if exception_counter: @@ -766,7 +766,7 @@ async def _awrap( ) if duration_histogram: - duration = time.time() - start_time + duration = max(0, time.perf_counter() - start_time) duration_histogram.record( duration, attributes=metric_attributes, diff --git a/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/streaming.py b/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/streaming.py index 704a401113..672701148a 100644 --- a/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/streaming.py +++ b/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/streaming.py @@ -50,7 +50,7 @@ def _streaming_latency_ms(start_time, end_time): - return int(round((end_time - start_time) * 1000)) + return int(round(max(0, end_time - start_time) * 1000)) def _is_streaming_token_item(item): @@ -306,7 +306,7 @@ def __next__(self): # FR: rewritten for pending-item safety buffer item = self._streaming_safety.process_item(item) # FR: streaming safety processing if self._first_token_time is None and _is_streaming_token_item(item): - self._first_token_time = time.time() + self._first_token_time = time.perf_counter() _set_streaming_first_token_latency( self._span, self._start_time, @@ -335,14 +335,14 @@ def _handle_completion(self): GenAIAttributes.GEN_AI_RESPONSE_ID, self._complete_response.get("id"), ) - end_time = time.time() + end_time = time.perf_counter() _set_streaming_time_to_generate( self._span, self._first_token_time, end_time, ) if self._duration_histogram: - duration = end_time - self._start_time + duration = max(0, end_time - self._start_time) self._duration_histogram.record( duration, attributes=metric_attributes, @@ -510,7 +510,7 @@ async def __anext__(self): # FR: rewritten for pending-item safety buffer item = self._streaming_safety.process_item(item) # FR: streaming safety processing if self._first_token_time is None and _is_streaming_token_item(item): - self._first_token_time = time.time() + self._first_token_time = time.perf_counter() _set_streaming_first_token_latency( self._span, self._start_time, @@ -537,7 +537,7 @@ def _complete_instrumentation(self): request_model=self._kwargs.get("model"), ) set_span_attribute(self._span, GenAIAttributes.GEN_AI_RESPONSE_ID, self._complete_response.get("id")) - end_time = time.time() + end_time = time.perf_counter() _set_streaming_time_to_generate( self._span, self._first_token_time, @@ -545,7 +545,7 @@ def _complete_instrumentation(self): ) if self._duration_histogram: - duration = end_time - self._start_time + duration = max(0, end_time - self._start_time) self._duration_histogram.record( duration, attributes=metric_attributes, diff --git a/packages/opentelemetry-instrumentation-anthropic/tests/test_safety_unit.py b/packages/opentelemetry-instrumentation-anthropic/tests/test_safety_unit.py index cfa2125c0a..e07b5d67c7 100644 --- a/packages/opentelemetry-instrumentation-anthropic/tests/test_safety_unit.py +++ b/packages/opentelemetry-instrumentation-anthropic/tests/test_safety_unit.py @@ -12,6 +12,7 @@ AnthropicStream, FR_STREAMING_TIME_TO_FIRST_TOKEN_MS, FR_STREAMING_TIME_TO_GENERATE_MS, + _streaming_latency_ms, ) from opentelemetry.instrumentation.fortifyroot import SafetyDecision, SafetyResult from opentelemetry.instrumentation.fortifyroot import ( @@ -27,6 +28,10 @@ pytestmark = pytest.mark.fr +def test_streaming_latency_ms_clamps_negative_duration(): + assert _streaming_latency_ms(10.0, 9.0) == 0 + + @pytest.fixture(autouse=True) def _cleanup_safety_handlers(): """Ensure safety handlers are cleared after each test in this file. @@ -419,7 +424,7 @@ def test_anthropic_sync_stream_sets_streaming_latency_span_attrs(monkeypatch): timestamps = iter([101.25, 103.0]) monkeypatch.setattr( - "opentelemetry.instrumentation.anthropic.streaming.time.time", + "opentelemetry.instrumentation.anthropic.streaming.time.perf_counter", lambda: next(timestamps), ) @@ -464,7 +469,7 @@ def test_anthropic_sync_stream_without_token_omits_streaming_latency_attrs(monke timestamps = iter([301.0]) monkeypatch.setattr( - "opentelemetry.instrumentation.anthropic.streaming.time.time", + "opentelemetry.instrumentation.anthropic.streaming.time.perf_counter", lambda: next(timestamps), ) @@ -540,7 +545,7 @@ async def test_anthropic_async_stream_sets_streaming_latency_span_attrs(monkeypa timestamps = iter([201.1, 202.4]) monkeypatch.setattr( - "opentelemetry.instrumentation.anthropic.streaming.time.time", + "opentelemetry.instrumentation.anthropic.streaming.time.perf_counter", lambda: next(timestamps), ) diff --git a/packages/opentelemetry-instrumentation-bedrock/opentelemetry/instrumentation/bedrock/__init__.py b/packages/opentelemetry-instrumentation-bedrock/opentelemetry/instrumentation/bedrock/__init__.py index 1c109a4802..1bb4e8187d 100644 --- a/packages/opentelemetry-instrumentation-bedrock/opentelemetry/instrumentation/bedrock/__init__.py +++ b/packages/opentelemetry-instrumentation-bedrock/opentelemetry/instrumentation/bedrock/__init__.py @@ -263,7 +263,7 @@ def with_instrumentation(*args, **kwargs): # the response stream is fully consumed (which is why # ``start_as_current_span`` cannot be used directly). kwargs = _apply_invoke_prompt_safety(span, kwargs, _BEDROCK_INVOKE_SPAN_NAME) - stream_start_time = time.time() + stream_start_time = time.perf_counter() with trace.use_span(span, end_on_exit=False): response = fn(*args, **kwargs) _handle_stream_call( @@ -306,7 +306,7 @@ def with_instrumentation(*args, **kwargs): kwargs = _apply_converse_prompt_safety(span, kwargs, _BEDROCK_CONVERSE_SPAN_NAME) # ST-10.4: see _instrumented_model_invoke_with_response_stream # for the rationale on use_span(end_on_exit=False). - stream_start_time = time.time() + stream_start_time = time.perf_counter() with trace.use_span(span, end_on_exit=False): response = fn(*args, **kwargs) if span.is_recording(): diff --git a/packages/opentelemetry-instrumentation-bedrock/opentelemetry/instrumentation/bedrock/streaming_safety.py b/packages/opentelemetry-instrumentation-bedrock/opentelemetry/instrumentation/bedrock/streaming_safety.py index 3ab0d68c7c..7df4844a51 100644 --- a/packages/opentelemetry-instrumentation-bedrock/opentelemetry/instrumentation/bedrock/streaming_safety.py +++ b/packages/opentelemetry-instrumentation-bedrock/opentelemetry/instrumentation/bedrock/streaming_safety.py @@ -25,6 +25,10 @@ FR_STREAMING_TIME_TO_GENERATE_MS = "fortifyroot.llm.streaming.time_to_generate_ms" +def _streaming_latency_ms(start_time, end_time): + return round(max(0, end_time - start_time) * 1000) + + class _BedrockChunkStreamingSafety: def __init__(self, *, span, span_name: str, request_type: str): self._streams = CompletionTextStreamGroup( @@ -157,7 +161,7 @@ def __init__(self, response, *, span, stream_done_callback=None, stream_start_ti self._self_accumulating_body = {} self._self_pending_event = None self._self_stream_start_time = ( - time.time() if stream_start_time is None else stream_start_time + time.perf_counter() if stream_start_time is None else stream_start_time ) self._self_first_token_time = None self._self_streaming_safety = _BedrockChunkStreamingSafety( @@ -233,12 +237,12 @@ def _accumulate_events(self, payload): def _observe_streaming_text(self, text): if not isinstance(text, str) or not text or self._self_first_token_time is not None: return - now = time.time() + now = time.perf_counter() self._self_first_token_time = now if self._self_span.is_recording(): self._self_span.set_attribute( FR_STREAMING_TIME_TO_FIRST_TOKEN_MS, - round((now - self._self_stream_start_time) * 1000), + _streaming_latency_ms(self._self_stream_start_time, now), ) def _finish_streaming_latency(self): @@ -246,7 +250,7 @@ def _finish_streaming_latency(self): return self._self_span.set_attribute( FR_STREAMING_TIME_TO_GENERATE_MS, - round((time.time() - self._self_first_token_time) * 1000), + _streaming_latency_ms(self._self_first_token_time, time.perf_counter()), ) @@ -367,7 +371,7 @@ def __init__( self._self_response_msg = [] self._self_span_ended = False self._self_stream_start_time = ( - time.time() if stream_start_time is None else stream_start_time + time.perf_counter() if stream_start_time is None else stream_start_time ) self._self_first_token_time = None self._self_streaming_safety = _BedrockConverseStreamingSafety( @@ -449,12 +453,12 @@ def _observe_event(self, event): def _observe_streaming_text(self, text): if not isinstance(text, str) or not text or self._self_first_token_time is not None: return - now = time.time() + now = time.perf_counter() self._self_first_token_time = now if self._self_span.is_recording(): self._self_span.set_attribute( FR_STREAMING_TIME_TO_FIRST_TOKEN_MS, - round((now - self._self_stream_start_time) * 1000), + _streaming_latency_ms(self._self_stream_start_time, now), ) def _finish_streaming_latency(self): @@ -462,7 +466,7 @@ def _finish_streaming_latency(self): return self._self_span.set_attribute( FR_STREAMING_TIME_TO_GENERATE_MS, - round((time.time() - self._self_first_token_time) * 1000), + _streaming_latency_ms(self._self_first_token_time, time.perf_counter()), ) diff --git a/packages/opentelemetry-instrumentation-bedrock/tests/test_safety_hooks.py b/packages/opentelemetry-instrumentation-bedrock/tests/test_safety_hooks.py index 13510281f1..414d89fac1 100644 --- a/packages/opentelemetry-instrumentation-bedrock/tests/test_safety_hooks.py +++ b/packages/opentelemetry-instrumentation-bedrock/tests/test_safety_hooks.py @@ -20,6 +20,7 @@ _decode_chunk_event, _encode_chunk_event, _set_payload_text, + _streaming_latency_ms, create_converse_stream_wrapper, create_invoke_stream_wrapper, ) @@ -53,6 +54,10 @@ pytestmark = pytest.mark.fr +def test_streaming_latency_ms_clamps_negative_duration(): + assert _streaming_latency_ms(10.0, 9.0) == 0 + + def setup_function(): clear_safety_handlers() clear_completion_safety_stream_factory() @@ -411,7 +416,7 @@ def test_invoke_streaming_wrapper_sets_latency_span_attrs(monkeypatch): ) ticks = iter([100.125, 101.250]) monkeypatch.setattr( - "opentelemetry.instrumentation.bedrock.streaming_safety.time.time", + "opentelemetry.instrumentation.bedrock.streaming_safety.time.perf_counter", lambda: next(ticks), ) @@ -496,7 +501,7 @@ def test_converse_streaming_wrapper_sets_latency_span_attrs(monkeypatch): ) ticks = iter([200.050, 200.900]) monkeypatch.setattr( - "opentelemetry.instrumentation.bedrock.streaming_safety.time.time", + "opentelemetry.instrumentation.bedrock.streaming_safety.time.perf_counter", lambda: next(ticks), ) diff --git a/packages/opentelemetry-instrumentation-langchain/opentelemetry/instrumentation/langchain/callback_handler.py b/packages/opentelemetry-instrumentation-langchain/opentelemetry/instrumentation/langchain/callback_handler.py index 50f069a01c..7cb152d254 100644 --- a/packages/opentelemetry-instrumentation-langchain/opentelemetry/instrumentation/langchain/callback_handler.py +++ b/packages/opentelemetry-instrumentation-langchain/opentelemetry/instrumentation/langchain/callback_handler.py @@ -623,14 +623,14 @@ def on_llm_new_token( if span_holder.streaming_first_token_time is not None: return - first_token_time = time.time() + first_token_time = time.perf_counter() span_holder.streaming_first_token_time = first_token_time span = span_holder.span if span.is_recording(): _set_span_attribute( span, FR_STREAMING_TIME_TO_FIRST_TOKEN_MS, - round((first_token_time - span_holder.start_time) * 1000), + round(max(0, first_token_time - span_holder.start_time) * 1000), ) @dont_throw @@ -735,15 +735,15 @@ def on_llm_end( set_chat_response(span, response) # Record duration before ending span - end_time = time.time() + end_time = time.perf_counter() streaming_first_token_time = self.spans[run_id].streaming_first_token_time if streaming_first_token_time is not None and span.is_recording(): _set_span_attribute( span, FR_STREAMING_TIME_TO_GENERATE_MS, - round((end_time - streaming_first_token_time) * 1000), + round(max(0, end_time - streaming_first_token_time) * 1000), ) - duration = end_time - self.spans[run_id].start_time + duration = max(0, end_time - self.spans[run_id].start_time) vendor = span.attributes.get(GenAIAttributes.GEN_AI_SYSTEM, "Langchain") self.duration_histogram.record( duration, diff --git a/packages/opentelemetry-instrumentation-langchain/opentelemetry/instrumentation/langchain/span_utils.py b/packages/opentelemetry-instrumentation-langchain/opentelemetry/instrumentation/langchain/span_utils.py index 04c0809dee..c39a4d256e 100644 --- a/packages/opentelemetry-instrumentation-langchain/opentelemetry/instrumentation/langchain/span_utils.py +++ b/packages/opentelemetry-instrumentation-langchain/opentelemetry/instrumentation/langchain/span_utils.py @@ -38,7 +38,7 @@ class SpanHolder: workflow_name: str entity_name: str entity_path: str - start_time: float = field(default_factory=time.time) + start_time: float = field(default_factory=time.perf_counter) request_model: Optional[str] = None streaming_first_token_time: Optional[float] = None # ST-10 review-round-2 fix (2026-05-11): every context_api.attach() diff --git a/packages/opentelemetry-instrumentation-langchain/tests/test_streaming_latency_span_attrs.py b/packages/opentelemetry-instrumentation-langchain/tests/test_streaming_latency_span_attrs.py index 39e80b713c..15dcedede0 100644 --- a/packages/opentelemetry-instrumentation-langchain/tests/test_streaming_latency_span_attrs.py +++ b/packages/opentelemetry-instrumentation-langchain/tests/test_streaming_latency_span_attrs.py @@ -78,7 +78,7 @@ def fake_time(): return times.pop(0) return 103.0 - monkeypatch.setattr(callback_handler.time, "time", fake_time) + monkeypatch.setattr(callback_handler.time, "perf_counter", fake_time) handler.on_llm_new_token("hello", run_id=run_id) handler.on_llm_end(_llm_result(), run_id=run_id) @@ -90,6 +90,38 @@ def fake_time(): assert attrs[FR_STREAMING_TIME_TO_GENERATE_MS] == 1750 +def test_streaming_callbacks_clamp_negative_latency_attrs(monkeypatch): + _install_noop_fortifyroot(monkeypatch) + handler, exporter = _handler_and_exporter() + run_id = uuid4() + + handler.on_llm_start( + serialized={"id": ["langchain_openai", "llms", "base", "OpenAI"]}, + prompts=["hello"], + run_id=run_id, + invocation_params={"model": "gpt-4o-mini"}, + ) + handler.spans[run_id].start_time = 100.0 + + times = [99.0, 98.0] + + def fake_time(): + if times: + return times.pop(0) + return 98.0 + + monkeypatch.setattr(callback_handler.time, "perf_counter", fake_time) + + handler.on_llm_new_token("hello", run_id=run_id) + handler.on_llm_end(_llm_result(), run_id=run_id) + + spans = exporter.get_finished_spans() + assert len(spans) == 1 + attrs = spans[0].attributes + assert attrs[FR_STREAMING_TIME_TO_FIRST_TOKEN_MS] == 0 + assert attrs[FR_STREAMING_TIME_TO_GENERATE_MS] == 0 + + def test_non_streaming_llm_end_does_not_set_streaming_latency_attrs(monkeypatch): _install_noop_fortifyroot(monkeypatch) handler, exporter = _handler_and_exporter() @@ -102,7 +134,7 @@ def test_non_streaming_llm_end_does_not_set_streaming_latency_attrs(monkeypatch) invocation_params={"model": "gpt-4o-mini"}, ) handler.spans[run_id].start_time = 100.0 - monkeypatch.setattr(callback_handler.time, "time", lambda: 103.0) + monkeypatch.setattr(callback_handler.time, "perf_counter", lambda: 103.0) handler.on_llm_end(_llm_result(), run_id=run_id) diff --git a/packages/opentelemetry-instrumentation-litellm/opentelemetry/instrumentation/litellm/streaming_safety.py b/packages/opentelemetry-instrumentation-litellm/opentelemetry/instrumentation/litellm/streaming_safety.py index f350354f18..b5d9cb1533 100644 --- a/packages/opentelemetry-instrumentation-litellm/opentelemetry/instrumentation/litellm/streaming_safety.py +++ b/packages/opentelemetry-instrumentation-litellm/opentelemetry/instrumentation/litellm/streaming_safety.py @@ -17,6 +17,10 @@ FR_STREAMING_TIME_TO_GENERATE_MS = "fortifyroot.llm.streaming.time_to_generate_ms" +def _elapsed_seconds(start_time, end_time): + return max(0, end_time - start_time) + + def is_sync_streaming_response(kwargs, response) -> bool: """Check if this is a streaming response.""" if kwargs.get("stream"): @@ -61,7 +65,7 @@ def wrap_sync_streaming_response( request_type=request_type, ) complete_response = {"choices": [], "model": None, "usage": None} - stream_started_at = time.time() + stream_started_at = time.perf_counter() first_token_at = None try: @@ -69,11 +73,11 @@ def wrap_sync_streaming_response( _mask_streaming_chunk(streams, chunk) _accumulate_streaming_chunk(complete_response, chunk) if first_token_at is None and _chunk_has_output_text(chunk): - first_token_at = time.time() + first_token_at = time.perf_counter() _set_streaming_latency_attr( span, FR_STREAMING_TIME_TO_FIRST_TOKEN_MS, - first_token_at - stream_started_at, + _elapsed_seconds(stream_started_at, first_token_at), set_canonical_span_attribute, ) yield chunk @@ -81,7 +85,7 @@ def wrap_sync_streaming_response( _set_streaming_latency_attr( span, FR_STREAMING_TIME_TO_GENERATE_MS, - time.time() - first_token_at, + _elapsed_seconds(first_token_at, time.perf_counter()), set_canonical_span_attribute, ) _finalize_streaming_span(span, complete_response, set_response_attributes) @@ -117,7 +121,7 @@ async def wrap_async_streaming_response( request_type=request_type, ) complete_response = {"choices": [], "model": None, "usage": None} - stream_started_at = time.time() + stream_started_at = time.perf_counter() first_token_at = None try: @@ -125,11 +129,11 @@ async def wrap_async_streaming_response( _mask_streaming_chunk(streams, chunk) _accumulate_streaming_chunk(complete_response, chunk) if first_token_at is None and _chunk_has_output_text(chunk): - first_token_at = time.time() + first_token_at = time.perf_counter() _set_streaming_latency_attr( span, FR_STREAMING_TIME_TO_FIRST_TOKEN_MS, - first_token_at - stream_started_at, + _elapsed_seconds(stream_started_at, first_token_at), set_canonical_span_attribute, ) yield chunk @@ -137,7 +141,7 @@ async def wrap_async_streaming_response( _set_streaming_latency_attr( span, FR_STREAMING_TIME_TO_GENERATE_MS, - time.time() - first_token_at, + _elapsed_seconds(first_token_at, time.perf_counter()), set_canonical_span_attribute, ) _finalize_streaming_span(span, complete_response, set_response_attributes) diff --git a/packages/opentelemetry-instrumentation-litellm/tests/test_safety_hooks.py b/packages/opentelemetry-instrumentation-litellm/tests/test_safety_hooks.py index 37eb358567..f2ff0927a1 100644 --- a/packages/opentelemetry-instrumentation-litellm/tests/test_safety_hooks.py +++ b/packages/opentelemetry-instrumentation-litellm/tests/test_safety_hooks.py @@ -36,6 +36,7 @@ FR_STREAMING_TIME_TO_GENERATE_MS, _accumulate_streaming_chunk, _chunk_has_output_text, + _elapsed_seconds, _mask_streaming_chunk, is_async_streaming_response, is_sync_streaming_response, @@ -59,6 +60,10 @@ _FR_SPAN_NAME = "fortifyroot.litellm.safety" +def test_streaming_elapsed_seconds_clamps_negative_duration(): + assert _elapsed_seconds(10.0, 9.0) == 0 + + def setup_function(): clear_safety_handlers() @@ -784,7 +789,7 @@ def test_sync_streaming_wrapper_sets_streaming_latency_attrs(): ] with patch( - "opentelemetry.instrumentation.litellm.streaming_safety.time.time", + "opentelemetry.instrumentation.litellm.streaming_safety.time.perf_counter", side_effect=[100.0, 100.25, 100.9], ): list( @@ -819,7 +824,7 @@ async def _response(): ) with patch( - "opentelemetry.instrumentation.litellm.streaming_safety.time.time", + "opentelemetry.instrumentation.litellm.streaming_safety.time.perf_counter", side_effect=[200.0, 200.125, 200.5], ): yielded = [ @@ -849,7 +854,7 @@ def test_streaming_wrapper_leaves_latency_attrs_unset_for_empty_stream(): span = tracer.start_span("litellm.completion") with patch( - "opentelemetry.instrumentation.litellm.streaming_safety.time.time", + "opentelemetry.instrumentation.litellm.streaming_safety.time.perf_counter", side_effect=[300.0], ): list( diff --git a/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/shared/chat_wrappers.py b/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/shared/chat_wrappers.py index 23db70fd97..ef0274e881 100644 --- a/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/shared/chat_wrappers.py +++ b/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/shared/chat_wrappers.py @@ -83,6 +83,15 @@ FR_STREAMING_TIME_TO_FIRST_TOKEN_MS = "fortifyroot.llm.streaming.time_to_first_token_ms" FR_STREAMING_TIME_TO_GENERATE_MS = "fortifyroot.llm.streaming.time_to_generate_ms" + +def _elapsed_seconds(start_time, end_time): + return max(0, end_time - start_time) + + +def _elapsed_ms(start_time, end_time): + return round(_elapsed_seconds(start_time, end_time) * 1000) + + logger = logging.getLogger(__name__) @@ -117,7 +126,7 @@ def chat_wrapper( kwargs = _apply_prompt_safety(span, kwargs) run_async(_handle_request(span, kwargs, instance)) try: - start_time = time.time() + start_time = time.perf_counter() # ST-10.4 (review-driven 2026-05-16): set # OPENAI_DIRECT_RETRY_PARENT_ACTIVE_KEY alongside the # existing SUPPRESS_LANGUAGE_MODEL_INSTRUMENTATION_KEY so @@ -138,10 +147,10 @@ def chat_wrapper( response = wrapped(*args, **kwargs) finally: context_api.detach(token) - end_time = time.time() + end_time = time.perf_counter() except Exception as e: # pylint: disable=broad-except - end_time = time.time() - duration = end_time - start_time if "start_time" in locals() else 0 + end_time = time.perf_counter() + duration = _elapsed_seconds(start_time, end_time) if "start_time" in locals() else 0 attributes = { "error.type": e.__class__.__name__, @@ -188,7 +197,7 @@ def chat_wrapper( kwargs, ) - duration = end_time - start_time + duration = _elapsed_seconds(start_time, end_time) _apply_completion_safety(span, response) _handle_response( @@ -238,7 +247,7 @@ async def achat_wrapper( await _handle_request(span, kwargs, instance) try: - start_time = time.time() + start_time = time.perf_counter() # ST-10.4 (review-driven 2026-05-16): see sync chat_wrapper # above for the rationale on OPENAI_DIRECT_RETRY_PARENT_ACTIVE_KEY. attempt_ctx = context_api.set_value( @@ -252,10 +261,10 @@ async def achat_wrapper( response = await wrapped(*args, **kwargs) finally: context_api.detach(token) - end_time = time.time() + end_time = time.perf_counter() except Exception as e: # pylint: disable=broad-except - end_time = time.time() - duration = end_time - start_time if "start_time" in locals() else 0 + end_time = time.perf_counter() + duration = _elapsed_seconds(start_time, end_time) if "start_time" in locals() else 0 common_attributes = Config.get_common_metrics_attributes() attributes = { @@ -304,7 +313,7 @@ async def achat_wrapper( kwargs, ) - duration = end_time - start_time + duration = _elapsed_seconds(start_time, end_time) await asyncio.to_thread(_apply_completion_safety, span, response) # FR: async safety _handle_response( @@ -781,10 +790,10 @@ def _process_item(self, item): name=f"{SpanAttributes.LLM_CONTENT_COMPLETION_CHUNK}") if self._first_token: - self._time_of_first_token = time.time() + self._time_of_first_token = time.perf_counter() if self._streaming_time_to_first_token: self._streaming_time_to_first_token.record( - self._time_of_first_token - self._start_time, + _elapsed_seconds(self._start_time, self._time_of_first_token), attributes=self._shared_attributes(), ) self._first_token = False @@ -794,7 +803,7 @@ def _process_item(self, item): _set_span_attribute( self._span, FR_STREAMING_TIME_TO_FIRST_TOKEN_MS, - round((self._time_of_first_token - self._start_time) * 1000), + _elapsed_ms(self._start_time, self._time_of_first_token), ) item = self._streaming_safety.process_chunk(item) @@ -841,7 +850,7 @@ def _process_complete_response(self): # duration metrics if self._start_time and isinstance(self._start_time, (float, int)): - duration = time.time() - self._start_time + duration = _elapsed_seconds(self._start_time, time.perf_counter()) else: duration = None if duration and isinstance(duration, (float, int)) and self._duration_histogram: @@ -850,7 +859,7 @@ def _process_complete_response(self): ) if self._streaming_time_to_generate and self._time_of_first_token: self._streaming_time_to_generate.record( - time.time() - self._time_of_first_token, + _elapsed_seconds(self._time_of_first_token, time.perf_counter()), attributes=self._shared_attributes(), ) @@ -874,7 +883,7 @@ def _process_complete_response(self): _set_span_attribute( self._span, FR_STREAMING_TIME_TO_GENERATE_MS, - round((time.time() - self._time_of_first_token) * 1000), + _elapsed_ms(self._time_of_first_token, time.perf_counter()), ) self._span.set_status(Status(StatusCode.OK)) @@ -924,7 +933,7 @@ def _record_partial_metrics(self): """Record metrics based on available partial data""" # Always record duration if we have start time if self._start_time and isinstance(self._start_time, (float, int)) and self._duration_histogram: - duration = time.time() - self._start_time + duration = _elapsed_seconds(self._start_time, time.perf_counter()) self._duration_histogram.record( duration, attributes=self._shared_attributes() ) @@ -981,17 +990,17 @@ def _build_from_streaming_response( item_to_yield = streaming_safety.process_chunk(item) if first_token: - time_of_first_token = time.time() + time_of_first_token = time.perf_counter() if streaming_time_to_first_token: streaming_time_to_first_token.record( - time_of_first_token - start_time) + _elapsed_seconds(start_time, time_of_first_token)) first_token = False # FR: also persist TTFT (ms, int) on the span for RDS extraction. if span and span.is_recording(): _set_span_attribute( span, FR_STREAMING_TIME_TO_FIRST_TOKEN_MS, - round((time_of_first_token - start_time) * 1000), + _elapsed_ms(start_time, time_of_first_token), ) _accumulate_stream_items(item, complete_response) @@ -1020,13 +1029,13 @@ def _build_from_streaming_response( # duration metrics if start_time and isinstance(start_time, (float, int)): - duration = time.time() - start_time + duration = _elapsed_seconds(start_time, time.perf_counter()) else: duration = None if duration and isinstance(duration, (float, int)) and duration_histogram: duration_histogram.record(duration, attributes=shared_attributes) if streaming_time_to_generate and time_of_first_token: - streaming_time_to_generate.record(time.time() - time_of_first_token) + streaming_time_to_generate.record(_elapsed_seconds(time_of_first_token, time.perf_counter())) # FR: persist STTG (ms, int) on the span for RDS extraction, only when a # first token actually arrived (not first_token). @@ -1034,7 +1043,7 @@ def _build_from_streaming_response( _set_span_attribute( span, FR_STREAMING_TIME_TO_GENERATE_MS, - round((time.time() - time_of_first_token) * 1000), + _elapsed_ms(time_of_first_token, time.perf_counter()), ) _set_response_attributes(span, complete_response) @@ -1075,17 +1084,17 @@ async def _abuild_from_streaming_response( item_to_yield = streaming_safety.process_chunk(item) if first_token: - time_of_first_token = time.time() + time_of_first_token = time.perf_counter() if streaming_time_to_first_token: streaming_time_to_first_token.record( - time_of_first_token - start_time) + _elapsed_seconds(start_time, time_of_first_token)) first_token = False # FR: also persist TTFT (ms, int) on the span for RDS extraction. if span and span.is_recording(): _set_span_attribute( span, FR_STREAMING_TIME_TO_FIRST_TOKEN_MS, - round((time_of_first_token - start_time) * 1000), + _elapsed_ms(start_time, time_of_first_token), ) _accumulate_stream_items(item, complete_response) @@ -1114,13 +1123,13 @@ async def _abuild_from_streaming_response( # duration metrics if start_time and isinstance(start_time, (float, int)): - duration = time.time() - start_time + duration = _elapsed_seconds(start_time, time.perf_counter()) else: duration = None if duration and isinstance(duration, (float, int)) and duration_histogram: duration_histogram.record(duration, attributes=shared_attributes) if streaming_time_to_generate and time_of_first_token: - streaming_time_to_generate.record(time.time() - time_of_first_token) + streaming_time_to_generate.record(_elapsed_seconds(time_of_first_token, time.perf_counter())) # FR: persist STTG (ms, int) on the span for RDS extraction, only when a # first token actually arrived (not first_token). @@ -1128,7 +1137,7 @@ async def _abuild_from_streaming_response( _set_span_attribute( span, FR_STREAMING_TIME_TO_GENERATE_MS, - round((time.time() - time_of_first_token) * 1000), + _elapsed_ms(time_of_first_token, time.perf_counter()), ) _set_response_attributes(span, complete_response) diff --git a/packages/opentelemetry-instrumentation-openai/tests/test_streaming_latency_span_attrs.py b/packages/opentelemetry-instrumentation-openai/tests/test_streaming_latency_span_attrs.py index fcb099c892..2ce32d379d 100644 --- a/packages/opentelemetry-instrumentation-openai/tests/test_streaming_latency_span_attrs.py +++ b/packages/opentelemetry-instrumentation-openai/tests/test_streaming_latency_span_attrs.py @@ -66,13 +66,17 @@ def _recording_span(): return span, exporter +def test_elapsed_ms_clamps_negative_duration(): + assert cw._elapsed_ms(10.0, 9.0) == 0 + + def test_streaming_sets_positive_integer_latency_attrs(monkeypatch): """A streaming response that yields tokens sets both attributes as non-negative integer milliseconds on the single LLM span.""" _neutralize_helpers(monkeypatch) span, exporter = _recording_span() - start = time.time() - 0.05 # 50ms ago => TTFT/STTG are clearly > 0 + start = time.perf_counter() - 0.05 # 50ms ago => TTFT/STTG are clearly > 0 ttft_hist, sttg_hist = _StubHistogram(), _StubHistogram() chunks = [object(), object(), object()] @@ -109,7 +113,7 @@ def test_streaming_latency_attrs_do_not_require_metrics_histograms(monkeypatch): iter([object(), object()]), streaming_time_to_first_token=None, streaming_time_to_generate=None, - start_time=time.time() - 0.05, + start_time=time.perf_counter() - 0.05, ) ) @@ -133,7 +137,7 @@ def test_empty_stream_sets_no_latency_attrs(monkeypatch): iter([]), streaming_time_to_first_token=_StubHistogram(), streaming_time_to_generate=_StubHistogram(), - start_time=time.time(), + start_time=time.perf_counter(), ) )