mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-26 01:12:21 +00:00
refactor(langfuse): export OTLP spans and fetch prompts through litellm's HTTPHandler instead of a private requests session
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yucheng
parent
879d8b27f0
commit
50d4bc1852
3 changed files with 172 additions and 222 deletions
|
|
@ -9,6 +9,7 @@ from typing import TYPE_CHECKING, Any, Final, Literal, TypeAlias, cast
|
|||
from litellm.integrations.custom_logger import CustomLogger
|
||||
from litellm.integrations.prompt_management_base import PromptManagementClient
|
||||
from litellm.litellm_core_utils.asyncify import run_async_function
|
||||
from litellm.llms.custom_httpx.http_handler import HTTPHandler
|
||||
from litellm.types.integrations.langfuse import LangfuseLoggedEvent
|
||||
from litellm.types.llms.openai import AllMessageValues, ChatCompletionSystemMessage
|
||||
from litellm.types.prompts.init_prompts import PromptSpec
|
||||
|
|
@ -92,20 +93,7 @@ def langfuse_client_init(
|
|||
raise_if_unsupported_langfuse_version(installed_langfuse_version())
|
||||
warn_if_upstream_langfuse_configured()
|
||||
|
||||
import httpx
|
||||
|
||||
import litellm
|
||||
|
||||
from ...llms.custom_httpx.http_handler import get_ssl_configuration
|
||||
|
||||
httpx_client: Final = (
|
||||
create_mock_langfuse_client()
|
||||
if should_use_langfuse_mock()
|
||||
else httpx.Client(
|
||||
verify=get_ssl_configuration(),
|
||||
cert=os.getenv("SSL_CERTIFICATE", litellm.ssl_certificate),
|
||||
)
|
||||
)
|
||||
httpx_client: Final = create_mock_langfuse_client() if should_use_langfuse_mock() else HTTPHandler().client
|
||||
return build_langfuse_client(
|
||||
public_key=public_key,
|
||||
secret_key=secret_key,
|
||||
|
|
|
|||
|
|
@ -21,6 +21,7 @@ from langfuse import LangfuseOtelSpanAttributes
|
|||
from langfuse.api import LangfuseAPI, Prompt, Prompt_Chat
|
||||
from langfuse.model import BasePromptClient, ChatPromptClient, PromptClient, TextPromptClient
|
||||
from opentelemetry.context import Context
|
||||
from opentelemetry.exporter.otlp.proto.common.trace_encoder import encode_spans
|
||||
from opentelemetry.sdk.resources import Resource
|
||||
from opentelemetry.sdk.trace import ReadableSpan, SpanLimits, TracerProvider
|
||||
from opentelemetry.sdk.trace.export import BatchSpanProcessor, SpanExporter, SpanExportResult
|
||||
|
|
@ -28,18 +29,18 @@ from opentelemetry.sdk.trace.id_generator import RandomIdGenerator
|
|||
from opentelemetry.sdk.trace.sampling import ALWAYS_ON, Decision, Sampler, SamplingResult
|
||||
from opentelemetry.trace import Link, NonRecordingSpan, Span, SpanContext, SpanKind, TraceFlags, Tracer, TraceState
|
||||
from opentelemetry.util.types import Attributes, AttributeValue
|
||||
from requests import PreparedRequest, RequestException, Response, Session
|
||||
from requests.adapters import HTTPAdapter
|
||||
|
||||
import litellm
|
||||
from litellm._logging import verbose_logger
|
||||
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
|
||||
from litellm.llms.custom_httpx.http_handler import HTTPHandler, _get_httpx_client
|
||||
|
||||
__all__ = (
|
||||
"DiscardingSpanExporter",
|
||||
"LangfuseApiClient",
|
||||
"LangfuseObservation",
|
||||
"LangfuseSpanExporter",
|
||||
"LangfuseTracing",
|
||||
"RetryingSpanExporter",
|
||||
"TraceIdHashSampler",
|
||||
"acquire_langfuse_tracing",
|
||||
"build_langfuse_client",
|
||||
|
|
@ -461,90 +462,78 @@ class DiscardingSpanExporter(SpanExporter):
|
|||
return True
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class RetryingSpanExporter(SpanExporter):
|
||||
"""Retry a batch whose HTTP round trip raised, as the v2 ingestion consumer did.
|
||||
_RETRYABLE_EXPORT_STATUSES: Final = frozenset({408, 429, 500, 502, 503, 504})
|
||||
_ExportOutcome = Literal["delivered", "retry", "rejected"]
|
||||
|
||||
The OTLP http exporter only retries 429 and 5xx responses; a connect or
|
||||
read timeout propagates, and ``BatchSpanProcessor`` drops the whole batch
|
||||
on any exception. A destination that stalls for a few seconds therefore
|
||||
lost every observation in flight, where v2 backed off three times first.
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class LangfuseSpanExporter(SpanExporter):
|
||||
"""OTLP/HTTP protobuf export through litellm's own HTTP handler.
|
||||
|
||||
The handler carries litellm's TLS material (``ssl_verify``, CA bundle, client certificate) exactly
|
||||
as v2's injected httpx client did. A connect or read failure and a retryable status are re-sent after
|
||||
each delay, matching the v2 ingestion consumer; ``BatchSpanProcessor`` would otherwise drop the whole
|
||||
batch on the first exception.
|
||||
"""
|
||||
|
||||
exporter: SpanExporter
|
||||
handler: HTTPHandler
|
||||
endpoint: str
|
||||
headers: Mapping[str, str]
|
||||
timeout: float
|
||||
delays: Sequence[float] = (1.0, 2.0, 4.0)
|
||||
|
||||
def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
|
||||
body: Final = encode_spans(spans).SerializeToString()
|
||||
for delay in self.delays:
|
||||
try:
|
||||
return self.exporter.export(spans)
|
||||
except RequestException as error:
|
||||
verbose_logger.warning("Langfuse export raised %s, retrying in %ss", error, delay)
|
||||
sleep(delay)
|
||||
outcome: _ExportOutcome = self._post(body)
|
||||
if outcome != "retry":
|
||||
return SpanExportResult.SUCCESS if outcome == "delivered" else SpanExportResult.FAILURE
|
||||
verbose_logger.warning("Langfuse export to %s failed, retrying in %ss", self.endpoint, delay)
|
||||
sleep(delay)
|
||||
last: Final = self._post(body)
|
||||
if last == "retry":
|
||||
verbose_logger.error("Langfuse export to %s failed after %d retries", self.endpoint, len(self.delays))
|
||||
return SpanExportResult.SUCCESS if last == "delivered" else SpanExportResult.FAILURE
|
||||
|
||||
def _post(self, body: bytes) -> _ExportOutcome:
|
||||
try:
|
||||
return self.exporter.export(spans)
|
||||
except RequestException as error:
|
||||
verbose_logger.error("Langfuse export failed after %d retries: %s", len(self.delays), error)
|
||||
return SpanExportResult.FAILURE
|
||||
self.handler.post(self.endpoint, data=body, headers=dict(self.headers), timeout=self.timeout)
|
||||
except httpx.HTTPStatusError as error:
|
||||
status: Final = error.response.status_code
|
||||
if status in _RETRYABLE_EXPORT_STATUSES:
|
||||
return "retry"
|
||||
verbose_logger.error("Langfuse rejected an export to %s with HTTP %d", self.endpoint, status)
|
||||
return "rejected"
|
||||
except (httpx.TransportError, litellm.Timeout) as error:
|
||||
verbose_logger.warning("Langfuse export to %s raised %s", self.endpoint, error)
|
||||
return "retry"
|
||||
return "delivered"
|
||||
|
||||
def shutdown(self) -> None:
|
||||
self.exporter.shutdown()
|
||||
return None
|
||||
|
||||
def force_flush(self, timeout_millis: int = 30_000) -> bool:
|
||||
return self.exporter.force_flush(timeout_millis)
|
||||
return True
|
||||
|
||||
|
||||
class _UnverifiedTlsAdapter(HTTPAdapter):
|
||||
"""Honour ``ssl_verify=False``: the exporter passes ``verify`` per request, which outranks ``Session.verify``."""
|
||||
|
||||
def send( # pyright: ignore[reportIncompatibleMethodOverride] # the stub types verify as bool | str, the base accepts both
|
||||
self,
|
||||
request: PreparedRequest,
|
||||
stream: bool = False,
|
||||
timeout: float | tuple[float, float] | tuple[float, None] | None = None,
|
||||
verify: bool | str = True,
|
||||
cert: str | tuple[str, str] | None = None,
|
||||
proxies: Mapping[str, str] | None = None,
|
||||
) -> Response:
|
||||
return super().send(request, stream=stream, timeout=timeout, verify=False, cert=cert, proxies=proxies)
|
||||
|
||||
|
||||
def _build_span_exporter(*, public_key: str, secret_key: str, base_url: str) -> RetryingSpanExporter:
|
||||
"""Build the OTLP export channel with litellm's TLS material and v2's retry behaviour.
|
||||
|
||||
v2 ingested through the injected httpx client, which carried litellm's CA
|
||||
bundle and client certificate. Endpoint, headers and timeout mirror the SDK's
|
||||
own span processor so the server treats the spans as v4 SDK traffic.
|
||||
"""
|
||||
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
|
||||
|
||||
import litellm
|
||||
from litellm.llms.custom_httpx.http_handler import get_ssl_verify
|
||||
|
||||
ssl_verify: Final = get_ssl_verify()
|
||||
ca_bundle: Final = ssl_verify if isinstance(ssl_verify, str) and os.path.exists(ssl_verify) else None
|
||||
configured_certificate: Final = os.getenv("SSL_CERTIFICATE") or litellm.ssl_certificate
|
||||
client_certificate: Final = configured_certificate if isinstance(configured_certificate, str) else None
|
||||
def _build_span_exporter(*, public_key: str, secret_key: str, base_url: str) -> LangfuseSpanExporter:
|
||||
"""Endpoint, headers and timeout mirror the SDK's span processor so the server treats the spans as v4 SDK traffic."""
|
||||
export_path: Final = os.getenv("LANGFUSE_OTEL_TRACES_EXPORT_PATH") or "/api/public/otel/v1/traces"
|
||||
endpoint: Final = f"{base_url.rstrip('/')}/{export_path.lstrip('/')}"
|
||||
encoded_auth: Final = b64encode(f"{public_key}:{secret_key}".encode()).decode("ascii")
|
||||
session: Final = Session()
|
||||
if ssl_verify is False:
|
||||
session.mount("https://", _UnverifiedTlsAdapter())
|
||||
exporter: Final = OTLPSpanExporter(
|
||||
session=session,
|
||||
endpoint=endpoint,
|
||||
headers={ # mutable-ok: the exporter copies these into its session headers
|
||||
"Authorization": "Basic " + encoded_auth,
|
||||
"x-langfuse-sdk-name": "python",
|
||||
"x-langfuse-sdk-version": version("langfuse"),
|
||||
"x-langfuse-public-key": public_key,
|
||||
},
|
||||
timeout=int(os.getenv("LANGFUSE_TIMEOUT", "5")),
|
||||
certificate_file=ca_bundle,
|
||||
client_certificate_file=client_certificate,
|
||||
return LangfuseSpanExporter(
|
||||
handler=_get_httpx_client(),
|
||||
endpoint=f"{base_url.rstrip('/')}/{export_path.lstrip('/')}",
|
||||
headers=MappingProxyType(
|
||||
{
|
||||
"Authorization": "Basic " + encoded_auth,
|
||||
"Content-Type": "application/x-protobuf",
|
||||
"x-langfuse-sdk-name": "python",
|
||||
"x-langfuse-sdk-version": version("langfuse"),
|
||||
"x-langfuse-public-key": public_key,
|
||||
}
|
||||
),
|
||||
timeout=float(os.getenv("LANGFUSE_TIMEOUT", "5")),
|
||||
)
|
||||
return RetryingSpanExporter(exporter)
|
||||
|
||||
|
||||
def _resource(*, environment: str | None, release: str | None) -> Resource:
|
||||
|
|
|
|||
|
|
@ -29,8 +29,8 @@ from litellm.integrations.langfuse.langfuse import (
|
|||
)
|
||||
from litellm.integrations.langfuse.langfuse_sdk import (
|
||||
DiscardingSpanExporter,
|
||||
LangfuseSpanExporter,
|
||||
LangfuseTracing,
|
||||
RetryingSpanExporter,
|
||||
_build_span_exporter,
|
||||
acquire_langfuse_tracing,
|
||||
build_langfuse_client,
|
||||
|
|
@ -843,28 +843,117 @@ def test_rest_client_does_not_take_over_the_process_tracer_provider():
|
|||
assert otel_trace.get_tracer_provider() is provider_before
|
||||
|
||||
|
||||
def test_ssl_exporter_carries_litellm_tls_material(monkeypatch, tmp_path):
|
||||
"""The channel is litellm's own OTLP client, so litellm's CA bundle must be rebuilt onto it."""
|
||||
import litellm
|
||||
def _finished_span():
|
||||
provider = TracerProvider()
|
||||
span = provider.get_tracer("t").start_span("generation")
|
||||
span.end()
|
||||
return span
|
||||
|
||||
for name in ("SSL_CERTIFICATE", "SSL_VERIFY", "SSL_CERT_FILE", "LANGFUSE_TIMEOUT"):
|
||||
monkeypatch.delenv(name, raising=False)
|
||||
monkeypatch.setattr(litellm, "ssl_verify", True)
|
||||
monkeypatch.setattr(litellm, "ssl_certificate", None)
|
||||
default = _build_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example").exporter
|
||||
assert default._certificate_file is True
|
||||
assert default._timeout == 5
|
||||
|
||||
ca_path = tmp_path / "private-ca.pem"
|
||||
ca_path.write_text("dummy")
|
||||
monkeypatch.setattr(litellm, "ssl_verify", str(ca_path))
|
||||
def _exporter_over(responses, *, delays=(0.5, 1.5), timeout=5.0):
|
||||
"""A LangfuseSpanExporter whose litellm HTTPHandler talks to a scripted transport instead of the network."""
|
||||
from litellm.llms.custom_httpx.http_handler import HTTPHandler
|
||||
|
||||
seen = []
|
||||
script = list(responses)
|
||||
|
||||
def transport(request: httpx.Request) -> httpx.Response:
|
||||
seen.append(request)
|
||||
step = script.pop(0)
|
||||
if isinstance(step, Exception):
|
||||
raise step
|
||||
return httpx.Response(step, request=request)
|
||||
|
||||
handler = HTTPHandler(client=httpx.Client(transport=httpx.MockTransport(transport)))
|
||||
exporter = LangfuseSpanExporter(
|
||||
handler=handler,
|
||||
endpoint="https://lf.internal.example/api/public/otel/v1/traces",
|
||||
headers=MappingProxyType({"Authorization": "Basic cGs6c2s=", "Content-Type": "application/x-protobuf"}),
|
||||
timeout=timeout,
|
||||
delays=delays,
|
||||
)
|
||||
return exporter, seen
|
||||
|
||||
|
||||
def test_exporter_posts_the_otlp_batch_through_litellm_http_handler(monkeypatch):
|
||||
"""Traces travel through litellm's own handler, so litellm's TLS and proxy settings apply to them."""
|
||||
from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ExportTraceServiceRequest
|
||||
|
||||
slept = []
|
||||
monkeypatch.setattr("litellm.integrations.langfuse.langfuse_sdk.sleep", slept.append)
|
||||
exporter, seen = _exporter_over([200])
|
||||
span = _finished_span()
|
||||
|
||||
assert exporter.export((span,)) is SpanExportResult.SUCCESS
|
||||
|
||||
(request,) = seen
|
||||
assert request.method == "POST"
|
||||
assert str(request.url) == "https://lf.internal.example/api/public/otel/v1/traces"
|
||||
assert request.headers["Authorization"] == "Basic cGs6c2s="
|
||||
assert request.headers["Content-Type"] == "application/x-protobuf"
|
||||
decoded = ExportTraceServiceRequest()
|
||||
decoded.ParseFromString(request.content)
|
||||
exported = decoded.resource_spans[0].scope_spans[0].spans[0]
|
||||
assert exported.name == "generation"
|
||||
assert exported.span_id == span.context.span_id.to_bytes(8, "big")
|
||||
assert slept == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"failure",
|
||||
[httpx.ReadTimeout("stalled"), httpx.ConnectError("refused"), 503, 429],
|
||||
ids=["read-timeout", "connect-error", "http-503", "http-429"],
|
||||
)
|
||||
def test_exporter_retries_a_failed_round_trip_and_then_succeeds(monkeypatch, failure):
|
||||
"""A stalled or restarting destination used to drop the batch outright; v2 backed off and re-sent it."""
|
||||
slept = []
|
||||
monkeypatch.setattr("litellm.integrations.langfuse.langfuse_sdk.sleep", slept.append)
|
||||
exporter, seen = _exporter_over([failure, failure, 200], delays=(0.5, 1.5, 2.5))
|
||||
|
||||
assert exporter.export((_finished_span(),)) is SpanExportResult.SUCCESS
|
||||
assert len(seen) == 3
|
||||
assert len({request.content for request in seen}) == 1
|
||||
assert slept == [0.5, 1.5]
|
||||
|
||||
|
||||
def test_exporter_gives_up_after_the_last_delay(monkeypatch):
|
||||
slept = []
|
||||
monkeypatch.setattr("litellm.integrations.langfuse.langfuse_sdk.sleep", slept.append)
|
||||
exporter, seen = _exporter_over([httpx.ConnectError("refused")] * 3, delays=(1.0, 2.0))
|
||||
|
||||
assert exporter.export((_finished_span(),)) is SpanExportResult.FAILURE
|
||||
assert len(seen) == 3
|
||||
assert slept == [1.0, 2.0]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("status", [400, 401, 403, 404])
|
||||
def test_exporter_does_not_retry_a_rejected_batch(monkeypatch, status):
|
||||
"""Bad credentials or a bad payload will not get better on the next attempt, so retrying only delays the flush."""
|
||||
slept = []
|
||||
monkeypatch.setattr("litellm.integrations.langfuse.langfuse_sdk.sleep", slept.append)
|
||||
exporter, seen = _exporter_over([status, 200])
|
||||
|
||||
assert exporter.export((_finished_span(),)) is SpanExportResult.FAILURE
|
||||
assert len(seen) == 1
|
||||
assert slept == []
|
||||
|
||||
|
||||
def test_built_exporter_uses_the_shared_litellm_handler_and_langfuse_headers(monkeypatch):
|
||||
"""No private requests session or TLS adapter: the channel is the same handler the rest of litellm uses."""
|
||||
from litellm.llms.custom_httpx.http_handler import _get_httpx_client
|
||||
|
||||
monkeypatch.delenv("LANGFUSE_TIMEOUT", raising=False)
|
||||
default = _build_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example")
|
||||
assert default.handler is _get_httpx_client()
|
||||
assert default.timeout == 5
|
||||
|
||||
monkeypatch.setenv("LANGFUSE_TIMEOUT", "20")
|
||||
exporter = _build_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example").exporter
|
||||
assert exporter._endpoint == "https://lf.internal.example/api/public/otel/v1/traces"
|
||||
assert exporter._certificate_file == str(ca_path)
|
||||
assert exporter._timeout == 20
|
||||
assert exporter._headers["x-langfuse-public-key"] == "pk"
|
||||
assert exporter._headers["x-langfuse-sdk-version"] == installed_langfuse_version()
|
||||
exporter = _build_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example")
|
||||
assert exporter.endpoint == "https://lf.internal.example/api/public/otel/v1/traces"
|
||||
assert exporter.timeout == 20
|
||||
assert exporter.headers["Authorization"] == "Basic " + b64encode(b"pk:sk").decode()
|
||||
assert exporter.headers["x-langfuse-public-key"] == "pk"
|
||||
assert exporter.headers["x-langfuse-sdk-version"] == installed_langfuse_version()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
@ -902,122 +991,6 @@ def test_export_endpoint_never_doubles_the_slash_or_leaves_the_configured_host(
|
|||
else:
|
||||
monkeypatch.setenv("LANGFUSE_OTEL_TRACES_EXPORT_PATH", export_path)
|
||||
|
||||
exporter = _build_span_exporter(public_key="pk", secret_key="sk", base_url=base_url).exporter
|
||||
exporter = _build_span_exporter(public_key="pk", secret_key="sk", base_url=base_url)
|
||||
|
||||
assert exporter._endpoint == expected
|
||||
|
||||
|
||||
def test_retrying_exporter_retries_a_raised_export_and_then_succeeds(monkeypatch):
|
||||
"""A read timeout used to drop the batch outright; v2 backed off and re-sent it."""
|
||||
from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
|
||||
from requests import ReadTimeout
|
||||
|
||||
attempts = []
|
||||
slept = []
|
||||
|
||||
class Flaky(SpanExporter):
|
||||
def export(self, spans):
|
||||
attempts.append(spans)
|
||||
if len(attempts) < 3:
|
||||
raise ReadTimeout("destination stalled")
|
||||
return SpanExportResult.SUCCESS
|
||||
|
||||
monkeypatch.setattr("litellm.integrations.langfuse.langfuse_sdk.sleep", slept.append)
|
||||
result = RetryingSpanExporter(Flaky(), delays=(0.5, 1.5, 2.5)).export(("span",))
|
||||
|
||||
assert result is SpanExportResult.SUCCESS
|
||||
assert attempts == [("span",)] * 3
|
||||
assert slept == [0.5, 1.5]
|
||||
|
||||
|
||||
def test_retrying_exporter_gives_up_after_the_last_delay(monkeypatch):
|
||||
from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
|
||||
from requests import ConnectionError as RequestsConnectionError
|
||||
|
||||
attempts = []
|
||||
slept = []
|
||||
|
||||
class Down(SpanExporter):
|
||||
def export(self, spans):
|
||||
attempts.append(spans)
|
||||
raise RequestsConnectionError("refused")
|
||||
|
||||
monkeypatch.setattr("litellm.integrations.langfuse.langfuse_sdk.sleep", slept.append)
|
||||
result = RetryingSpanExporter(Down(), delays=(1.0, 2.0)).export(("span",))
|
||||
|
||||
assert result is SpanExportResult.FAILURE
|
||||
assert len(attempts) == 3
|
||||
assert slept == [1.0, 2.0]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("switch", ["attribute", "env"])
|
||||
def test_ssl_exporter_disables_verification_when_litellm_does(monkeypatch, switch):
|
||||
"""v2 exported through the httpx client, so ``ssl_verify=False`` reached ingestion; the OTLP channel must match."""
|
||||
from requests.adapters import HTTPAdapter
|
||||
|
||||
import litellm
|
||||
|
||||
for name in ("SSL_CERTIFICATE", "SSL_VERIFY", "SSL_CERT_FILE"):
|
||||
monkeypatch.delenv(name, raising=False)
|
||||
monkeypatch.setattr(litellm, "ssl_certificate", None)
|
||||
if switch == "attribute":
|
||||
monkeypatch.setattr(litellm, "ssl_verify", False)
|
||||
else:
|
||||
monkeypatch.setattr(litellm, "ssl_verify", True)
|
||||
monkeypatch.setenv("SSL_VERIFY", "False")
|
||||
|
||||
exporter = _build_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example").exporter
|
||||
|
||||
sent = []
|
||||
|
||||
def send(self, request, **kwargs):
|
||||
sent.append((request.url, kwargs["verify"]))
|
||||
raise ConnectionError("stop before the network")
|
||||
|
||||
monkeypatch.setattr(HTTPAdapter, "send", send)
|
||||
with pytest.raises(ConnectionError):
|
||||
exporter._export(b"payload")
|
||||
assert sent[0] == ("https://lf.internal.example/api/public/otel/v1/traces", False)
|
||||
|
||||
|
||||
def test_ssl_exporter_verifies_by_default(monkeypatch):
|
||||
from requests.adapters import HTTPAdapter
|
||||
|
||||
import litellm
|
||||
|
||||
for name in ("SSL_CERTIFICATE", "SSL_VERIFY", "SSL_CERT_FILE"):
|
||||
monkeypatch.delenv(name, raising=False)
|
||||
monkeypatch.setattr(litellm, "ssl_certificate", None)
|
||||
monkeypatch.setattr(litellm, "ssl_verify", True)
|
||||
exporter = _build_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example").exporter
|
||||
|
||||
sent = []
|
||||
|
||||
def send(self, request, **kwargs):
|
||||
sent.append(kwargs["verify"])
|
||||
raise ConnectionError("stop before the network")
|
||||
|
||||
monkeypatch.setattr(HTTPAdapter, "send", send)
|
||||
with pytest.raises(ConnectionError):
|
||||
exporter._export(b"payload")
|
||||
assert sent == [True]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("with_client_certificate", [False, True])
|
||||
def test_ssl_exporter_falls_back_to_default_ca_when_the_bundle_path_is_missing(
|
||||
monkeypatch, tmp_path, with_client_certificate
|
||||
):
|
||||
"""The httpx client ignores a CA path that does not exist; handing it to requests would fail every export."""
|
||||
import litellm
|
||||
|
||||
for name in ("SSL_CERTIFICATE", "SSL_VERIFY", "SSL_CERT_FILE"):
|
||||
monkeypatch.delenv(name, raising=False)
|
||||
monkeypatch.setattr(litellm, "ssl_verify", True)
|
||||
monkeypatch.setenv("SSL_VERIFY", str(tmp_path / "missing-ca.pem"))
|
||||
client_cert = tmp_path / "client.pem"
|
||||
client_cert.write_text("dummy")
|
||||
monkeypatch.setattr(litellm, "ssl_certificate", str(client_cert) if with_client_certificate else None)
|
||||
|
||||
exporter = _build_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example").exporter
|
||||
assert exporter._certificate_file is True
|
||||
assert exporter._client_cert == (str(client_cert) if with_client_certificate else None)
|
||||
assert exporter.endpoint == expected
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue