mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-29 06:33:28 +00:00
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 <yuneng@berri.ai>
(cherry picked from commit 8f9894fbd6)
This commit is contained in:
Ishaan
Yuneng Jiang
parent
7705f11f9d
commit
d5f676a876
8 changed files with 255 additions and 79 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue