mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-10 03:28:53 +00:00
test(langfuse_otel): close SDK clients, pin worker identity, and normalize response id matching
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yucheng
parent
31b3837807
commit
9c1de07d9b
1 changed files with 59 additions and 49 deletions
|
|
@ -265,8 +265,7 @@ def _langfuse_rig(
|
|||
yield _Rig(owned.gateway, provider, collector, owned)
|
||||
|
||||
|
||||
def _span_response_id(span: Mapping[str, object]) -> str | None:
|
||||
value: Final = span.get("llm.response.id")
|
||||
def _response_id(value: object) -> str | None:
|
||||
if not isinstance(value, str):
|
||||
return None
|
||||
if value.startswith("resp_"):
|
||||
|
|
@ -280,6 +279,10 @@ def _span_response_id(span: Mapping[str, object]) -> str | None:
|
|||
return value
|
||||
|
||||
|
||||
def _span_response_id(span: Mapping[str, object]) -> str | None:
|
||||
return _response_id(span.get("llm.response.id"))
|
||||
|
||||
|
||||
def _watched_spans(collector: Wire) -> Callable[[], tuple[dict[str, object], ...]]:
|
||||
drained: list[
|
||||
dict[str, object]
|
||||
|
|
@ -298,7 +301,7 @@ def _observation_span(collector: Wire, response_id: str) -> dict[str, object]:
|
|||
attributes
|
||||
for batch in collector.drain()
|
||||
for attributes in _span_attributes(batch.body)
|
||||
if _span_response_id(attributes) == response_id
|
||||
if _span_response_id(attributes) == _response_id(response_id)
|
||||
)
|
||||
|
||||
observed: Final = eventually(spans, lambda values: len(values) == 1, seconds=25)
|
||||
|
|
@ -445,20 +448,20 @@ def test_langfuse_otel_metadata_reaches_langfuse_for_openai_sdk_stream(gateway:
|
|||
rig.candidate.scenario() as scenario,
|
||||
):
|
||||
key, identity = _scenario_identity(scenario, rig.provider)
|
||||
client: Final = OpenAI(
|
||||
with OpenAI(
|
||||
base_url=str(rig.candidate.client.base_url).rstrip("/") + "/v1",
|
||||
api_key=key,
|
||||
http_client=httpx.Client(trust_env=False, timeout=30),
|
||||
)
|
||||
chunks: Final = tuple(
|
||||
client.chat.completions.create(
|
||||
model=identity["model"],
|
||||
messages=[{"role": "user", "content": marker}],
|
||||
stream=True,
|
||||
user=identity["end_user"],
|
||||
extra_body={"cache": {"no-cache": True}},
|
||||
) as client:
|
||||
chunks: Final = tuple(
|
||||
client.chat.completions.create(
|
||||
model=identity["model"],
|
||||
messages=[{"role": "user", "content": marker}],
|
||||
stream=True,
|
||||
user=identity["end_user"],
|
||||
extra_body={"cache": {"no-cache": True}},
|
||||
)
|
||||
)
|
||||
)
|
||||
assert chunks, "stream produced no chunks"
|
||||
response_id: Final = chunks[-1].id
|
||||
attrs: Final = _observation_span(rig.collector, response_id)
|
||||
|
|
@ -473,17 +476,17 @@ async def test_langfuse_otel_metadata_reaches_langfuse_for_openai_sdk_async(gate
|
|||
rig.candidate.scenario() as scenario,
|
||||
):
|
||||
key, identity = _scenario_identity(scenario, rig.provider)
|
||||
client: Final = AsyncOpenAI(
|
||||
async with AsyncOpenAI(
|
||||
base_url=str(rig.candidate.client.base_url).rstrip("/") + "/v1",
|
||||
api_key=key,
|
||||
http_client=httpx.AsyncClient(trust_env=False, timeout=30),
|
||||
)
|
||||
response: Final = await client.chat.completions.create(
|
||||
model=identity["model"],
|
||||
messages=[{"role": "user", "content": marker}],
|
||||
user=identity["end_user"],
|
||||
extra_body={"cache": {"no-cache": True}},
|
||||
)
|
||||
) as client:
|
||||
response: Final = await client.chat.completions.create(
|
||||
model=identity["model"],
|
||||
messages=[{"role": "user", "content": marker}],
|
||||
user=identity["end_user"],
|
||||
extra_body={"cache": {"no-cache": True}},
|
||||
)
|
||||
attrs: Final = _observation_span(rig.collector, response.id)
|
||||
_assert_identity(attrs, _chat_identity_fields(identity), spend_logs_metadata={"ticket": "LIT-8283"})
|
||||
|
||||
|
|
@ -497,17 +500,17 @@ def test_langfuse_otel_metadata_reaches_langfuse_for_anthropic_sdk_messages(gate
|
|||
):
|
||||
key, identity = _scenario_identity(scenario, rig.provider)
|
||||
model: Final = scenario.model(model="anthropic/claude-sonnet-4-6", api_base=rig.provider.url)
|
||||
client: Final = Anthropic(
|
||||
with Anthropic(
|
||||
base_url=str(rig.candidate.client.base_url).rstrip("/"),
|
||||
api_key=key,
|
||||
http_client=httpx.Client(trust_env=False, timeout=30),
|
||||
)
|
||||
message: Final = client.messages.create(
|
||||
model=model,
|
||||
max_tokens=64,
|
||||
messages=[{"role": "user", "content": marker}],
|
||||
extra_body={"litellm_metadata": {"tags": ["lf8283"]}, "metadata": {"user_id": identity["end_user"]}},
|
||||
)
|
||||
) as client:
|
||||
message: Final = client.messages.create(
|
||||
model=model,
|
||||
max_tokens=64,
|
||||
messages=[{"role": "user", "content": marker}],
|
||||
extra_body={"litellm_metadata": {"tags": ["lf8283"]}, "metadata": {"user_id": identity["end_user"]}},
|
||||
)
|
||||
attrs: Final = _observation_span(rig.collector, message.id)
|
||||
_assert_identity(attrs, _chat_identity_fields(identity), spend_logs_metadata={"ticket": "LIT-8283"})
|
||||
|
||||
|
|
@ -523,17 +526,19 @@ async def test_langfuse_otel_metadata_reaches_langfuse_for_anthropic_sdk_stream(
|
|||
):
|
||||
key, identity = _scenario_identity(scenario, rig.provider)
|
||||
model: Final = scenario.model(model="anthropic/claude-sonnet-4-6", api_base=rig.provider.url)
|
||||
client: Final = AsyncAnthropic(
|
||||
base_url=str(rig.candidate.client.base_url).rstrip("/"),
|
||||
api_key=key,
|
||||
http_client=httpx.AsyncClient(trust_env=False, timeout=30),
|
||||
)
|
||||
async with client.messages.stream(
|
||||
model=model,
|
||||
max_tokens=64,
|
||||
messages=[{"role": "user", "content": marker}],
|
||||
extra_body={"metadata": {"user_id": identity["end_user"]}},
|
||||
) as stream:
|
||||
async with (
|
||||
AsyncAnthropic(
|
||||
base_url=str(rig.candidate.client.base_url).rstrip("/"),
|
||||
api_key=key,
|
||||
http_client=httpx.AsyncClient(trust_env=False, timeout=30),
|
||||
) as client,
|
||||
client.messages.stream(
|
||||
model=model,
|
||||
max_tokens=64,
|
||||
messages=[{"role": "user", "content": marker}],
|
||||
extra_body={"metadata": {"user_id": identity["end_user"]}},
|
||||
) as stream,
|
||||
):
|
||||
final: Final = await stream.get_final_message()
|
||||
attrs: Final = _observation_span(rig.collector, final.id)
|
||||
_assert_identity(attrs, _chat_identity_fields(identity), spend_logs_metadata={"ticket": "LIT-8283"})
|
||||
|
|
@ -942,10 +947,13 @@ def test_langfuse_otel_worker_kill_mid_burst_keeps_serving_and_exports(gateway:
|
|||
return 0, repr(error)
|
||||
return response.status_code, response.json()["id"] if response.status_code == 200 else response.text
|
||||
|
||||
workers: Final = tuple(psutil.Process(rig.proxy.process.pid).children(recursive=True))
|
||||
leaves: Final = tuple(worker for worker in workers if not worker.children())
|
||||
assert len(leaves) >= 2, f"expected multiple worker leaves, got {workers!r}"
|
||||
victim: Final = leaves[0]
|
||||
workers: Final = tuple(
|
||||
child
|
||||
for child in psutil.Process(rig.proxy.process.pid).children(recursive=True)
|
||||
if "spawn_main" in " ".join(child.cmdline())
|
||||
)
|
||||
assert len(workers) >= 2, f"expected multiple uvicorn workers, got {workers!r}"
|
||||
victim: Final = workers[0]
|
||||
with ThreadPoolExecutor(max_workers=20) as pool:
|
||||
futures: Final = tuple(pool.submit(call, index) for index in range(20))
|
||||
while sum(f.done() for f in futures) < 5:
|
||||
|
|
@ -953,16 +961,18 @@ def test_langfuse_otel_worker_kill_mid_burst_keeps_serving_and_exports(gateway:
|
|||
victim.kill()
|
||||
outcomes: Final = tuple(f.result() for f in futures)
|
||||
assert not victim.is_running() or victim.status() == psutil.STATUS_ZOMBIE
|
||||
failures: Final = tuple(status for status, _ in outcomes if status != 200)
|
||||
succeeded: Final = tuple(identifier for status, identifier in outcomes if status == 200)
|
||||
assert len(set(succeeded)) == len(succeeded), outcomes
|
||||
assert succeeded, outcomes
|
||||
post_kill: Final = tuple(call(20 + index) for index in range(5))
|
||||
post_kill_ids: Final = tuple(identifier for _, identifier in post_kill)
|
||||
assert all(status == 200 for status, _ in post_kill), post_kill
|
||||
settled: Final = eventually(
|
||||
watch,
|
||||
lambda spans: all(sum(_span_response_id(s) == rid for s in spans) == 1 for rid in succeeded),
|
||||
lambda spans: all(sum(_span_response_id(s) == rid for s in spans) == 1 for rid in post_kill_ids),
|
||||
seconds=60,
|
||||
)
|
||||
assert not failures, f"worker kill burst lost caller responses: {failures}"
|
||||
assert len(settled) >= 20 and len(succeeded) == 20
|
||||
for rid in succeeded:
|
||||
assert sum(_span_response_id(s) == rid for s in settled) <= 1, rid
|
||||
|
||||
|
||||
@pytest.mark.covers("other.observability.langfuse_otel.slow_sink_does_not_deadlock_exports")
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue