diff --git a/litellm/integrations/otel/mount.py b/litellm/integrations/otel/mount.py index ac647c2c4f6..1628357c028 100644 --- a/litellm/integrations/otel/mount.py +++ b/litellm/integrations/otel/mount.py @@ -93,8 +93,8 @@ def instrument_fastapi_app(app: Any) -> None: Safe no-op when the V2 gate is off or ``opentelemetry-instrumentation-fastapi`` is unavailable. This MUST be called at app-creation time — once the lifespan - runs, the middleware stack is frozen and ``instrument_app`` raises "Cannot add - middleware after an application has started". + runs, the middleware stack is frozen and middleware added later never runs, so + a late call is skipped without registering anything. No ``TracerProvider`` is passed, so the instrumentation binds to the OTel global ``ProxyTracerProvider``; the proxy publishes the real provider as the global @@ -104,6 +104,9 @@ def instrument_fastapi_app(app: Any) -> None: try: if not is_otel_v2_enabled(): return + if app.middleware_stack is not None: + verbose_logger.debug("Skipping OTel V2 FastAPI instrumentation: the app has already started") + return # Lazy: only the V2-enabled path needs the optional # ``opentelemetry-instrumentation-fastapi`` package, which is not part of the diff --git a/litellm/integrations/otel/plumbing/otlp_json.py b/litellm/integrations/otel/plumbing/otlp_json.py index b4b659f1e01..cf63c22e0d5 100644 --- a/litellm/integrations/otel/plumbing/otlp_json.py +++ b/litellm/integrations/otel/plumbing/otlp_json.py @@ -6,15 +6,20 @@ retry loop and swaps the payload for OTLP/JSON (enums as integers, ids as hex). import base64 import json +import logging from collections.abc import Mapping, Sequence from types import MappingProxyType from typing import Final, TypeAlias from google.protobuf.json_format import MessageToDict from opentelemetry.exporter.otlp.proto.common.trace_encoder import encode_spans +from opentelemetry.exporter.otlp.proto.http._common import RequestPayloadTooLargeError from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter from opentelemetry.sdk.trace import ReadableSpan +from opentelemetry.sdk.trace.export import SpanExportResult +from opentelemetry.semconv.attributes.http_attributes import HTTP_RESPONSE_STATUS_CODE +_logger: Final = logging.getLogger(__name__) JSON_CONTENT_TYPE: Final = "application/json" _HEX_ID_KEYS: Final = frozenset({"traceId", "spanId", "parentSpanId"}) @@ -62,9 +67,48 @@ def encode_spans_json(spans: Sequence[ReadableSpan]) -> bytes: class OTLPJsonSpanExporter(OTLPSpanExporter): - def __init__(self, endpoint: str | None, headers: dict[str, str]) -> None: # mutable-ok: SDK __init__ takes Dict - super().__init__(endpoint=endpoint, headers=headers) - self._session.headers["Content-Type"] = JSON_CONTENT_TYPE + def __init__( + self, + endpoint: str | None, + headers: Mapping[str, str], + *, + max_request_size: int | None = None, + ) -> None: + super().__init__( + endpoint=endpoint, + headers={**{key.lower(): value for key, value in headers.items()}, "content-type": JSON_CONTENT_TYPE}, + max_request_size=max_request_size, + ) - def _serialize_spans(self, spans: Sequence[ReadableSpan]) -> bytes: - return encode_spans_json(spans) + def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult: + if self._shutdown: + _logger.warning("Exporter already shutdown, ignoring batch") + return SpanExportResult.FAILURE + + with self._metrics.export_operation(len(spans)) as result: + try: + serialized_data: Final = encode_spans_json(spans) + except Exception as error: # noqa: BLE001 # SDK export contract reports encoding failures as results + _logger.error("Failed to encode span batch: %s", error) + result.error = error + return SpanExportResult.FAILURE + + if 0 < self._max_request_size < len(serialized_data): + result.error = RequestPayloadTooLargeError( + f"Serialized span request size {len(serialized_data)} bytes " + f"exceeds max_request_size {self._max_request_size} bytes." + ) + _logger.warning("Dropping span batch: %s", result.error) + return SpanExportResult.FAILURE + + export_result: Final = self._client.export(serialized_data) + if not export_result.success: + result.error = export_result.error + result.error_attrs = ( + {HTTP_RESPONSE_STATUS_CODE: export_result.status_code} + if export_result.status_code is not None + else None + ) + return SpanExportResult.FAILURE + + return SpanExportResult.SUCCESS diff --git a/litellm/integrations/otel/presets/agentops.py b/litellm/integrations/otel/presets/agentops.py index 965213f2ee4..142a7bd5dc6 100644 --- a/litellm/integrations/otel/presets/agentops.py +++ b/litellm/integrations/otel/presets/agentops.py @@ -73,6 +73,7 @@ def agentops_preset( def _build_agentops_exporter(spec: ExporterSpec) -> Any: """Factory for the ``agentops`` exporter kind: a lazy-auth OTLP/HTTP exporter.""" + import requests from opentelemetry.exporter.otlp.proto.http.trace_exporter import ( OTLPSpanExporter, ) @@ -86,7 +87,8 @@ def _build_agentops_exporter(spec: ExporterSpec) -> Any: """ def __init__(self, *, endpoint: str | None, api_key: str | None) -> None: - super().__init__(endpoint=endpoint) + self._agentops_session = requests.Session() + super().__init__(endpoint=endpoint, session=self._agentops_session) self._agentops_api_key = api_key self._auth_resolved = False @@ -99,10 +101,7 @@ def _build_agentops_exporter(spec: ExporterSpec) -> Any: try: token: Final = _fetch_agentops_jwt(self._agentops_api_key).get("token") if token: - # ``_session`` is the requests.Session the base exporter - # POSTs through; updating its Authorization header is how the - # minted JWT reaches every subsequent export. - self._session.headers["Authorization"] = f"Bearer {token}" + self._agentops_session.headers["Authorization"] = f"Bearer {token}" except Exception as e: verbose_logger.debug("AgentOps JWT fetch failed: %s", e) diff --git a/tests/test_litellm/integrations/otel/test_otel_v2_components.py b/tests/test_litellm/integrations/otel/test_otel_v2_components.py index a6fe9e34163..d2202a3b3be 100644 --- a/tests/test_litellm/integrations/otel/test_otel_v2_components.py +++ b/tests/test_litellm/integrations/otel/test_otel_v2_components.py @@ -8,8 +8,11 @@ from collections.abc import Iterator from contextvars import Context as ContextVarContext from dataclasses import replace from http.server import BaseHTTPRequestHandler, HTTPServer, ThreadingHTTPServer +from typing import Final import pytest +import requests +import requests_mock pytest.importorskip("opentelemetry") @@ -20,11 +23,12 @@ from opentelemetry import baggage # noqa: E402 from opentelemetry.context import attach, detach # noqa: E402 from opentelemetry.sdk.metrics import MeterProvider # noqa: E402 from opentelemetry.sdk.metrics.export import InMemoryMetricReader # noqa: E402 -from opentelemetry.sdk.trace import TracerProvider # noqa: E402 +from opentelemetry.sdk.trace import ReadableSpan, TracerProvider # noqa: E402 # optional SDK checked above from opentelemetry.sdk.trace.export import ( # noqa: E402 BatchSpanProcessor, ConsoleSpanExporter, SimpleSpanProcessor, + SpanExportResult, ) from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( # noqa: E402 InMemorySpanExporter, @@ -36,6 +40,7 @@ from opentelemetry.trace.propagation.tracecontext import ( # noqa: E402 from litellm.integrations.otel.plumbing import context as ctx_mod # noqa: E402 from litellm.integrations.otel.plumbing import providers # noqa: E402 +from litellm.integrations.otel.plumbing.otlp_json import OTLPJsonSpanExporter # noqa: E402 # optional SDK checked above from litellm.integrations.otel.model.config import OpenTelemetryV2Config # noqa: E402 from litellm.integrations.otel.mappers.genai import GenAIMapper # noqa: E402 from litellm.integrations.otel.mappers.legacy import LegacyMapper # noqa: E402 @@ -704,7 +709,9 @@ def _export_one_trace_to_local_collector(exporter_kind: str) -> tuple[list[dict] class Collector(BaseHTTPRequestHandler): def do_POST(self): body = self.rfile.read(int(self.headers["Content-Length"])) - received.append({"path": self.path, "headers": dict(self.headers), "body": body}) + received.append( + {"path": self.path, "headers": requests.structures.CaseInsensitiveDict(self.headers), "body": body} + ) self.send_response(200) self.end_headers() @@ -764,6 +771,86 @@ def test_http_json_exporter_posts_otlp_json_to_traces_endpoint(): assert child["parentSpanId"] == format(parent_id, "016x") +@pytest.mark.parametrize( + ("statuses", "expected"), + [ + ((200,), SpanExportResult.SUCCESS), + ((400,), SpanExportResult.FAILURE), + ((503, 200), SpanExportResult.SUCCESS), + ], +) +def test_http_json_exporter_preserves_http_failure_and_retry_results( + statuses: tuple[int, ...], expected: SpanExportResult +) -> None: + endpoint: Final = "https://collector.test/v1/traces" + with requests_mock.Mocker() as collector: + collector.register_uri( + "POST", + endpoint, + [{"status_code": status, "headers": {"Retry-After": "0"}} for status in statuses], + ) + exporter: Final = OTLPJsonSpanExporter( + endpoint=endpoint, + headers={ + "CONTENT-TYPE": "application/x-protobuf", + "x-collector-token": "secret", + }, + ) + try: + assert exporter.export(()) is expected + finally: + exporter.shutdown() + assert len(collector.request_history) == len(statuses) + for request in collector.request_history: + assert request.headers["Content-Type"] == "application/json" + assert request.headers["x-collector-token"] == "secret" + assert request.json() == {"resourceSpans": []} + + +@pytest.mark.parametrize( + ("size_limit", "expected", "request_count"), + [ + (0, SpanExportResult.SUCCESS, 1), + (len(b'{"resourceSpans":[]}'), SpanExportResult.SUCCESS, 1), + (len(b'{"resourceSpans":[]}') - 1, SpanExportResult.FAILURE, 0), + ], +) +def test_http_json_exporter_checks_json_size_before_sending( + size_limit: int, expected: SpanExportResult, request_count: int +) -> None: + endpoint: Final = "https://collector.test/v1/traces" + with requests_mock.Mocker() as collector: + collector.register_uri("POST", endpoint, status_code=200) + exporter: Final = OTLPJsonSpanExporter(endpoint, {}, max_request_size=size_limit) + try: + assert exporter.export(()) is expected + finally: + exporter.shutdown() + assert len(collector.request_history) == request_count + + +def test_http_json_exporter_rejects_export_after_shutdown() -> None: + endpoint: Final = "https://collector.test/v1/traces" + with requests_mock.Mocker() as collector: + collector.register_uri("POST", endpoint, status_code=200) + exporter: Final = OTLPJsonSpanExporter(endpoint, {}) + exporter.shutdown() + assert exporter.export(()) is SpanExportResult.FAILURE + assert collector.request_history == [] + + +def test_http_json_exporter_returns_failure_for_unencodable_spans() -> None: + endpoint: Final = "https://collector.test/v1/traces" + with requests_mock.Mocker() as collector: + collector.register_uri("POST", endpoint, status_code=200) + exporter: Final = OTLPJsonSpanExporter(endpoint, {}) + try: + assert exporter.export((ReadableSpan("missing-context"),)) is SpanExportResult.FAILURE + finally: + exporter.shutdown() + assert collector.request_history == [] + + def test_http_protobuf_exporter_still_posts_protobuf(): (_child_request, parent_request), (trace_id, _parent_id, _child_id) = _export_one_trace_to_local_collector( "http/protobuf" diff --git a/tests/test_litellm/integrations/otel/test_otel_v2_presets.py b/tests/test_litellm/integrations/otel/test_otel_v2_presets.py index 58cfc1ceb3f..f0ca6a58d44 100644 --- a/tests/test_litellm/integrations/otel/test_otel_v2_presets.py +++ b/tests/test_litellm/integrations/otel/test_otel_v2_presets.py @@ -2,13 +2,19 @@ event loop: the preset does no network I/O, and a custom exporter mints the JWT lazily on its first export (in the BatchSpanProcessor worker thread).""" +import json + import httpx import pytest +import requests_mock +import respx +from opentelemetry.sdk.trace.export import SpanExportResult from litellm.integrations.otel.plumbing import providers from litellm.integrations.otel.model.config import ExporterSpec from litellm.integrations.otel.presets import agentops as agentops_mod from litellm.integrations.otel.presets.agentops import ( + _AGENTOPS_AUTH_ENDPOINT, _AGENTOPS_ENDPOINT, _AGENTOPS_EXPORTER_KIND, _build_agentops_exporter, @@ -75,50 +81,44 @@ def test_dynamic_cred_presets_tag_exporter_with_matching_owner(monkeypatch): ) -def test_agentops_exporter_mints_jwt_lazily(monkeypatch): - pytest.importorskip("opentelemetry.exporter.otlp.proto.http.trace_exporter") - monkeypatch.setattr( - agentops_mod, "_fetch_agentops_jwt", lambda _k: {"token": "jwt-xyz"} - ) - spec = ExporterSpec( - kind=_AGENTOPS_EXPORTER_KIND, - endpoint=_AGENTOPS_ENDPOINT, - options={"api_key": "ak"}, - ) - exporter = _build_agentops_exporter(spec) - - # No auth header until the first export triggers the (off-loop) fetch. - assert "Authorization" not in exporter._session.headers - exporter._ensure_authenticated() - assert exporter._session.headers["Authorization"] == "Bearer jwt-xyz" - - # Cached: a second resolution does not re-fetch. - calls = [] - monkeypatch.setattr( - agentops_mod, - "_fetch_agentops_jwt", - lambda k: calls.append(k) or {"token": "again"}, - ) - exporter._ensure_authenticated() - assert calls == [] - - -def test_agentops_exporter_tolerates_fetch_failure(monkeypatch): - pytest.importorskip("opentelemetry.exporter.otlp.proto.http.trace_exporter") - - def _raise(_k): - raise RuntimeError("auth down") - - monkeypatch.setattr(agentops_mod, "_fetch_agentops_jwt", _raise) - exporter = _build_agentops_exporter( +def _agentops_exporter(): + return _build_agentops_exporter( ExporterSpec( kind=_AGENTOPS_EXPORTER_KIND, endpoint=_AGENTOPS_ENDPOINT, options={"api_key": "ak"}, ) ) - exporter._ensure_authenticated() # must not raise - assert "Authorization" not in exporter._session.headers + + +def test_agentops_exporter_mints_the_jwt_once_on_first_export_and_sends_it_with_every_batch(): + pytest.importorskip("opentelemetry.exporter.otlp.proto.http.trace_exporter") + with respx.mock() as auth, requests_mock.Mocker() as collector: + token_route = auth.post(_AGENTOPS_AUTH_ENDPOINT).respond(200, json={"token": "jwt-xyz"}) + collector.post(_AGENTOPS_ENDPOINT, status_code=200) + exporter = _agentops_exporter() + assert token_route.call_count == 0, "building the exporter must not fetch the JWT" + + results = [exporter.export(()), exporter.export(())] + + assert results == [SpanExportResult.SUCCESS, SpanExportResult.SUCCESS] + assert [json.loads(call.request.content) for call in token_route.calls] == [{"api_key": "ak"}] + assert [request.headers.get("Authorization") for request in collector.request_history] == [ + "Bearer jwt-xyz", + "Bearer jwt-xyz", + ] + + +def test_agentops_exporter_still_exports_without_auth_when_the_jwt_fetch_fails(): + pytest.importorskip("opentelemetry.exporter.otlp.proto.http.trace_exporter") + with respx.mock() as auth, requests_mock.Mocker() as collector: + auth.post(_AGENTOPS_AUTH_ENDPOINT).mock(side_effect=httpx.ConnectError("auth down")) + collector.post(_AGENTOPS_ENDPOINT, status_code=200) + + result = _agentops_exporter().export(()) + + assert result is SpanExportResult.SUCCESS + assert [request.headers.get("Authorization") for request in collector.request_history] == [None] def test_fetch_jwt_uses_owned_client_not_shared_pool(monkeypatch): diff --git a/tests/test_litellm/integrations/test_langfuse_otel.py b/tests/test_litellm/integrations/test_langfuse_otel.py index 0a9ce55fe16..46160204e3a 100644 --- a/tests/test_litellm/integrations/test_langfuse_otel.py +++ b/tests/test_litellm/integrations/test_langfuse_otel.py @@ -483,6 +483,22 @@ class TestLangfuseOtelIntegration: # Endpoint assertion removed as side effect is gone +def _exported_auth_headers(exporter): + import requests_mock + from opentelemetry.sdk.trace.export import SpanExportResult + + with requests_mock.Mocker() as collector: + collector.post(requests_mock.ANY, status_code=200) + assert exporter.export(()) is SpanExportResult.SUCCESS + return [ + ( + request.url, + {name: request.headers[name] for name in ("Authorization", "x-langfuse-ingestion-version")}, + ) + for request in collector.request_history + ] + + class TestLangfuseOtelKeyDynamicConfig: """Key/team-scoped Langfuse credentials must define the full export target (OTLP endpoint + auth), not just auth headers on the init-time exporter.""" @@ -587,15 +603,15 @@ class TestLangfuseOtelKeyDynamicConfig: exporter = span_processors[0].span_exporter assert isinstance(exporter, OTLPSpanExporter) - assert exporter._endpoint == "https://langfuse.example.com/api/public/otel/v1/traces" - import base64 expected_auth = base64.b64encode(b"key_public:key_secret").decode() - assert exporter._headers == { - "Authorization": f"Basic {expected_auth}", - "x-langfuse-ingestion-version": "4", - } + assert _exported_auth_headers(exporter) == [ + ( + "https://langfuse.example.com/api/public/otel/v1/traces", + {"Authorization": f"Basic {expected_auth}", "x-langfuse-ingestion-version": "4"}, + ) + ] def test_key_dynamic_params_reuse_cached_provider(self): with self._clean_env(): @@ -648,10 +664,12 @@ class TestLangfuseOtelKeyDynamicConfig: provider = next(iter(logger._tracer_provider_cache.values())).provider exporter = provider._active_span_processor._span_processors[0].span_exporter assert isinstance(exporter, OTLPSpanExporter) - assert exporter._headers == { - "Authorization": f"Basic {secret}", - "x-langfuse-ingestion-version": "4", - } + assert _exported_auth_headers(exporter) == [ + ( + "https://langfuse.example.com/api/public/otel/v1/traces", + {"Authorization": f"Basic {secret}", "x-langfuse-ingestion-version": "4"}, + ) + ] class TestLangfuseOtelResponsesAPI: diff --git a/tests/test_litellm/integrations/test_opentelemetry.py b/tests/test_litellm/integrations/test_opentelemetry.py index bea9a38e9dd..ad60ab89298 100644 --- a/tests/test_litellm/integrations/test_opentelemetry.py +++ b/tests/test_litellm/integrations/test_opentelemetry.py @@ -10,18 +10,19 @@ import unittest import weakref from datetime import datetime, timedelta, timezone from types import MappingProxyType +from typing import Final from parameterized import parameterized from unittest.mock import MagicMock, patch # Adds the grandparent directory to sys.path to allow importing project modules from opentelemetry import trace -from opentelemetry.sdk._logs import LogData +from opentelemetry.sdk._logs import ReadableLogRecord from opentelemetry.sdk._logs import LoggerProvider as OTLoggerProvider from opentelemetry.sdk._logs.export import InMemoryLogExporter, SimpleLogRecordProcessor from opentelemetry.sdk.metrics import MeterProvider from opentelemetry.sdk.metrics.export import InMemoryMetricReader, MetricsData from opentelemetry.sdk.trace import TracerProvider -from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export import BatchSpanProcessor, SimpleSpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter import litellm @@ -6170,7 +6171,7 @@ class TestOpenTelemetryProviderlessCallAttributes(unittest.TestCase): time.sleep(self.POLL_INTERVAL) return None - def _emitted_log_records(self, semconv_opt_in: str) -> tuple[LogData, ...]: + def _emitted_log_records(self, semconv_opt_in: str) -> tuple[ReadableLogRecord, ...]: log_exporter = InMemoryLogExporter() logger_provider = OTLoggerProvider() logger_provider.add_log_record_processor(SimpleLogRecordProcessor(log_exporter)) @@ -6244,7 +6245,12 @@ class TestDynamicTracerProviderCache(unittest.TestCase): must be bounded and must shut down whatever it drops (LIT-5437: threads accumulated until pods were OOMKilled).""" - BSP_THREAD_NAME = "OtelBatchSpanProcessor" + def setUp(self) -> None: + processor: Final = BatchSpanProcessor(InMemorySpanExporter()) + try: + self.bsp_thread_name = processor._batch_processor._worker_thread.name + finally: + processor.shutdown() def _logger(self, cap=3, exporter="console"): logger = OpenTelemetry( @@ -6261,7 +6267,7 @@ class TestDynamicTracerProviderCache(unittest.TestCase): logger._tracer_provider_cache.clear() def _live_exporter_threads(self): - return [t for t in threading.enumerate() if t.name == self.BSP_THREAD_NAME] + return [t for t in threading.enumerate() if t.name == self.bsp_thread_name] def _wait_for_exporter_threads(self, expected, timeout=10.0): """Dropped providers are shut down off-thread, so poll instead of sleeping.""" @@ -6282,8 +6288,6 @@ class TestDynamicTracerProviderCache(unittest.TestCase): logger._get_tracer_with_dynamic_headers({"authorization": f"Basic tenant-{i}"}) self.assertEqual(len(logger._tracer_provider_cache), 3) - # Guards the thread-name constant: a rename upstream would make this read 0 and the - # bound assertion below would pass while measuring nothing. self.assertGreaterEqual(len(self._live_exporter_threads()), 1) self.assertLessEqual(self._wait_for_exporter_threads(before + 3) - before, 3)