mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-16 23:41:43 +00:00
Merge pull request #34166 from BerriAI/litellm_lit_4562_weekly_anomaly_load_test
test(e2e): add weekly session-anomaly load test against real providers
This commit is contained in:
commit
6375923f65
11 changed files with 697 additions and 7 deletions
81
.github/workflows/weekly_load_anomaly.yml
vendored
Normal file
81
.github/workflows/weekly_load_anomaly.yml
vendored
Normal file
|
|
@ -0,0 +1,81 @@
|
||||||
|
name: "Weekly Load Anomaly Check"
|
||||||
|
|
||||||
|
on:
|
||||||
|
schedule:
|
||||||
|
- cron: "0 12 * * 6"
|
||||||
|
workflow_dispatch:
|
||||||
|
|
||||||
|
permissions:
|
||||||
|
contents: read
|
||||||
|
|
||||||
|
jobs:
|
||||||
|
weekly-load-anomaly:
|
||||||
|
if: github.event_name != 'schedule' || github.repository == 'BerriAI/litellm'
|
||||||
|
runs-on: ubuntu-latest
|
||||||
|
timeout-minutes: 45
|
||||||
|
services:
|
||||||
|
postgres:
|
||||||
|
image: postgres:16.6
|
||||||
|
env:
|
||||||
|
POSTGRES_USER: llmproxy
|
||||||
|
POSTGRES_PASSWORD: dbpassword9090
|
||||||
|
POSTGRES_DB: litellm
|
||||||
|
ports:
|
||||||
|
- 5432:5432
|
||||||
|
options: >-
|
||||||
|
--health-cmd "pg_isready -U llmproxy"
|
||||||
|
--health-interval 5s
|
||||||
|
--health-timeout 5s
|
||||||
|
--health-retries 10
|
||||||
|
env:
|
||||||
|
DATABASE_URL: postgresql://llmproxy:dbpassword9090@localhost:5432/litellm
|
||||||
|
LITELLM_MASTER_KEY: sk-weekly-anomaly-check
|
||||||
|
ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
|
||||||
|
AWS_BEARER_TOKEN_BEDROCK: ${{ secrets.AWS_BEARER_TOKEN_BEDROCK }}
|
||||||
|
steps:
|
||||||
|
- uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
|
||||||
|
with:
|
||||||
|
persist-credentials: false
|
||||||
|
|
||||||
|
- name: Set up Python
|
||||||
|
uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
|
||||||
|
with:
|
||||||
|
python-version: "3.12"
|
||||||
|
|
||||||
|
- name: Set up uv
|
||||||
|
uses: ./.github/actions/setup-uv-with-retries
|
||||||
|
with:
|
||||||
|
version: "0.10.9"
|
||||||
|
|
||||||
|
- name: Install dependencies
|
||||||
|
run: |
|
||||||
|
.github/scripts/uv_sync_with_retries.sh --frozen --group ci --group proxy-dev --extra proxy
|
||||||
|
|
||||||
|
- name: Generate Prisma client
|
||||||
|
env:
|
||||||
|
PRISMA_BINARY_CACHE_DIR: ${{ runner.temp }}/prisma-cache
|
||||||
|
run: |
|
||||||
|
uv run --no-sync prisma generate --schema litellm/proxy/schema.prisma
|
||||||
|
|
||||||
|
- name: Start the proxy
|
||||||
|
run: |
|
||||||
|
nohup uv run --no-sync litellm --config tests/e2e/load/weekly_anomaly_config.yml --port 4000 > proxy.log 2>&1 &
|
||||||
|
for _ in $(seq 1 90); do
|
||||||
|
if curl -fs http://localhost:4000/health/liveliness > /dev/null; then
|
||||||
|
exit 0
|
||||||
|
fi
|
||||||
|
sleep 2
|
||||||
|
done
|
||||||
|
echo "proxy never became live"
|
||||||
|
tail -n 100 proxy.log
|
||||||
|
exit 1
|
||||||
|
|
||||||
|
- name: Run the weekly session anomaly test
|
||||||
|
env:
|
||||||
|
E2E_WEEKLY_ANOMALY: "1"
|
||||||
|
run: |
|
||||||
|
uv run --no-sync pytest tests/e2e/load/test_weekly_session_anomaly_e2e.py -v --tb=short -rA
|
||||||
|
|
||||||
|
- name: Show proxy log on failure
|
||||||
|
if: failure()
|
||||||
|
run: tail -n 300 proxy.log
|
||||||
|
|
@ -18,7 +18,7 @@ Each subdirectory under `tests/e2e/` is one suite, scoped to an endpoint family
|
||||||
- `logging/` - logging-integration delivery (datadog and friends)
|
- `logging/` - logging-integration delivery (datadog and friends)
|
||||||
- `security/` - secret handling and log-leak protection
|
- `security/` - secret handling and log-leak protection
|
||||||
- `router/` - routing and reliability behavior (fallbacks, cooldowns)
|
- `router/` - routing and reliability behavior (fallbacks, cooldowns)
|
||||||
- `load/` - throughput/performance under concurrency: drives real concurrent traffic through the whole stack with Locust and asserts a throughput SLO; marked `load` so the parent conftest collects it last and it never perturbs latency-sensitive suites
|
- `load/` - throughput/performance under concurrency: drives real concurrent traffic through the whole stack with Locust and asserts a throughput SLO; marked `load` so the parent conftest collects it last and it never perturbs latency-sensitive suites. Also home of the weekly session-anomaly test (`test_weekly_session_anomaly_e2e.py`): Claude Code-shaped multi-turn sessions against real providers with ceilings on error rate, cache read/write, turn time, and spend; additionally marked `weekly` and deselected unless `E2E_WEEKLY_ANOMALY` is set, because it spends real provider money (driven by `.github/workflows/weekly_load_anomaly.yml`)
|
||||||
- `other/` - the holding-pen suite for the `other.*` registry cluster with no home of its own yet: the master-key auth gate and the process-lifecycle health probes (liveness, public readiness, authenticated readiness diagnostics). Promote a cluster out once it is large/stable enough for its own suite
|
- `other/` - the holding-pen suite for the `other.*` registry cluster with no home of its own yet: the master-key auth gate and the process-lifecycle health probes (liveness, public readiness, authenticated readiness diagnostics). Promote a cluster out once it is large/stable enough for its own suite
|
||||||
- `gateway/` - proxy configuration only (`litellm-config.yml`); no tests
|
- `gateway/` - proxy configuration only (`litellm-config.yml`); no tests
|
||||||
- `claude_code/` - the Claude Code compatibility matrix: drives the real `claude` CLI (and HTTP probes) against a proxy for each feature x provider cell, reporting tagged-union outcomes via the `compat_result` fixture; ships its own driver/builder/publisher plus `_*_unit_tests/` trees. The HTTP probes ride the shared transport (`ProxyClient.count_tokens` / `ProxyClient.messages`); the CLI-driving path stays bespoke
|
- `claude_code/` - the Claude Code compatibility matrix: drives the real `claude` CLI (and HTTP probes) against a proxy for each feature x provider cell, reporting tagged-union outcomes via the `compat_result` fixture; ships its own driver/builder/publisher plus `_*_unit_tests/` trees. The HTTP probes ride the shared transport (`ProxyClient.count_tokens` / `ProxyClient.messages`); the CLI-driving path stays bespoke
|
||||||
|
|
@ -132,7 +132,7 @@ reliability.<behavior>.<variant>.<assertion>
|
||||||
behavior : fallback | retry | cooldown | timeout | routing | cache | circuit_breaker | perf
|
behavior : fallback | retry | cooldown | timeout | routing | cache | circuit_breaker | perf
|
||||||
variant : <trigger> 5xx | context_window | content_policy | 429 | timeout
|
variant : <trigger> 5xx | context_window | content_policy | 429 | timeout
|
||||||
<strategy> simple_shuffle | usage_based | latency_based | cost_based | least_busy
|
<strategy> simple_shuffle | usage_based | latency_based | cost_based | least_busy
|
||||||
<dimension> latency | throughput (perf only; SLO/threshold assertion, not binary)
|
<dimension> latency | throughput | session_anomaly (perf only; SLO/threshold assertion, not binary)
|
||||||
assertion : routes_to_fallback | succeeds_within_retries | picks_under_tpm | returns_cached
|
assertion : routes_to_fallback | succeeds_within_retries | picks_under_tpm | returns_cached
|
||||||
| trips_then_recovers | under_slo
|
| trips_then_recovers | under_slo
|
||||||
e.g. reliability.fallback.context_window.routes_to_fallback exercised_on=[chat_completions]
|
e.g. reliability.fallback.context_window.routes_to_fallback exercised_on=[chat_completions]
|
||||||
|
|
|
||||||
|
|
@ -43,6 +43,10 @@ def pytest_configure(config: pytest.Config) -> None:
|
||||||
"markers",
|
"markers",
|
||||||
"load: heavy throughput/load test; collected last so it never perturbs latency-sensitive suites",
|
"load: heavy throughput/load test; collected last so it never perturbs latency-sensitive suites",
|
||||||
)
|
)
|
||||||
|
config.addinivalue_line(
|
||||||
|
"markers",
|
||||||
|
"weekly: real-provider anomaly load test that spends real money; deselected unless E2E_WEEKLY_ANOMALY is set",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def pytest_collection_modifyitems(items: list[pytest.Item]) -> None:
|
def pytest_collection_modifyitems(items: list[pytest.Item]) -> None:
|
||||||
|
|
|
||||||
|
|
@ -24,3 +24,4 @@
|
||||||
- {id: reliability.timeout.request_timeout.exceeds_deadline, module: reliability, tier: P1, behavior: timeout, variant: request_timeout, assertions: [exceeds_deadline], exercised_on: [chat_completions, messages], source: "litellm/router.py:545-551", rationale: "Per-request timeout raises Timeout"}
|
- {id: reliability.timeout.request_timeout.exceeds_deadline, module: reliability, tier: P1, behavior: timeout, variant: request_timeout, assertions: [exceeds_deadline], exercised_on: [chat_completions, messages], source: "litellm/router.py:545-551", rationale: "Per-request timeout raises Timeout"}
|
||||||
- {id: reliability.timeout.stream_timeout.exceeds_deadline, module: reliability, tier: P1, behavior: timeout, variant: stream_timeout, assertions: [exceeds_deadline], exercised_on: [chat_completions], source: "litellm/router.py:551", rationale: "Streaming chunk-delivery timeout"}
|
- {id: reliability.timeout.stream_timeout.exceeds_deadline, module: reliability, tier: P1, behavior: timeout, variant: stream_timeout, assertions: [exceeds_deadline], exercised_on: [chat_completions], source: "litellm/router.py:551", rationale: "Streaming chunk-delivery timeout"}
|
||||||
- {id: reliability.perf.throughput.under_slo, module: reliability, tier: P1, behavior: perf, variant: throughput, assertions: [under_slo], exercised_on: [chat_completions, messages], source: grammar, rationale: "Throughput SLO under load"}
|
- {id: reliability.perf.throughput.under_slo, module: reliability, tier: P1, behavior: perf, variant: throughput, assertions: [under_slo], exercised_on: [chat_completions, messages], source: grammar, rationale: "Throughput SLO under load"}
|
||||||
|
- {id: reliability.perf.session_anomaly.under_slo, module: reliability, tier: P1, behavior: perf, variant: session_anomaly, assertions: [under_slo], exercised_on: [messages], source: grammar, rationale: "Weekly Claude Code-shaped multi-turn session load against real providers; ceilings on error rate, warm-turn cache read/write, p95 turn time, and gateway-recorded spend (LIT-4562)"}
|
||||||
|
|
|
||||||
|
|
@ -78,6 +78,24 @@ LOAD_DURATION_SECONDS = float(os.environ.get("E2E_LOAD_DURATION_SECONDS", "60"))
|
||||||
LOAD_MIN_RPS = float(os.environ.get("E2E_LOAD_MIN_RPS", "355"))
|
LOAD_MIN_RPS = float(os.environ.get("E2E_LOAD_MIN_RPS", "355"))
|
||||||
LOAD_MAX_FAILURE_RATIO = float(os.environ.get("E2E_LOAD_MAX_FAILURE_RATIO", "0.01"))
|
LOAD_MAX_FAILURE_RATIO = float(os.environ.get("E2E_LOAD_MAX_FAILURE_RATIO", "0.01"))
|
||||||
|
|
||||||
|
WEEKLY_ANOMALY_OPT_IN_ENV = "E2E_WEEKLY_ANOMALY"
|
||||||
|
ANOMALY_SESSIONS = int(os.environ.get("E2E_ANOMALY_SESSIONS", "6"))
|
||||||
|
ANOMALY_TURNS_PER_SESSION = int(os.environ.get("E2E_ANOMALY_TURNS_PER_SESSION", "6"))
|
||||||
|
ANOMALY_TURN_ATTEMPTS = int(os.environ.get("E2E_ANOMALY_TURN_ATTEMPTS", "3"))
|
||||||
|
ANOMALY_MAX_ERROR_RATIO = float(os.environ.get("E2E_ANOMALY_MAX_ERROR_RATIO", "0.05"))
|
||||||
|
ANOMALY_MIN_WARM_CACHE_READ_SHARE = float(
|
||||||
|
os.environ.get("E2E_ANOMALY_MIN_WARM_CACHE_READ_SHARE", "0.65")
|
||||||
|
)
|
||||||
|
ANOMALY_MAX_P95_TURN_SECONDS = float(
|
||||||
|
os.environ.get("E2E_ANOMALY_MAX_P95_TURN_SECONDS", "30")
|
||||||
|
)
|
||||||
|
ANOMALY_MAX_KEY_SPEND_USD = float(
|
||||||
|
os.environ.get("E2E_ANOMALY_MAX_KEY_SPEND_USD", "0.60")
|
||||||
|
)
|
||||||
|
ANOMALY_SPEND_SETTLE_SECONDS = float(
|
||||||
|
os.environ.get("E2E_ANOMALY_SPEND_SETTLE_SECONDS", "75")
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def require_env(*names: str) -> tuple[str, ...]:
|
def require_env(*names: str) -> tuple[str, ...]:
|
||||||
"""Return the non-empty values for each env name, or hard-fail naming which are missing.
|
"""Return the non-empty values for each env name, or hard-fail naming which are missing.
|
||||||
|
|
|
||||||
|
|
@ -1,10 +1,12 @@
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import os
|
||||||
from collections.abc import Iterator
|
from collections.abc import Iterator
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
from requests import RequestException
|
from requests import RequestException
|
||||||
|
|
||||||
|
from e2e_config import WEEKLY_ANOMALY_OPT_IN_ENV
|
||||||
from e2e_http import NoBody, Success
|
from e2e_http import NoBody, Success
|
||||||
from load_client import LoadClient, build_client
|
from load_client import LoadClient, build_client
|
||||||
from load_constants import LOAD_MODEL
|
from load_constants import LOAD_MODEL
|
||||||
|
|
@ -18,6 +20,22 @@ LOAD_MODEL_PARAMS = LiteLLMParamsBody(
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def pytest_collection_modifyitems(
|
||||||
|
config: pytest.Config, items: list[pytest.Item]
|
||||||
|
) -> None:
|
||||||
|
if os.environ.get(WEEKLY_ANOMALY_OPT_IN_ENV):
|
||||||
|
return
|
||||||
|
deselected = [
|
||||||
|
item for item in items if item.get_closest_marker("weekly") is not None
|
||||||
|
]
|
||||||
|
if not deselected:
|
||||||
|
return
|
||||||
|
config.hook.pytest_deselected(items=deselected)
|
||||||
|
items[:] = [
|
||||||
|
item for item in items if item.get_closest_marker("weekly") is None
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope="session")
|
@pytest.fixture(scope="session")
|
||||||
def client(proxy: ProxyClient) -> LoadClient:
|
def client(proxy: ProxyClient) -> LoadClient:
|
||||||
return build_client(proxy)
|
return build_client(proxy)
|
||||||
|
|
@ -33,10 +51,8 @@ def _model_is_servable(proxy: ProxyClient, model_name: str) -> bool:
|
||||||
return isinstance(result, Success) and any(entry.id == model_name for entry in result.data.data)
|
return isinstance(result, Success) and any(entry.id == model_name for entry in result.data.data)
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope="session", autouse=True)
|
@pytest.fixture(scope="session")
|
||||||
def _ensure_load_model( # pyright: ignore[reportUnusedFunction] # pytest autouse session fixture, wired by name
|
def ensure_load_model(client: LoadClient) -> Iterator[None]:
|
||||||
client: LoadClient,
|
|
||||||
) -> Iterator[None]:
|
|
||||||
proxy = client.proxy
|
proxy = client.proxy
|
||||||
if _model_is_servable(proxy, LOAD_MODEL):
|
if _model_is_servable(proxy, LOAD_MODEL):
|
||||||
yield
|
yield
|
||||||
|
|
@ -60,7 +76,9 @@ def _ensure_load_model( # pyright: ignore[reportUnusedFunction] # pytest autou
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def load_key(resources: ResourceManager, client: LoadClient) -> str:
|
def load_key(
|
||||||
|
resources: ResourceManager, client: LoadClient, ensure_load_model: None
|
||||||
|
) -> str:
|
||||||
key = client.proxy.generate_key(KeyGenerateBody(models=[LOAD_MODEL], user_id="e2e-load"))
|
key = client.proxy.generate_key(KeyGenerateBody(models=[LOAD_MODEL], user_id="e2e-load"))
|
||||||
resources.defer(lambda: client.proxy.delete_key(key))
|
resources.defer(lambda: client.proxy.delete_key(key))
|
||||||
return key
|
return key
|
||||||
|
|
|
||||||
299
tests/e2e/load/session_anomaly.py
Normal file
299
tests/e2e/load/session_anomaly.py
Normal file
|
|
@ -0,0 +1,299 @@
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import time
|
||||||
|
from collections.abc import Callable
|
||||||
|
from concurrent.futures import ThreadPoolExecutor
|
||||||
|
from dataclasses import dataclass
|
||||||
|
|
||||||
|
from pydantic import BaseModel
|
||||||
|
|
||||||
|
from e2e_config import unique_marker
|
||||||
|
from e2e_http import Result, Success
|
||||||
|
from models import CacheControl, RichMessage, TextBlock
|
||||||
|
from transport import Transport
|
||||||
|
|
||||||
|
|
||||||
|
class SessionMessagesRequest(BaseModel):
|
||||||
|
model: str
|
||||||
|
max_tokens: int = 128
|
||||||
|
system: list[TextBlock]
|
||||||
|
messages: list[RichMessage]
|
||||||
|
|
||||||
|
|
||||||
|
class SessionUsage(BaseModel):
|
||||||
|
input_tokens: int = 0
|
||||||
|
output_tokens: int = 0
|
||||||
|
cache_creation_input_tokens: int = 0
|
||||||
|
cache_read_input_tokens: int = 0
|
||||||
|
|
||||||
|
|
||||||
|
class SessionContentBlock(BaseModel):
|
||||||
|
type: str | None = None
|
||||||
|
text: str | None = None
|
||||||
|
|
||||||
|
|
||||||
|
class SessionMessagesResponse(BaseModel):
|
||||||
|
content: list[SessionContentBlock] = []
|
||||||
|
usage: SessionUsage = SessionUsage()
|
||||||
|
|
||||||
|
@property
|
||||||
|
def text(self) -> str:
|
||||||
|
return "".join(block.text or "" for block in self.content)
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True, slots=True)
|
||||||
|
class TurnMetric:
|
||||||
|
turn_index: int
|
||||||
|
ok: bool
|
||||||
|
latency_seconds: float
|
||||||
|
uncached_input_tokens: int
|
||||||
|
cache_read_tokens: int
|
||||||
|
cache_creation_tokens: int
|
||||||
|
failure: str | None
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True, slots=True)
|
||||||
|
class AnomalyReport:
|
||||||
|
planned_turns: int
|
||||||
|
attempted_turns: int
|
||||||
|
failed_turns: int
|
||||||
|
warm_turns: int
|
||||||
|
warm_uncached_input_tokens: int
|
||||||
|
warm_cache_read_tokens: int
|
||||||
|
warm_cache_creation_tokens: int
|
||||||
|
p95_turn_seconds: float
|
||||||
|
|
||||||
|
@property
|
||||||
|
def error_ratio(self) -> float:
|
||||||
|
return self.failed_turns / self.planned_turns if self.planned_turns else 1.0
|
||||||
|
|
||||||
|
@property
|
||||||
|
def warm_cache_read_share(self) -> float:
|
||||||
|
billed = (
|
||||||
|
self.warm_uncached_input_tokens
|
||||||
|
+ self.warm_cache_read_tokens
|
||||||
|
+ self.warm_cache_creation_tokens
|
||||||
|
)
|
||||||
|
return self.warm_cache_read_tokens / billed if billed else 0.0
|
||||||
|
|
||||||
|
|
||||||
|
def _system_prefix_block(marker: str) -> TextBlock:
|
||||||
|
text = " ".join(
|
||||||
|
f"Project context paragraph {index} for session {marker}." for index in range(300)
|
||||||
|
)
|
||||||
|
return TextBlock(text=text, cache_control=CacheControl())
|
||||||
|
|
||||||
|
|
||||||
|
def _user_turn_text(marker: str, turn_index: int) -> str:
|
||||||
|
notes = " ".join(
|
||||||
|
f"Working note {index} of turn {turn_index} in session {marker}."
|
||||||
|
for index in range(80)
|
||||||
|
)
|
||||||
|
return f"Reply with one short sentence.\n{notes}"
|
||||||
|
|
||||||
|
|
||||||
|
def _reminder_turn() -> RichMessage:
|
||||||
|
return RichMessage(
|
||||||
|
role="system",
|
||||||
|
content=[
|
||||||
|
TextBlock(
|
||||||
|
text="<system-reminder>Keep the answer to one short sentence.</system-reminder>"
|
||||||
|
)
|
||||||
|
],
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _without_cache_control(message: RichMessage) -> RichMessage:
|
||||||
|
return RichMessage(
|
||||||
|
role=message.role,
|
||||||
|
content=[TextBlock(text=block.text) for block in message.content],
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
RETRY_BACKOFF_SECONDS = 2.0
|
||||||
|
|
||||||
|
|
||||||
|
def retried(
|
||||||
|
call: Callable[[], Result[SessionMessagesResponse]],
|
||||||
|
attempts: int,
|
||||||
|
backoff_seconds: float = RETRY_BACKOFF_SECONDS,
|
||||||
|
sleep: Callable[[float], None] = time.sleep,
|
||||||
|
) -> Result[SessionMessagesResponse]:
|
||||||
|
result = call()
|
||||||
|
if isinstance(result, Success) or attempts <= 1:
|
||||||
|
return result
|
||||||
|
sleep(backoff_seconds)
|
||||||
|
return retried(call, attempts - 1, backoff_seconds, sleep)
|
||||||
|
|
||||||
|
|
||||||
|
def _metric(
|
||||||
|
result: Result[SessionMessagesResponse], turn_index: int, latency_seconds: float
|
||||||
|
) -> TurnMetric:
|
||||||
|
if isinstance(result, Success):
|
||||||
|
usage = result.data.usage
|
||||||
|
return TurnMetric(
|
||||||
|
turn_index=turn_index,
|
||||||
|
ok=True,
|
||||||
|
latency_seconds=latency_seconds,
|
||||||
|
uncached_input_tokens=usage.input_tokens,
|
||||||
|
cache_read_tokens=usage.cache_read_input_tokens,
|
||||||
|
cache_creation_tokens=usage.cache_creation_input_tokens,
|
||||||
|
failure=None,
|
||||||
|
)
|
||||||
|
return TurnMetric(
|
||||||
|
turn_index=turn_index,
|
||||||
|
ok=False,
|
||||||
|
latency_seconds=latency_seconds,
|
||||||
|
uncached_input_tokens=0,
|
||||||
|
cache_read_tokens=0,
|
||||||
|
cache_creation_tokens=0,
|
||||||
|
failure=repr(result),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _drive_turns(
|
||||||
|
transport: Transport,
|
||||||
|
key: str,
|
||||||
|
model: str,
|
||||||
|
marker: str,
|
||||||
|
system_block: TextBlock,
|
||||||
|
history: tuple[RichMessage, ...],
|
||||||
|
turn_index: int,
|
||||||
|
remaining_turns: int,
|
||||||
|
attempts_per_turn: int,
|
||||||
|
) -> tuple[TurnMetric, ...]:
|
||||||
|
if remaining_turns == 0:
|
||||||
|
return ()
|
||||||
|
user_turn = RichMessage(
|
||||||
|
role="user",
|
||||||
|
content=[
|
||||||
|
TextBlock(
|
||||||
|
text=_user_turn_text(marker, turn_index), cache_control=CacheControl()
|
||||||
|
)
|
||||||
|
],
|
||||||
|
)
|
||||||
|
started = time.monotonic()
|
||||||
|
result = retried(
|
||||||
|
lambda: transport.post(
|
||||||
|
"/v1/messages",
|
||||||
|
headers=transport.bearer(key),
|
||||||
|
json=SessionMessagesRequest(
|
||||||
|
model=model,
|
||||||
|
system=[system_block],
|
||||||
|
messages=[*history, user_turn],
|
||||||
|
),
|
||||||
|
response_type=SessionMessagesResponse,
|
||||||
|
),
|
||||||
|
attempts_per_turn,
|
||||||
|
)
|
||||||
|
turn = _metric(result, turn_index, time.monotonic() - started)
|
||||||
|
if not isinstance(result, Success):
|
||||||
|
return (turn,)
|
||||||
|
assistant_turn = RichMessage(
|
||||||
|
role="assistant", content=[TextBlock(text=result.data.text or "Understood.")]
|
||||||
|
)
|
||||||
|
return (
|
||||||
|
turn,
|
||||||
|
*_drive_turns(
|
||||||
|
transport,
|
||||||
|
key,
|
||||||
|
model,
|
||||||
|
marker,
|
||||||
|
system_block,
|
||||||
|
(
|
||||||
|
*history,
|
||||||
|
_without_cache_control(user_turn),
|
||||||
|
_reminder_turn(),
|
||||||
|
assistant_turn,
|
||||||
|
),
|
||||||
|
turn_index + 1,
|
||||||
|
remaining_turns - 1,
|
||||||
|
attempts_per_turn,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def run_session(
|
||||||
|
transport: Transport, key: str, model: str, turns: int, attempts_per_turn: int
|
||||||
|
) -> tuple[TurnMetric, ...]:
|
||||||
|
marker = unique_marker()
|
||||||
|
return _drive_turns(
|
||||||
|
transport,
|
||||||
|
key,
|
||||||
|
model,
|
||||||
|
marker,
|
||||||
|
_system_prefix_block(marker),
|
||||||
|
(),
|
||||||
|
1,
|
||||||
|
turns,
|
||||||
|
attempts_per_turn,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def run_concurrent_sessions(
|
||||||
|
transport: Transport,
|
||||||
|
key: str,
|
||||||
|
model: str,
|
||||||
|
sessions: int,
|
||||||
|
turns_per_session: int,
|
||||||
|
attempts_per_turn: int,
|
||||||
|
) -> tuple[TurnMetric, ...]:
|
||||||
|
with ThreadPoolExecutor(max_workers=sessions) as pool:
|
||||||
|
futures = [
|
||||||
|
pool.submit(
|
||||||
|
run_session, transport, key, model, turns_per_session, attempts_per_turn
|
||||||
|
)
|
||||||
|
for _ in range(sessions)
|
||||||
|
]
|
||||||
|
return tuple(turn for future in futures for turn in future.result())
|
||||||
|
|
||||||
|
|
||||||
|
def settled_spend(
|
||||||
|
read_spend: Callable[[], float],
|
||||||
|
poll_interval: float,
|
||||||
|
settle_seconds: float,
|
||||||
|
timeout_seconds: float,
|
||||||
|
now: Callable[[], float] = time.monotonic,
|
||||||
|
sleep: Callable[[float], None] = time.sleep,
|
||||||
|
) -> float:
|
||||||
|
deadline = now() + timeout_seconds + settle_seconds
|
||||||
|
|
||||||
|
def settle(previous: float, stable_since: float) -> float:
|
||||||
|
current = read_spend()
|
||||||
|
observed = now()
|
||||||
|
since = stable_since if current == previous else observed
|
||||||
|
if current > 0 and observed - since >= settle_seconds:
|
||||||
|
return current
|
||||||
|
if observed >= deadline:
|
||||||
|
raise AssertionError(
|
||||||
|
f"key spend never held a stable non-zero value for {settle_seconds}s "
|
||||||
|
f"within {timeout_seconds + settle_seconds}s (last read {current}); "
|
||||||
|
f"spend stopped being recorded, which is itself a spend anomaly"
|
||||||
|
)
|
||||||
|
sleep(poll_interval)
|
||||||
|
return settle(current, since)
|
||||||
|
|
||||||
|
return settle(-1.0, now())
|
||||||
|
|
||||||
|
|
||||||
|
def _p95(latencies: tuple[float, ...]) -> float:
|
||||||
|
if not latencies:
|
||||||
|
return 0.0
|
||||||
|
ranked = sorted(latencies)
|
||||||
|
return ranked[max(0, -(-len(ranked) * 95 // 100) - 1)]
|
||||||
|
|
||||||
|
|
||||||
|
def summarize(turns: tuple[TurnMetric, ...], planned_turns: int) -> AnomalyReport:
|
||||||
|
warm = tuple(turn for turn in turns if turn.ok and turn.turn_index >= 2)
|
||||||
|
return AnomalyReport(
|
||||||
|
planned_turns=planned_turns,
|
||||||
|
attempted_turns=len(turns),
|
||||||
|
failed_turns=planned_turns - sum(1 for turn in turns if turn.ok),
|
||||||
|
warm_turns=len(warm),
|
||||||
|
warm_uncached_input_tokens=sum(turn.uncached_input_tokens for turn in warm),
|
||||||
|
warm_cache_read_tokens=sum(turn.cache_read_tokens for turn in warm),
|
||||||
|
warm_cache_creation_tokens=sum(turn.cache_creation_tokens for turn in warm),
|
||||||
|
p95_turn_seconds=_p95(
|
||||||
|
tuple(turn.latency_seconds for turn in turns if turn.ok)
|
||||||
|
),
|
||||||
|
)
|
||||||
141
tests/e2e/load/test_session_anomaly.py
Normal file
141
tests/e2e/load/test_session_anomaly.py
Normal file
|
|
@ -0,0 +1,141 @@
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from itertools import count, repeat
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from e2e_http import NetworkError, Success
|
||||||
|
from session_anomaly import (
|
||||||
|
SessionMessagesResponse,
|
||||||
|
TurnMetric,
|
||||||
|
retried,
|
||||||
|
settled_spend,
|
||||||
|
summarize,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _ok_turn(turn_index: int) -> TurnMetric:
|
||||||
|
return TurnMetric(
|
||||||
|
turn_index=turn_index,
|
||||||
|
ok=True,
|
||||||
|
latency_seconds=1.0,
|
||||||
|
uncached_input_tokens=10,
|
||||||
|
cache_read_tokens=100,
|
||||||
|
cache_creation_tokens=5,
|
||||||
|
failure=None,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _failed_turn(turn_index: int) -> TurnMetric:
|
||||||
|
return TurnMetric(
|
||||||
|
turn_index=turn_index,
|
||||||
|
ok=False,
|
||||||
|
latency_seconds=1.0,
|
||||||
|
uncached_input_tokens=0,
|
||||||
|
cache_read_tokens=0,
|
||||||
|
cache_creation_tokens=0,
|
||||||
|
failure="NetworkError()",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class TestSummarizePlannedTurns:
|
||||||
|
def test_session_aborted_on_first_turn_counts_all_its_planned_turns_as_failed(
|
||||||
|
self,
|
||||||
|
) -> None:
|
||||||
|
completed_session = tuple(_ok_turn(index) for index in range(1, 7))
|
||||||
|
aborted_session = (_failed_turn(1),)
|
||||||
|
|
||||||
|
report = summarize((*completed_session, *aborted_session), planned_turns=12)
|
||||||
|
|
||||||
|
assert report.attempted_turns == 7
|
||||||
|
assert report.failed_turns == 6
|
||||||
|
assert report.error_ratio == 0.5
|
||||||
|
|
||||||
|
def test_all_planned_turns_completing_reports_zero_failures(self) -> None:
|
||||||
|
report = summarize(
|
||||||
|
tuple(_ok_turn(index) for index in range(1, 7)), planned_turns=6
|
||||||
|
)
|
||||||
|
|
||||||
|
assert report.failed_turns == 0
|
||||||
|
assert report.error_ratio == 0.0
|
||||||
|
|
||||||
|
|
||||||
|
class TestRetried:
|
||||||
|
def test_transient_failures_then_success_returns_the_success(self) -> None:
|
||||||
|
outcome = Success(data=SessionMessagesResponse())
|
||||||
|
calls = iter(
|
||||||
|
(NetworkError(message="overloaded"), NetworkError(message="overloaded"), outcome)
|
||||||
|
)
|
||||||
|
|
||||||
|
result = retried(lambda: next(calls), attempts=3, sleep=lambda _: None)
|
||||||
|
|
||||||
|
assert result is outcome
|
||||||
|
|
||||||
|
def test_exhausted_attempts_return_the_last_failure(self) -> None:
|
||||||
|
last_attempt = NetworkError(message="still overloaded")
|
||||||
|
never_reached = NetworkError(message="a fourth attempt would break the budget")
|
||||||
|
calls = iter(
|
||||||
|
(NetworkError(message="overloaded"), last_attempt, never_reached)
|
||||||
|
)
|
||||||
|
|
||||||
|
result = retried(lambda: next(calls), attempts=2, sleep=lambda _: None)
|
||||||
|
|
||||||
|
assert result is last_attempt
|
||||||
|
assert next(calls) is never_reached
|
||||||
|
|
||||||
|
def test_first_try_success_never_sleeps(self) -> None:
|
||||||
|
def sleep_means_retry(_: float) -> None:
|
||||||
|
raise AssertionError("slept after a successful attempt")
|
||||||
|
|
||||||
|
result = retried(
|
||||||
|
lambda: Success(data=SessionMessagesResponse()),
|
||||||
|
attempts=3,
|
||||||
|
sleep=sleep_means_retry,
|
||||||
|
)
|
||||||
|
|
||||||
|
assert isinstance(result, Success)
|
||||||
|
|
||||||
|
|
||||||
|
class TestSettledSpend:
|
||||||
|
def test_partial_total_between_batch_flushes_is_not_accepted_as_final(self) -> None:
|
||||||
|
reads = iter((0.1, 0.1, 0.1, 0.35, 0.35, 0.35, 0.35, 0.35))
|
||||||
|
ticks = count(0.0, 2.5)
|
||||||
|
|
||||||
|
spend = settled_spend(
|
||||||
|
lambda: next(reads),
|
||||||
|
poll_interval=5.0,
|
||||||
|
settle_seconds=10.0,
|
||||||
|
timeout_seconds=100.0,
|
||||||
|
now=lambda: next(ticks),
|
||||||
|
sleep=lambda _: None,
|
||||||
|
)
|
||||||
|
|
||||||
|
assert spend == 0.35
|
||||||
|
|
||||||
|
def test_spend_that_never_stabilizes_raises(self) -> None:
|
||||||
|
reads = (0.1 * step for step in count(1))
|
||||||
|
ticks = count(0.0, 2.5)
|
||||||
|
|
||||||
|
with pytest.raises(AssertionError, match="spend anomaly"):
|
||||||
|
settled_spend(
|
||||||
|
lambda: next(reads),
|
||||||
|
poll_interval=5.0,
|
||||||
|
settle_seconds=5.0,
|
||||||
|
timeout_seconds=10.0,
|
||||||
|
now=lambda: next(ticks),
|
||||||
|
sleep=lambda _: None,
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_spend_that_never_becomes_nonzero_raises(self) -> None:
|
||||||
|
reads = repeat(0.0)
|
||||||
|
ticks = count(0.0, 2.5)
|
||||||
|
|
||||||
|
with pytest.raises(AssertionError, match="spend anomaly"):
|
||||||
|
settled_spend(
|
||||||
|
lambda: next(reads),
|
||||||
|
poll_interval=5.0,
|
||||||
|
settle_seconds=5.0,
|
||||||
|
timeout_seconds=10.0,
|
||||||
|
now=lambda: next(ticks),
|
||||||
|
sleep=lambda _: None,
|
||||||
|
)
|
||||||
124
tests/e2e/load/test_weekly_session_anomaly_e2e.py
Normal file
124
tests/e2e/load/test_weekly_session_anomaly_e2e.py
Normal file
|
|
@ -0,0 +1,124 @@
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from dataclasses import dataclass
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from e2e_config import (
|
||||||
|
ANOMALY_MAX_ERROR_RATIO,
|
||||||
|
ANOMALY_MAX_KEY_SPEND_USD,
|
||||||
|
ANOMALY_MAX_P95_TURN_SECONDS,
|
||||||
|
ANOMALY_MIN_WARM_CACHE_READ_SHARE,
|
||||||
|
ANOMALY_SESSIONS,
|
||||||
|
ANOMALY_SPEND_SETTLE_SECONDS,
|
||||||
|
ANOMALY_TURN_ATTEMPTS,
|
||||||
|
ANOMALY_TURNS_PER_SESSION,
|
||||||
|
unique_marker,
|
||||||
|
)
|
||||||
|
from lifecycle import ResourceManager
|
||||||
|
from load_client import LoadClient
|
||||||
|
from models import KeyGenerateBody, LiteLLMParamsBody
|
||||||
|
from proxy_client import ProxyClient
|
||||||
|
from session_anomaly import run_concurrent_sessions, settled_spend, summarize
|
||||||
|
|
||||||
|
pytestmark = [pytest.mark.e2e, pytest.mark.load, pytest.mark.weekly]
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True, slots=True)
|
||||||
|
class AnomalyRoute:
|
||||||
|
route_id: str
|
||||||
|
params: LiteLLMParamsBody
|
||||||
|
|
||||||
|
|
||||||
|
ANOMALY_ROUTES = (
|
||||||
|
AnomalyRoute(
|
||||||
|
route_id="anthropic",
|
||||||
|
params=LiteLLMParamsBody(model="anthropic/claude-sonnet-5"),
|
||||||
|
),
|
||||||
|
AnomalyRoute(
|
||||||
|
route_id="bedrock_invoke",
|
||||||
|
params=LiteLLMParamsBody(
|
||||||
|
model="bedrock/invoke/us.anthropic.claude-sonnet-5",
|
||||||
|
aws_region_name="us-east-1",
|
||||||
|
),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _route_id(route: AnomalyRoute) -> str:
|
||||||
|
return route.route_id
|
||||||
|
|
||||||
|
|
||||||
|
def _settled_key_spend(proxy: ProxyClient, key: str) -> float:
|
||||||
|
return settled_spend(
|
||||||
|
lambda: proxy.key_info(key).spend or 0.0,
|
||||||
|
proxy.poll_interval,
|
||||||
|
ANOMALY_SPEND_SETTLE_SECONDS,
|
||||||
|
proxy.poll_timeout,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class TestWeeklySessionAnomaly:
|
||||||
|
@pytest.mark.covers("reliability.perf.session_anomaly.under_slo")
|
||||||
|
@pytest.mark.parametrize("route", ANOMALY_ROUTES, ids=_route_id)
|
||||||
|
def test_session_load_stays_within_baselines(
|
||||||
|
self, client: LoadClient, resources: ResourceManager, route: AnomalyRoute
|
||||||
|
) -> None:
|
||||||
|
model_name = f"weekly-anomaly-{route.route_id}-{unique_marker()}"
|
||||||
|
model_id = client.proxy.create_model(model_name, route.params)
|
||||||
|
resources.defer(lambda: client.proxy.delete_model(model_id))
|
||||||
|
key = client.proxy.generate_key(
|
||||||
|
KeyGenerateBody(models=[model_name], key_alias=model_name)
|
||||||
|
)
|
||||||
|
resources.defer(lambda: client.proxy.delete_key(key))
|
||||||
|
|
||||||
|
turns = run_concurrent_sessions(
|
||||||
|
client.proxy.transport,
|
||||||
|
key,
|
||||||
|
model_name,
|
||||||
|
ANOMALY_SESSIONS,
|
||||||
|
ANOMALY_TURNS_PER_SESSION,
|
||||||
|
ANOMALY_TURN_ATTEMPTS,
|
||||||
|
)
|
||||||
|
report = summarize(turns, ANOMALY_SESSIONS * ANOMALY_TURNS_PER_SESSION)
|
||||||
|
failures = tuple(turn.failure for turn in turns if turn.failure)
|
||||||
|
print(f"{route.route_id} anomaly report: {report}")
|
||||||
|
|
||||||
|
assert report.error_ratio <= ANOMALY_MAX_ERROR_RATIO, (
|
||||||
|
f"{route.route_id}: {report.failed_turns}/{report.planned_turns} planned "
|
||||||
|
f"turns failed or never ran because their session aborted "
|
||||||
|
f"({report.error_ratio:.1%} > {ANOMALY_MAX_ERROR_RATIO:.1%} allowed); "
|
||||||
|
f"error rate is anomalously high. Failures: {failures}"
|
||||||
|
)
|
||||||
|
assert report.warm_turns > 0, (
|
||||||
|
f"{route.route_id}: no session got past its first turn, so cache and "
|
||||||
|
f"latency baselines have nothing to read. Failures: {failures}"
|
||||||
|
)
|
||||||
|
assert report.warm_cache_read_share >= ANOMALY_MIN_WARM_CACHE_READ_SHARE, (
|
||||||
|
f"{route.route_id}: warm turns read only {report.warm_cache_read_share:.1%} "
|
||||||
|
f"of billed input tokens from the prompt cache "
|
||||||
|
f"(read={report.warm_cache_read_tokens}, "
|
||||||
|
f"creation={report.warm_cache_creation_tokens}, "
|
||||||
|
f"uncached={report.warm_uncached_input_tokens}), below the "
|
||||||
|
f"{ANOMALY_MIN_WARM_CACHE_READ_SHARE:.0%} floor; the cached prefix is "
|
||||||
|
f"being invalidated between turns (the mid-conversation-system cache "
|
||||||
|
f"collapse signature) or caching stopped working"
|
||||||
|
)
|
||||||
|
assert report.warm_cache_creation_tokens > 0, (
|
||||||
|
f"{route.route_id}: warm turns wrote 0 cache-creation tokens across "
|
||||||
|
f"{report.warm_turns} turns; the moving cache breakpoint stopped writing "
|
||||||
|
f"new prefix increments"
|
||||||
|
)
|
||||||
|
assert report.p95_turn_seconds <= ANOMALY_MAX_P95_TURN_SECONDS, (
|
||||||
|
f"{route.route_id}: p95 turn time {report.p95_turn_seconds:.1f}s exceeds "
|
||||||
|
f"the {ANOMALY_MAX_P95_TURN_SECONDS:.0f}s ceiling under "
|
||||||
|
f"{ANOMALY_SESSIONS} concurrent sessions; turn times are anomalously slow"
|
||||||
|
)
|
||||||
|
|
||||||
|
spend = _settled_key_spend(client.proxy, key)
|
||||||
|
assert spend <= ANOMALY_MAX_KEY_SPEND_USD, (
|
||||||
|
f"{route.route_id}: gateway recorded ${spend:.4f} for "
|
||||||
|
f"{report.attempted_turns} turns, above the "
|
||||||
|
f"${ANOMALY_MAX_KEY_SPEND_USD} ceiling; spend per session is "
|
||||||
|
f"anomalously high (cache regressions surface here as 2-3x spend)"
|
||||||
|
)
|
||||||
3
tests/e2e/load/weekly_anomaly_config.yml
Normal file
3
tests/e2e/load/weekly_anomaly_config.yml
Normal file
|
|
@ -0,0 +1,3 @@
|
||||||
|
general_settings:
|
||||||
|
master_key: os.environ/LITELLM_MASTER_KEY
|
||||||
|
store_model_in_db: true
|
||||||
|
|
@ -6,3 +6,4 @@ addopts = --strict-markers --strict-config
|
||||||
markers =
|
markers =
|
||||||
e2e: live test that requires a running proxy and real provider keys
|
e2e: live test that requires a running proxy and real provider keys
|
||||||
load: heavy throughput/load test; collected last so it never perturbs latency-sensitive suites
|
load: heavy throughput/load test; collected last so it never perturbs latency-sensitive suites
|
||||||
|
weekly: real-provider anomaly load test that spends real money; deselected unless E2E_WEEKLY_ANOMALY is set
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue