mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-01 07:03:02 +00:00
fix(ci): unblock dependency checks (#48055)
* fix(deps): upgrade 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 1.103 line in the following commits
ClientUsagePanel.integration.test.tsx does not exist on this line, so its hunk was dropped
The traceresponse hunks in test_otel_v2_mount.py target tests this line does not have, so that file is unchanged
The Langtrace export test the squash edits in test_litellm_logging.py is not on this line, so that file is unchanged
OTLPJsonSpanExporter on this line never took certificate_file or session, so the fix keeps that signature and adds only the header merge, max_request_size and the export override
The new JSON exporter tests intercept the collector with requests_mock.Mocker instead of injecting a session, and the components test imports Final, requests and requests_mock for them
The TLS export tests and their log record import hunks are not on this line, so those hunks were dropped
test_opentelemetry.py imports Final for the worker thread lookup and keeps its trace import without SpanLimits, LogRecord, Resource or the TLS helpers
Co-authored-by: Yuneng Jiang <yuneng@berri.ai>
(cherry picked from commit 8f9894fbd6)
This commit is contained in:
parent
3ba758322d
commit
fe7485c837
7 changed files with 224 additions and 69 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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):
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue