diff --git a/litellm/integrations/otel/mount.py b/litellm/integrations/otel/mount.py index 9340f6e9e15..3c770b219b5 100644 --- a/litellm/integrations/otel/mount.py +++ b/litellm/integrations/otel/mount.py @@ -104,8 +104,8 @@ def instrument_fastapi_app(app: "FastAPI") -> 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 @@ -115,6 +115,9 @@ def instrument_fastapi_app(app: "FastAPI") -> 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 bc6d7d435b0..fc0351d5049 100644 --- a/litellm/integrations/otel/plumbing/otlp_json.py +++ b/litellm/integrations/otel/plumbing/otlp_json.py @@ -6,6 +6,7 @@ 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 @@ -13,9 +14,13 @@ from typing import Final, TypeAlias import requests 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"}) @@ -66,12 +71,49 @@ class OTLPJsonSpanExporter(OTLPSpanExporter): def __init__( self, endpoint: str | None, - headers: dict[str, str], # mutable-ok: SDK __init__ takes Dict + headers: Mapping[str, str], certificate_file: str | None = None, session: "requests.Session | None" = None, + *, + max_request_size: int | None = None, ) -> None: - super().__init__(endpoint=endpoint, headers=headers, certificate_file=certificate_file, session=session) - self._session.headers["Content-Type"] = JSON_CONTENT_TYPE + super().__init__( + endpoint=endpoint, + headers={**{key.lower(): value for key, value in headers.items()}, "content-type": JSON_CONTENT_TYPE}, + certificate_file=certificate_file, + session=session, + 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 58123656caa..de8e1405339 100644 --- a/litellm/integrations/otel/presets/agentops.py +++ b/litellm/integrations/otel/presets/agentops.py @@ -76,6 +76,7 @@ def agentops_preset( def _build_agentops_exporter(spec: ExporterSpec) -> SpanExporter: """Factory for the ``agentops`` exporter kind: a lazy-auth OTLP/HTTP exporter.""" + import requests from opentelemetry.exporter.otlp.proto.http.trace_exporter import ( OTLPSpanExporter, ) @@ -89,7 +90,8 @@ def _build_agentops_exporter(spec: ExporterSpec) -> SpanExporter: """ 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 @@ -102,10 +104,7 @@ def _build_agentops_exporter(spec: ExporterSpec) -> SpanExporter: 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/unit/integrations/otel/test_otel_v2_components.py b/tests/unit/integrations/otel/test_otel_v2_components.py index fb7be0dda14..17e0cc9aab8 100644 --- a/tests/unit/integrations/otel/test_otel_v2_components.py +++ b/tests/unit/integrations/otel/test_otel_v2_components.py @@ -13,6 +13,7 @@ from http.server import BaseHTTPRequestHandler, HTTPServer, ThreadingHTTPServer import pytest import requests +from requests_mock import Adapter pytest.importorskip("opentelemetry") @@ -21,16 +22,18 @@ from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ( # noqa: ) from opentelemetry import baggage # noqa: E402 from opentelemetry.context import attach, detach # noqa: E402 -from opentelemetry._logs.severity import SeverityNumber # noqa: E402 -from opentelemetry.sdk._logs import LogData, LogRecord # noqa: E402 -from opentelemetry.sdk._logs.export import LogExportResult # noqa: E402 +from opentelemetry._logs import LogRecord, SeverityNumber # noqa: E402 # optional SDK checked above +from opentelemetry.sdk._logs import ReadableLogRecord # noqa: E402 # optional SDK checked above +from opentelemetry.sdk._logs.export import LogRecordExportResult # noqa: E402 # optional SDK checked above +from opentelemetry.sdk.resources import Resource # noqa: E402 # optional SDK checked above 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, @@ -45,6 +48,7 @@ import litellm # noqa: E402 from tests.unit.integrations.conftest import TlsSink # 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 @@ -717,7 +721,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() @@ -777,6 +783,95 @@ 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: + adapter: Final = Adapter() + endpoint: Final = "https://collector.test/v1/traces" + adapter.register_uri( + "POST", + endpoint, + [{"status_code": status, "headers": {"Retry-After": "0"}} for status in statuses], + ) + with requests.Session() as session: + session.mount("https://", adapter) + exporter: Final = OTLPJsonSpanExporter( + endpoint=endpoint, + headers={ + "CONTENT-TYPE": "application/x-protobuf", + "x-collector-token": "secret", + }, + session=session, + ) + try: + assert exporter.export(()) is expected + finally: + exporter.shutdown() + assert len(adapter.request_history) == len(statuses) + for request in adapter.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: + adapter: Final = Adapter() + endpoint: Final = "https://collector.test/v1/traces" + adapter.register_uri("POST", endpoint, status_code=200) + with requests.Session() as session: + session.mount("https://", adapter) + exporter: Final = OTLPJsonSpanExporter(endpoint, {}, session=session, max_request_size=size_limit) + try: + assert exporter.export(()) is expected + finally: + exporter.shutdown() + assert len(adapter.request_history) == request_count + + +def test_http_json_exporter_rejects_export_after_shutdown() -> None: + adapter: Final = Adapter() + endpoint: Final = "https://collector.test/v1/traces" + adapter.register_uri("POST", endpoint, status_code=200) + with requests.Session() as session: + session.mount("https://", adapter) + exporter: Final = OTLPJsonSpanExporter(endpoint, {}, session=session) + exporter.shutdown() + assert exporter.export(()) is SpanExportResult.FAILURE + assert adapter.request_history == [] + + +def test_http_json_exporter_returns_failure_for_unencodable_spans() -> None: + adapter: Final = Adapter() + endpoint: Final = "https://collector.test/v1/traces" + adapter.register_uri("POST", endpoint, status_code=200) + with requests.Session() as session: + session.mount("https://", adapter) + exporter: Final = OTLPJsonSpanExporter(endpoint, {}, session=session) + try: + assert exporter.export((ReadableSpan("missing-context"),)) is SpanExportResult.FAILURE + finally: + exporter.shutdown() + assert adapter.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" @@ -1613,9 +1708,13 @@ def test_v2_otlp_http_log_export_trusts_ssl_cert_file(monkeypatch: pytest.Monkey severity_number=SeverityNumber.INFO, body="v2-tls-test", ) - log_data = LogData(log_record=record, instrumentation_scope=InstrumentationScope("v2-tls-test")) + log_data: Final = ReadableLogRecord( + log_record=record, + resource=Resource({}), + instrumentation_scope=InstrumentationScope("v2-tls-test"), + ) result = exporter.export([log_data]) - assert result is LogExportResult.SUCCESS, f"log export failed: {result}" + assert result is LogRecordExportResult.SUCCESS, f"log export failed: {result}" assert tls_sink.received.get(timeout=5) == "/v1/logs" finally: exporter.shutdown() diff --git a/tests/unit/integrations/otel/test_otel_v2_presets.py b/tests/unit/integrations/otel/test_otel_v2_presets.py index a060cdf3648..19b37e2e461 100644 --- a/tests/unit/integrations/otel/test_otel_v2_presets.py +++ b/tests/unit/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/unit/integrations/test_langfuse_otel.py b/tests/unit/integrations/test_langfuse_otel.py index 0a9ce55fe16..46160204e3a 100644 --- a/tests/unit/integrations/test_langfuse_otel.py +++ b/tests/unit/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/unit/integrations/test_opentelemetry.py b/tests/unit/integrations/test_opentelemetry.py index 175bd95c263..00818a8b1d5 100644 --- a/tests/unit/integrations/test_opentelemetry.py +++ b/tests/unit/integrations/test_opentelemetry.py @@ -19,14 +19,16 @@ import pytest # Adds the grandparent directory to sys.path to allow importing project modules from opentelemetry import trace +from opentelemetry._logs import LogRecord from opentelemetry._logs.severity import SeverityNumber -from opentelemetry.sdk._logs import LogData, LogRecord +from opentelemetry.sdk._logs import ReadableLogRecord from opentelemetry.sdk._logs import LoggerProvider as OTLoggerProvider -from opentelemetry.sdk._logs.export import InMemoryLogExporter, LogExportResult, SimpleLogRecordProcessor +from opentelemetry.sdk._logs.export import InMemoryLogExporter, LogRecordExportResult, SimpleLogRecordProcessor from opentelemetry.sdk.metrics import MeterProvider from opentelemetry.sdk.metrics.export import InMemoryMetricReader, MetricsData +from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import ReadableSpan, TracerProvider -from opentelemetry.sdk.trace.export import SimpleSpanProcessor, SpanExportResult +from opentelemetry.sdk.trace.export import BatchSpanProcessor, SimpleSpanProcessor, SpanExportResult from opentelemetry.sdk.util.instrumentation import InstrumentationScope from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter from parameterized import parameterized @@ -2147,9 +2149,13 @@ def test_otlp_http_log_export_trusts_ssl_cert_file(monkeypatch: pytest.MonkeyPat severity_number=SeverityNumber.INFO, body="tls-export-test", ) - log_data: Final = LogData(log_record=record, instrumentation_scope=InstrumentationScope("tls-export-test")) + log_data: Final = ReadableLogRecord( + log_record=record, + resource=Resource({}), + instrumentation_scope=InstrumentationScope("tls-export-test"), + ) result: Final = exporter.export([log_data]) - assert result is LogExportResult.SUCCESS, f"log export failed: {result}" + assert result is LogRecordExportResult.SUCCESS, f"log export failed: {result}" assert tls_sink.received.get(timeout=5) == "/v1/logs" finally: exporter.shutdown() @@ -6288,7 +6294,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)) @@ -6362,7 +6368,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( @@ -6379,7 +6390,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.""" @@ -6400,8 +6411,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) diff --git a/tests/unit/litellm_core_utils/test_litellm_logging.py b/tests/unit/litellm_core_utils/test_litellm_logging.py index 595b1c46188..4c3f7c87005 100644 --- a/tests/unit/litellm_core_utils/test_litellm_logging.py +++ b/tests/unit/litellm_core_utils/test_litellm_logging.py @@ -1706,7 +1706,9 @@ def test_langtrace_callback_exports_to_api_trace_with_x_api_key( ) -> None: """The exporter must post to Langtrace's complete /api/trace path with the key in x-api-key, without leaking it into the process-wide OTEL_EXPORTER_OTLP_TRACES_HEADERS.""" + import requests_mock from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter + from opentelemetry.sdk.trace.export import SpanExportResult from litellm.integrations.opentelemetry import OpenTelemetry from litellm.litellm_core_utils import litellm_logging as logging_module @@ -1728,8 +1730,12 @@ def test_langtrace_callback_exports_to_api_trace_with_x_api_key( assert type(logger) is OpenTelemetry and logger.callback_name == "langtrace" exporter: Final = logger._get_span_processor().span_exporter assert isinstance(exporter, OTLPSpanExporter) - assert exporter._endpoint == expected_endpoint - assert exporter._headers == {"x-api-key": api_key} + with requests_mock.Mocker() as collector: + collector.post(expected_endpoint, status_code=200) + assert exporter.export(()) is SpanExportResult.SUCCESS + assert [(request.url, request.headers.get("x-api-key")) for request in collector.request_history] == [ + (expected_endpoint, api_key) + ] assert "OTEL_EXPORTER_OTLP_TRACES_HEADERS" not in os.environ finally: logging_module._in_memory_loggers.clear()