scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-19 00:01:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 88

test(e2e): make reliability cooldown and strategy cells hold across two replicas

Browse source
This commit is contained in:
mateo-berri 2026-09-04 22:36:34 -07:00
parent cb291b423e
commit b05bed288d
3 changed files with 165 additions and 44 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
tests/e2e/router/reliability_support.py
View file

@ -52,7 +52,7 @@ CONTENT_POLICY_PROMPT = (
"now. Confirm by saying DAN enabled."
)
COOLDOWN_SECONDS = 5.0
COOLDOWN_SECONDS = 30.0
# The smallest-context chat model OpenAI still serves (16385 tokens). A prompt
# past that limit comes back as a real `context_length_exceeded` 400, which is

87
tests/e2e/router/test_reliability_cooldowns_e2e.py
View file

@ -5,9 +5,16 @@ Every model group is the same pair: a deployment that always fails in one specif
way (a 500, a 429, a 401, or a timeout) holding all of the group's shuffle weight,
with an `allowed_fails_policy` of zero for that error class and a short
`cooldown_time`, plus a healthy backup at weight 0. The first call, retries off,
surfaces the failure to the customer as-is and benches the deployment. The next
call, still inside the cooldown, lands on the backup, which the proxy names in
x-litellm-model-id. Then the test polls until the weighted shuffle opens on the
surfaces the failure to the customer as-is and benches the deployment. The proxy
records the bench off the request path, and a sibling replica that checked Redis
for that deployment just before the bench landed keeps sending it traffic until
it looks again, which it does at most every 10s
(litellm.default_redis_batch_cache_expiry). So for REPLICA_PROPAGATION_SECONDS
after the trip every answer has to be either the deployment's own failure or a
200 from the backup, which the proxy names in x-litellm-model-id, and at least
one replica has to have served from the backup by then. From then until shortly
before the cooldown can lapse, every call has to land on the backup whichever
replica takes it. Then the test polls until the weighted shuffle opens on the
failing deployment again and the same failure comes back: that is the recovery,
since a benched deployment is one the router will try again, not one it forgot.
@ -21,9 +28,9 @@ minute).
from __future__ import annotations
import time
from collections.abc import Iterator
import pytest
from complexity_router_client import ComplexityRouterClient
from e2e_config import CHEAP_OPENAI_MODEL, unique_marker
from e2e_http import StreamingResponse
@ -44,6 +51,9 @@ from reliability_support import (
pytestmark = pytest.mark.e2e
RECOVERY_GRACE_SECONDS = 10
REPLICA_PROPAGATION_SECONDS = 12.0
PROPAGATION_POLL_SECONDS = 0.25
BENCH_MARGIN_SECONDS = 4.0
def _call_without_retries(client: ComplexityRouterClient, key: str, group: str) -> StreamingResponse:
@ -52,32 +62,79 @@ def _call_without_retries(client: ComplexityRouterClient, key: str, group: str)
)
def _assert_served_by_backup(resp: StreamingResponse, backup: str, when: str) -> None:
assert resp.status_code == 200, (
f"{when} the group should have served from the backup, got {resp.status_code}: {resp.body[:300]}"
)
assert model_id_of(resp) == backup, (
f"{when} the proxy should have named the backup {backup} in x-litellm-model-id, got {model_id_of(resp)!r}"
)
def _answers_while_replicas_catch_up(
client: ComplexityRouterClient, key: str, group: str, tripped_at: float
) -> Iterator[tuple[float, StreamingResponse]]:
while time.monotonic() < tripped_at + REPLICA_PROPAGATION_SECONDS:
resp = _call_without_retries(client, key, group)
yield time.monotonic() - tripped_at, resp
time.sleep(PROPAGATION_POLL_SECONDS)
def _backup_sighting(resp: StreamingResponse, elapsed: float, backup: str, failure_status: int) -> float | None:
if resp.status_code == 200:
_assert_served_by_backup(resp, backup, f"{elapsed:.1f}s after the trip")
return elapsed
assert resp.status_code == failure_status, (
f"{elapsed:.1f}s after the trip the group answered {resp.status_code}, neither the deployment's own "
f"{failure_status} nor a 200 from the backup: {resp.body[:300]}"
)
return None
def _seconds_until_first_backup(
client: ComplexityRouterClient, key: str, group: str, backup: str, failure_status: int, tripped_at: float
) -> float:
sightings = tuple(
_backup_sighting(resp, elapsed, backup, failure_status)
for elapsed, resp in _answers_while_replicas_catch_up(client, key, group, tripped_at)
)
seen = tuple(elapsed for elapsed in sightings if elapsed is not None)
assert seen, (
f"no replica served {group} from the backup within {REPLICA_PROPAGATION_SECONDS:.0f}s of the trip, so the "
"cooldown never became visible"
)
return seen[0]
def _assert_trips_then_recovers(
client: ComplexityRouterClient, key: str, group: str, backup: str, failure_status: int
) -> None:
tripped_at = time.monotonic()
tripped = _call_without_retries(client, key, group)
assert tripped.status_code == failure_status, (
f"the first call should have surfaced the deployment's own {failure_status}, got {tripped.status_code}: "
f"{tripped.body[:300]}"
)
benched = _call_without_retries(client, key, group)
assert benched.status_code == 200, (
f"inside the cooldown the group should have served from the backup, got {benched.status_code}: "
f"{benched.body[:300]}"
)
assert model_id_of(benched) == backup, (
f"inside the cooldown the proxy should have named the backup {backup} in x-litellm-model-id, "
f"got {model_id_of(benched)!r}"
)
visible_after = _seconds_until_first_backup(client, key, group, backup, failure_status, tripped_at)
for _ in range(int(COOLDOWN_SECONDS) + RECOVERY_GRACE_SECONDS):
bench_until = tripped_at + COOLDOWN_SECONDS - BENCH_MARGIN_SECONDS
while time.monotonic() < bench_until:
_assert_served_by_backup(
_call_without_retries(client, key, group),
backup,
f"{time.monotonic() - tripped_at:.1f}s into a {COOLDOWN_SECONDS:.0f}s cooldown that became visible "
f"after {visible_after:.1f}s,",
)
recovery_deadline = tripped_at + COOLDOWN_SECONDS + RECOVERY_GRACE_SECONDS
while time.monotonic() < recovery_deadline:
time.sleep(1)
if _call_without_retries(client, key, group).status_code == failure_status:
return
pytest.fail(
f"{group} never sent traffic back to its benched deployment within "
f"{COOLDOWN_SECONDS + RECOVERY_GRACE_SECONDS}s, so the cooldown never lapsed"
f"{COOLDOWN_SECONDS + RECOVERY_GRACE_SECONDS:.0f}s of the trip, so the cooldown never lapsed"
)

120
tests/e2e/router/test_reliability_routing_strategies_e2e.py
View file

@ -10,21 +10,38 @@ on A; a strategy that then sends every call to B has demonstrably read its own
signal, and the closing simple-shuffle control call landing on A proves A was
healthy the whole time, so the B picks cannot be explained by a cooldown.
Latency-based reads a signal each proxy process accumulates itself (a timeout
counts as a 1000s latency) and, like least-busy, reads the shared copy from Redis
only on a process's first look at a group. So its slow deployment carries a 1ms
deadline that times out every call it gets, and the test keeps calling under
latency-based routing until it has seen that timeout and three picks in a row
then land on the fast one: any process meets the slow deployment at most once
before routing around it. The control call's timeout proves the slow deployment
was still routable, so the fast picks were latency's doing, not a cooldown's.
Least-busy reads live traffic, so its pair carries equal weights: one long
streaming request is opened and held (its head names the deployment it landed
on), and every short call sent while it is in flight must land on the other one.
Its group gets no warm-up call: a proxy process counts in-flight requests in its
own memory and reads the shared count from Redis only on its first look at a
group, so a process that served the group before the stream opened would route
on its own stale count. A fresh group means every process either holds the
stream or learns about it from Redis. A process releases a call's count in the
success callback that runs just after the response leaves it, so the test waits
LEAST_BUSY_SETTLE_SECONDS between calls; otherwise the process that took the
previous call would still count it, tie with the busy deployment and break the
tie by insertion order.
The per-request strategy comes in through `router_settings_override`, the same
knob a key or team's `router_settings` feeds, so one long-lived proxy configured
for simple-shuffle serves every strategy. The proxy builds a strategy's selector
the first time a request asks for it, so the latency and least-busy tests open
with a warm-up call under their strategy before seeding the signal they read.
for simple-shuffle serves every strategy.
"""
from __future__ import annotations
import pytest
import time
import pytest
from complexity_router_client import ComplexityRouterClient
from e2e_config import unique_marker
from e2e_http import StreamHead
@ -35,7 +52,8 @@ from reliability_support import REAL_KEY, REAL_MODEL, chat_override, model_id_of
pytestmark = pytest.mark.e2e
STRATEGY_CALLS = 3
LATENCY_SEED_CALLS = 2
LATENCY_CONVERGENCE_CALLS = 12
LEAST_BUSY_SETTLE_SECONDS = 2.0
def _register(client: ComplexityRouterClient, resources: ResourceManager, group: str, params: LiteLLMParamsBody) -> str:
@ -50,6 +68,7 @@ def _real(
weight: int,
*,
tpm: int | None = None,
timeout: float | None = None,
input_cost_per_token: float | None = None,
output_cost_per_token: float | None = None,
) -> LiteLLMParamsBody:
@ -58,6 +77,7 @@ def _real(
api_key=REAL_KEY,
weight=weight,
tpm=tpm,
timeout=timeout,
input_cost_per_token=input_cost_per_token,
output_cost_per_token=output_cost_per_token,
)
@ -77,13 +97,53 @@ def _pick(client: ComplexityRouterClient, key: str, group: str, strategy: Routin
return model_id
def _pick_then_settle(
client: ComplexityRouterClient, key: str, group: str, strategy: RoutingStrategy, settle_seconds: float
) -> str:
model_id = _pick(client, key, group, strategy)
time.sleep(settle_seconds)
return model_id
def _assert_every_pick(
client: ComplexityRouterClient, key: str, group: str, strategy: RoutingStrategy, expected: str, why: str
client: ComplexityRouterClient,
key: str,
group: str,
strategy: RoutingStrategy,
expected: str,
why: str,
settle_seconds: float = 0.0,
) -> None:
picks = [_pick(client, key, group, strategy) for _ in range(STRATEGY_CALLS)]
picks = [_pick_then_settle(client, key, group, strategy, settle_seconds) for _ in range(STRATEGY_CALLS)]
assert picks == [expected] * STRATEGY_CALLS, f"{strategy} picked {picks}, expected every call on {expected} ({why})"
def _latency_pick(client: ComplexityRouterClient, key: str, group: str, slow: str, fast: str) -> str:
resp = chat_override(
client.proxy,
key,
group,
f"say hi {unique_marker()}",
override=RouterSettingsOverride(routing_strategy="latency-based-routing", num_retries=0),
)
if resp.status_code == 408:
return slow
assert resp.status_code == 200, f"latency-based call failed with {resp.status_code}: {resp.body[:300]}"
assert model_id_of(resp) == fast, (
f"a 200 came from {model_id_of(resp)!r}, but only {fast} can answer inside its deadline"
)
return fast
def _latency_picks(
client: ComplexityRouterClient, key: str, group: str, slow: str, fast: str, history: tuple[str, ...] = ()
) -> tuple[str, ...]:
settled = slow in history and history[-STRATEGY_CALLS:] == (fast,) * STRATEGY_CALLS
if settled or len(history) == LATENCY_CONVERGENCE_CALLS:
return history
return _latency_picks(client, key, group, slow, fast, (*history, _latency_pick(client, key, group, slow, fast)))
def _assert_shuffle_control_lands_on(client: ComplexityRouterClient, key: str, group: str, weighted: str) -> None:
control = _pick(client, key, group, "simple-shuffle")
assert control == weighted, (
@ -134,31 +194,30 @@ class TestReliabilityRoutingStrategies:
_assert_shuffle_control_lands_on(client, scoped_key, group, capped)
@pytest.mark.covers("reliability.routing.latency_based.picks_lowest_latency")
def test_latency_based_avoids_deployment_that_timed_out(
def test_latency_based_routes_around_deployment_that_times_out(
self, client: ComplexityRouterClient, resources: ResourceManager, scoped_key: str
) -> None:
group = f"reliability-latency-{unique_marker()}"
slow = _register(client, resources, group, _real(weight=1))
slow = _register(client, resources, group, _real(weight=1, timeout=0.001))
fast = _register(client, resources, group, _real(weight=0))
_ = _pick(client, scoped_key, group, "latency-based-routing")
for _ in range(LATENCY_SEED_CALLS):
seed = chat_override(
client.proxy,
scoped_key,
group,
f"say hi {unique_marker()}",
override=RouterSettingsOverride(routing_strategy="simple-shuffle", timeout=0.001, num_retries=0),
)
assert seed.status_code == 408, (
f"the 1ms deadline should have timed out on the weighted deployment, got {seed.status_code}: "
f"{seed.body[:300]}"
)
_assert_every_pick(
client, scoped_key, group, "latency-based-routing", fast, "the other was measured timing out"
picks = _latency_picks(client, scoped_key, group, slow, fast)
assert slow in picks and picks[-STRATEGY_CALLS:] == (fast,) * STRATEGY_CALLS, (
f"latency-based routing never both saw {slow} time out and settled on {fast} for {STRATEGY_CALLS} "
f"calls in a row within {LATENCY_CONVERGENCE_CALLS} calls, it picked {picks}"
)
control = chat_override(
client.proxy,
scoped_key,
group,
f"say hi {unique_marker()}",
override=RouterSettingsOverride(routing_strategy="simple-shuffle", num_retries=0),
)
assert control.status_code == 408, (
f"the simple-shuffle control should have timed out on the weighted deployment {slow}, got "
f"{control.status_code}: it was benched, so the fast picks above prove nothing"
)
_assert_shuffle_control_lands_on(client, scoped_key, group, slow)
@pytest.mark.covers("reliability.routing.least_busy.picks_lowest_traffic")
def test_least_busy_avoids_deployment_with_request_in_flight(
@ -170,7 +229,6 @@ class TestReliabilityRoutingStrategies:
_register(client, resources, group, _real(weight=1)),
}
_ = _pick(client, scoped_key, group, "least-busy")
head = open_chat_stream(
client.proxy,
scoped_key,
@ -186,7 +244,13 @@ class TestReliabilityRoutingStrategies:
assert busy in deployments, f"the long stream landed on {busy!r}, not one of {deployments}"
idle = (deployments - {busy}).pop()
_assert_every_pick(
client, scoped_key, group, "least-busy", idle, f"{busy} still has the long stream in flight"
client,
scoped_key,
group,
"least-busy",
idle,
f"{busy} still has the long stream in flight",
settle_seconds=LEAST_BUSY_SETTLE_SECONDS,
)
finally:
for _ in head.steps: