From d5f676a8765ba857cb13cac20e202dde8895db29 Mon Sep 17 00:00:00 2001 From: Ishaan Date: Fri, 23 Oct 2026 14:05:46 -0700 Subject: [PATCH] fix(ci): unblock dependency checks (#48055) * fix(deps): upgrade flagged Python dependencies * fix(deps): lock patched Python dependency versions * fix(otel): preserve JSON export with the updated SDK * test(otel): cover JSON export with the updated SDK * test(otel): update SDK log records and worker discovery * fix(ui): align Lens peer dependency with patched Next.js * fix(otel): hand the AgentOps exporter a session it owns for the JWT header OpenTelemetry 1.45 moved the OTLP HTTP exporter's requests.Session behind a transport, so writing the minted JWT to the exporter's private _session raised, was swallowed, and every AgentOps batch went out unauthenticated. The exporter now creates the session, passes it through the public session= argument and sets the Authorization header on it The tests now drive a real export, serving the token endpoint with respx and capturing the OTLP POST with requests_mock, instead of reading SDK internals * fix(otel): skip FastAPI instrumentation once the app has started opentelemetry-instrumentation-fastapi 0.66b1 wraps build_middleware_stack instead of calling add_middleware, so instrumenting an app whose middleware stack is already built no longer raises. The call silently did nothing while we still registered the global response propagator. Check the built stack up front and skip both, which is what the existing test expects * test(otel): assert exported requests and span flags instead of SDK internals OpenTelemetry 1.45 dropped the exporter's private _headers and sets the W3C random-trace-id flag on new traces, so traceresponse ends in -03. The Langtrace and Langfuse tests now export through requests_mock and check the URL and headers the collector receives, and the traceresponse tests expect the server span's own trace flags * fix(deps): bump dashboard zod to 4.6.6 and make the vendored Lens use it zod 4.6.5 is flagged by the dependency scan and blocks osv-scan on every PR. A root bump alone leaves the vendored Lens tarball, which pins zod 4.6.5 exactly, on its own nested copy, so a top-level zod override points every consumer at the root version and the lock carries a single zod 4.6.6 * test(ui): stub the tag summary fetch the client panel retry refreshes Retrying the model breakdown refetches /tag/summary, which this integration test left on the real network. When that request failed before the assertion, the panel switched to its summary error and unmounted the model row, so the test passed or failed by timing * fix(deps): drop the duplicated Lens override from the dashboard package.json * style(ui): format the client panel fetch stub --------- Dependency files (pyproject.toml, uv.lock, ui/litellm-dashboard/package.json and package-lock.json) were left out of this pick and are re-derived on the line in the following commits ClientUsagePanel.integration.test.tsx does not exist on this line, so its stub hunk was dropped The traceresponse assertions in test_otel_v2_mount.py target tests this line does not have, so those two hunks were dropped and the file is unchanged Co-authored-by: Yuneng Jiang (cherry picked from commit 8f9894fbd6251d3efb8aebc8111270fc452270a7) --- litellm/integrations/otel/mount.py | 7 +- .../integrations/otel/plumbing/otlp_json.py | 52 +++++++- litellm/integrations/otel/presets/agentops.py | 9 +- .../otel/test_otel_v2_components.py | 113 ++++++++++++++++-- .../integrations/otel/test_otel_v2_presets.py | 76 ++++++------ tests/unit/integrations/test_langfuse_otel.py | 38 ++++-- tests/unit/integrations/test_opentelemetry.py | 29 +++-- .../test_litellm_logging.py | 10 +- 8 files changed, 255 insertions(+), 79 deletions(-) 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()