mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-14 23:21:35 +00:00
test(e2e): rewrite the Redis timeout test as a locust chaos load test
The sequential version sent one request at a time, so a Redis outage never reached the concurrency where the failed-tracking alert body actually grows. This drives the proxy with locust against one model group of three mock deployments, two failing at order 1 and one serving at order 2, so every request spends its retries on the failing pair and lands on the serving deployment through the order-based fallback. Two phases, a healthy baseline and a CLIENT PAUSE WRITE window, and every request must succeed in both. Latency, RSS and CPU are reported as p50/p90/p99 per phase rather than asserted on: RSS and CPU come from psutil on the proxy's process tree, since a multi-worker proxy serves /metrics from the prometheus multiprocess collector and that drops the process collector's series. Thresholds stay open until weekly runs give real baselines. Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
Kerry Lu
parent
be7dce30a3
commit
33ec56ed75
21 changed files with 673 additions and 354 deletions
1
.github/e2e-stack/select_tests.py
vendored
1
.github/e2e-stack/select_tests.py
vendored
|
|
@ -8,7 +8,6 @@ UNSUPPORTED: Final = re.compile(
|
|||
r"|^tests/e2e/llm_translation/realtime/test_realtime_pipecat_audio_e2e\.py$"
|
||||
r"|^tests/e2e/batches/test_managed_files_enforcement_e2e\.py$"
|
||||
r"|^tests/e2e/guardrails/test_presidio_masking_e2e\.py$"
|
||||
r"|^tests/e2e/router/test_redis_timeout_e2e\.py$"
|
||||
)
|
||||
HARNESS: Final = re.compile(
|
||||
r"^tests/e2e/[A-Za-z0-9_.-]+\.(py|ini)$"
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
name: "Weekly Redis Timeout E2E"
|
||||
name: "Weekly Redis Chaos E2E"
|
||||
|
||||
on:
|
||||
schedule:
|
||||
|
|
@ -9,9 +9,9 @@ permissions:
|
|||
contents: read
|
||||
|
||||
jobs:
|
||||
redis-timeout-e2e:
|
||||
redis-chaos-e2e:
|
||||
if: github.event_name != 'schedule' || github.repository == 'BerriAI/litellm'
|
||||
runs-on: ubuntu-latest
|
||||
runs-on: ubuntu-latest-16-cores
|
||||
timeout-minutes: 30
|
||||
services:
|
||||
postgres:
|
||||
|
|
@ -38,7 +38,7 @@ jobs:
|
|||
--health-retries 10
|
||||
env:
|
||||
DATABASE_URL: postgresql://llmproxy:dbpassword9090@localhost:5432/litellm
|
||||
LITELLM_MASTER_KEY: sk-redis-timeout-e2e
|
||||
LITELLM_MASTER_KEY: sk-redis-chaos-e2e
|
||||
LITELLM_LOG: WARNING
|
||||
steps:
|
||||
- uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
|
||||
|
|
@ -60,7 +60,7 @@ jobs:
|
|||
|
||||
- name: Install dependencies
|
||||
run: |
|
||||
.github/scripts/uv_sync_with_retries.sh --frozen --group ci --group proxy-dev --extra proxy
|
||||
.github/scripts/uv_sync_with_retries.sh --frozen --group ci --group proxy-dev --group e2e-dev --extra proxy
|
||||
|
||||
- name: Cache Prisma binaries
|
||||
uses: ./.github/actions/cache-prisma-binaries
|
||||
|
|
@ -69,9 +69,10 @@ jobs:
|
|||
run: |
|
||||
uv run --no-sync prisma generate --schema litellm/proxy/schema.prisma
|
||||
|
||||
- name: Start the proxy with a Redis that times out on every command
|
||||
- name: Start a multi-worker proxy on the chaos config
|
||||
run: |
|
||||
nohup uv run --no-sync litellm --config tests/e2e/gateway/redis_timeout_ci_config.yml --port 4000 > proxy.log 2>&1 &
|
||||
nohup uv run --no-sync litellm --config tests/e2e/gateway/redis_chaos_ci_config.yml --port 4000 --num_workers 4 > proxy.log 2>&1 &
|
||||
echo "E2E_PROXY_PID=$!" >> "$GITHUB_ENV"
|
||||
for _ in $(seq 1 90); do
|
||||
if curl -fs http://localhost:4000/health/liveliness > /dev/null; then
|
||||
exit 0
|
||||
|
|
@ -82,14 +83,14 @@ jobs:
|
|||
tail -n 100 proxy.log
|
||||
exit 1
|
||||
|
||||
- name: Run the Redis timeout e2e test
|
||||
- name: Run the Redis chaos load test
|
||||
env:
|
||||
E2E_REDIS_TIMEOUT: "1"
|
||||
E2E_REDIS_CHAOS: "1"
|
||||
LITELLM_PROXY_URL: http://localhost:4000
|
||||
REDIS_HOST: 127.0.0.1
|
||||
REDIS_PORT: "6379"
|
||||
run: |
|
||||
uv run --no-sync pytest tests/e2e/router/test_redis_timeout_e2e.py -v --tb=short -rA
|
||||
uv run --no-sync pytest tests/e2e/load/test_redis_chaos_e2e.py -v --tb=short -rA -s
|
||||
|
||||
- name: Show proxy log on failure
|
||||
if: failure()
|
||||
|
|
@ -220,6 +220,7 @@ e2e-dev = [
|
|||
"playwright==1.61.0",
|
||||
"websockets>=15.0.1,<16.0",
|
||||
"locust==2.45.0",
|
||||
"psutil==7.2.2",
|
||||
"mcp>=1.28.1,<2.0",
|
||||
]
|
||||
proxy-dev = [
|
||||
|
|
|
|||
|
|
@ -17,8 +17,8 @@ Each subdirectory under `tests/e2e/` is one suite, scoped to an endpoint family
|
|||
- `mcp/` - the MCP server surface over api_key auth against the real Datadog remote MCP server (see "MCP suite: real Datadog only" below); plus the gateway-managed OAuth (authorization_code) path exercised through `/chat/completions`, the one behavior Datadog's static-header auth cannot reach, seeding the per-user upstream token via the interactive authorize dance driven with the mcp SDK's own OAuth client (headless-browser consent from a saved session) and asserting the completion lists and executes the server's tools with the stored per-user token
|
||||
- `logging/` - logging-integration delivery (datadog and friends)
|
||||
- `security/` - secret handling and log-leak protection
|
||||
- `router/` - routing and reliability behavior (fallbacks, cooldowns). Also holds `test_redis_timeout_e2e.py`, which needs a proxy booted from `gateway/redis_timeout_ci_config.yml` (real Redis, `socket_timeout: 0.001`, so every command times out); it is marked `redis_timeout`, deselected unless `E2E_REDIS_TIMEOUT` is set, excluded from the per-PR check, and driven weekly by `.github/workflows/test-e2e-redis-timeout.yml`
|
||||
- `load/` - performance-category tests, kept OUT of the main suite: throughput/load SLO tests are a different testing category from functional e2e (variance-driven, historically flaky) and live outside this suite until re-implemented as their own pipeline (LIT-5163); do not add a live load test that runs in the default collection. What remains here: 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; marked `weekly` and deselected unless `E2E_WEEKLY_ANOMALY` is set, driven by `.github/workflows/weekly_load_anomaly.yml`) and markerless harness unit tests for the Locust/session-anomaly aggregation logic
|
||||
- `router/` - routing and reliability behavior (fallbacks, cooldowns)
|
||||
- `load/` - performance-category tests, kept OUT of the main suite: throughput/load SLO tests are a different testing category from functional e2e (variance-driven, historically flaky) and live outside this suite until re-implemented as their own pipeline (LIT-5163); do not add a live load test that runs in the default collection. What lives here: 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; marked `weekly` and deselected unless `E2E_WEEKLY_ANOMALY` is set, driven by `.github/workflows/weekly_load_anomaly.yml`), the Redis chaos test (`test_redis_chaos_e2e.py`, locust load against mock deployments with `CLIENT PAUSE WRITE` on the proxy's Redis mid-run, asserting zero failed requests and reporting latency, RSS, and CPU as p50/p90/p99 per phase; needs a proxy booted from `gateway/redis_chaos_ci_config.yml` on the same host with `E2E_PROXY_PID` set, marked `redis_chaos`, deselected unless `E2E_REDIS_CHAOS` is set, driven weekly by `.github/workflows/test-e2e-redis-chaos.yml`), and markerless harness unit tests for the locust, process-usage, and session-anomaly aggregation logic
|
||||
- `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
|
||||
- `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
|
||||
|
|
|
|||
|
|
@ -65,7 +65,7 @@ A couple of logging destinations are configured on the proxy rather than by the
|
|||
|
||||
### The pull request check
|
||||
|
||||
Every same-repository PR that adds, modifies, or renames a `tests/e2e/**/test_*.py` file runs those changed files three times. A change to the harness itself, meaning a root-level `tests/e2e/*.py` file or `pytest.ini`, `tests/e2e/gateway/`, `.github/e2e-stack/`, or the workflow, also runs the `access_control` suite as a canary, because those files have no test of their own that exercises the stack. `.github/e2e-stack/select_tests.py` applies both rules. The stack config at `tests/e2e/gateway/stage_mirror_ci_config.yml` must declare every model the selected suites use; a missing one shows up as a failed test id in the public log. The suite's own single rerun for network errors and 5xx responses (see `pytest.ini`) applies on every pass, so a transport blip does not fail the check while a race inside a test still does. The stage-mirror stack has a control-plane backend, two gateways behind nginx, Postgres, Jaeger, and TLS cluster-mode Valkey. The stack exports every gateway address in `LITELLM_PROXY_REPLICA_URLS`, so model registration waits until each gateway lists the new model rather than whichever one the load balancer answered from. Documentation, deleted-file, and application-only changes do not start the stack or request environment approval. The `ui/`, `claude_code/`, and `load/` directories, `batches/test_managed_files_enforcement_e2e.py`, `llm_translation/realtime/test_realtime_pipecat_audio_e2e.py`, `guardrails/test_presidio_masking_e2e.py`, and `router/test_redis_timeout_e2e.py` remain outside this check because they use separate tooling or need a differently configured stack: the pipecat audio suite skips itself at import time unless the NLTK `punkt_tab` data is installed, and the presidio suite fails without the analyzer and anonymizer services this stack does not start, and the Redis timeout test needs a proxy whose Redis times out on every command (`gateway/redis_timeout_ci_config.yml`), which `.github/workflows/test-e2e-redis-timeout.yml` boots on a weekly schedule
|
||||
Every same-repository PR that adds, modifies, or renames a `tests/e2e/**/test_*.py` file runs those changed files three times. A change to the harness itself, meaning a root-level `tests/e2e/*.py` file or `pytest.ini`, `tests/e2e/gateway/`, `.github/e2e-stack/`, or the workflow, also runs the `access_control` suite as a canary, because those files have no test of their own that exercises the stack. `.github/e2e-stack/select_tests.py` applies both rules. The stack config at `tests/e2e/gateway/stage_mirror_ci_config.yml` must declare every model the selected suites use; a missing one shows up as a failed test id in the public log. The suite's own single rerun for network errors and 5xx responses (see `pytest.ini`) applies on every pass, so a transport blip does not fail the check while a race inside a test still does. The stage-mirror stack has a control-plane backend, two gateways behind nginx, Postgres, Jaeger, and TLS cluster-mode Valkey. The stack exports every gateway address in `LITELLM_PROXY_REPLICA_URLS`, so model registration waits until each gateway lists the new model rather than whichever one the load balancer answered from. Documentation, deleted-file, and application-only changes do not start the stack or request environment approval. The `ui/`, `claude_code/`, and `load/` directories, `batches/test_managed_files_enforcement_e2e.py`, `llm_translation/realtime/test_realtime_pipecat_audio_e2e.py`, and `guardrails/test_presidio_masking_e2e.py` remain outside this check because they use separate tooling or need a differently configured stack: the pipecat audio suite skips itself at import time unless the NLTK `punkt_tab` data is installed, and the presidio suite fails without the analyzer and anonymizer services this stack does not start. The Redis chaos test under `load/` needs a proxy it can pause the Redis of on the same host (`gateway/redis_chaos_ci_config.yml`), which `.github/workflows/test-e2e-redis-chaos.yml` boots on a weekly schedule
|
||||
|
||||
Every selected file must execute at least one passing test in each pass, and any test failure, collection error, or entirely skipped or deselected file fails the check. A file whose tests are all marked skip therefore cannot pass this check, so unskip at least one of them, or add the file to `UNSUPPORTED` in `select_tests.py` with the reason, before changing one. A failed pass stops the run. The public log prints pytest's one-line summary for each pass, including the rerun count, and names each failed or errored test as `classname::name`, so a retried network error or a failing test is visible without the raw output. The final `e2e-changed-tests` job succeeds only when no supported test files changed or the approved run completed all three passes. Fork PRs with selected tests fail this gate until a maintainer brings the reviewed change onto a same-repository branch
|
||||
|
||||
|
|
|
|||
|
|
@ -60,8 +60,8 @@ def pytest_configure(config: pytest.Config) -> None:
|
|||
)
|
||||
config.addinivalue_line(
|
||||
"markers",
|
||||
"redis_timeout: needs a proxy booted from gateway/redis_timeout_ci_config.yml whose Redis times out on every "
|
||||
"command; deselected unless E2E_REDIS_TIMEOUT is set",
|
||||
"redis_chaos: load test that pauses the proxy's Redis writes mid-run; needs a proxy booted from "
|
||||
"gateway/redis_chaos_ci_config.yml on the same host, and is deselected unless E2E_REDIS_CHAOS is set",
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -31,7 +31,7 @@
|
|||
- {id: reliability.cache.exact.returns_cached, module: reliability, tier: P1, behavior: cache, variant: exact, assertions: [returns_cached], exercised_on: [chat_completions, messages, embeddings], source: "litellm/caching/caching.py", rationale: "Response cache returns cached on exact match"}
|
||||
- {id: reliability.cache.prompt_caching_model_select.returns_cached, module: reliability, tier: P1, behavior: cache, variant: prompt_caching_model_select, assertions: [returns_cached], exercised_on: [chat_completions], source: "router_utils/prompt_caching_cache.py", rationale: "Selects model supporting prompt caching for cacheable prefix"}
|
||||
- {id: reliability.circuit_breaker.redis.trips_then_recovers, module: reliability, tier: P0, behavior: circuit_breaker, variant: redis, assertions: [trips_then_recovers], exercised_on: [chat_completions, messages, embeddings], source: "litellm/caching/redis_cache.py:99", rationale: "Redis breaker CLOSED->OPEN->HALF_OPEN; guards all cache/rate-limit ops"}
|
||||
- {id: reliability.circuit_breaker.redis_timeout.stays_responsive, module: reliability, tier: P1, behavior: circuit_breaker, variant: redis_timeout, assertions: [stays_responsive], exercised_on: [chat_completions, responses, embeddings], source: "litellm/proxy/hooks/proxy_track_cost_callback.py:386", fail_before_fix: proven, rationale: "With every Redis command timing out and every request retrying then falling back, per-request latency stays flat, liveliness stays fast, and spend rows still land; on v1.100.0 the failed-tracking alert body doubled per request until the worker OOMed (LIT-6780)"}
|
||||
- {id: reliability.circuit_breaker.redis_timeout.stays_responsive, module: reliability, tier: P1, behavior: circuit_breaker, variant: redis_timeout, assertions: [stays_responsive], exercised_on: [chat_completions], source: "litellm/proxy/hooks/proxy_track_cost_callback.py:386", fail_before_fix: proven, rationale: "Under locust load with every request retrying through failing mock deployments, pausing Redis writes mid-run trips the breaker and every request still succeeds, with latency, RSS, and CPU reported as p50/p90/p99 against the pre-pause baseline; on v1.100.0 the failed-tracking alert body doubled per request until the worker OOMed (LIT-6780)"}
|
||||
- {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.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"}
|
||||
|
|
|
|||
|
|
@ -143,7 +143,7 @@ LOAD_MIN_CONCURRENCY_EFFICIENCY = float(os.environ.get("E2E_LOAD_MIN_CONCURRENCY
|
|||
|
||||
WEEKLY_ANOMALY_OPT_IN_ENV = "E2E_WEEKLY_ANOMALY"
|
||||
MANAGED_FILES_OPT_IN_ENV = "E2E_MANAGED_FILES_STACK"
|
||||
REDIS_TIMEOUT_OPT_IN_ENV = "E2E_REDIS_TIMEOUT"
|
||||
REDIS_CHAOS_OPT_IN_ENV = "E2E_REDIS_CHAOS"
|
||||
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"))
|
||||
|
|
|
|||
|
|
@ -5,17 +5,14 @@ general_settings:
|
|||
litellm_settings:
|
||||
callbacks: ["prometheus"]
|
||||
require_auth_for_metrics_endpoint: false
|
||||
enable_redis_auth_cache: true
|
||||
cache: true
|
||||
cache_params:
|
||||
type: redis
|
||||
host: 127.0.0.1
|
||||
port: 6379
|
||||
socket_timeout: 0.001
|
||||
socket_timeout: 0.1
|
||||
|
||||
router_settings:
|
||||
num_retries: 1
|
||||
fallbacks:
|
||||
- redis-timeout-primary:
|
||||
- redis-timeout-backup
|
||||
- redis-timeout-embed-primary:
|
||||
- redis-timeout-embed-backup
|
||||
num_retries: 2
|
||||
disable_cooldowns: true
|
||||
|
|
@ -4,25 +4,23 @@ import os
|
|||
|
||||
import pytest
|
||||
|
||||
from e2e_config import WEEKLY_ANOMALY_OPT_IN_ENV
|
||||
from e2e_config import REDIS_CHAOS_OPT_IN_ENV, WEEKLY_ANOMALY_OPT_IN_ENV
|
||||
from load_client import LoadClient, build_client
|
||||
from proxy_client import ProxyClient
|
||||
|
||||
_OPT_IN_MARKERS = (
|
||||
("weekly", WEEKLY_ANOMALY_OPT_IN_ENV),
|
||||
("redis_chaos", REDIS_CHAOS_OPT_IN_ENV),
|
||||
)
|
||||
|
||||
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
|
||||
]
|
||||
|
||||
def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item]) -> None:
|
||||
opted_out = {marker for marker, opt_in_env in _OPT_IN_MARKERS if not os.environ.get(opt_in_env)}
|
||||
deselected = [item for item in items if any(item.get_closest_marker(marker) is not None for marker in opted_out)]
|
||||
if not deselected:
|
||||
return
|
||||
config.hook.pytest_deselected(items=deselected)
|
||||
items[:] = [
|
||||
item for item in items if item.get_closest_marker("weekly") is None
|
||||
]
|
||||
items[:] = [item for item in items if item not in deselected]
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
|
|
|
|||
|
|
@ -1,12 +1,19 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from dataclasses import dataclass
|
||||
from itertools import accumulate
|
||||
from pathlib import Path
|
||||
from typing import Final
|
||||
|
||||
from pydantic import BaseModel, TypeAdapter
|
||||
|
||||
_LOCUSTFILE = Path(__file__).with_name("locustfile.py")
|
||||
_CSV_PREFIX = "locust"
|
||||
_GENERATOR_SATURATION_MARKER = "CPU usage above"
|
||||
_MAX_REPORTED_ERRORS = 5
|
||||
|
||||
|
|
@ -34,7 +41,9 @@ class LoadResult:
|
|||
requests: int
|
||||
failures: int
|
||||
requests_per_second: float
|
||||
median_response_seconds: float
|
||||
p50_seconds: float
|
||||
p90_seconds: float
|
||||
p99_seconds: float
|
||||
errors: tuple[LoadError, ...]
|
||||
generator_warnings: tuple[str, ...]
|
||||
|
||||
|
|
@ -53,18 +62,24 @@ class LoadResult:
|
|||
lines.append("locust recorded no error breakdown")
|
||||
return "; ".join((*lines, *self.generator_warnings))
|
||||
|
||||
def latency_summary(self) -> str:
|
||||
return f"p50 {self.p50_seconds:.3f}s, p90 {self.p90_seconds:.3f}s, p99 {self.p99_seconds:.3f}s"
|
||||
|
||||
def median_seconds(entries: list[LocustStatEntry]) -> float:
|
||||
samples = sorted(
|
||||
(milliseconds, count) for entry in entries for milliseconds, count in entry.response_times.items()
|
||||
)
|
||||
|
||||
def percentile_seconds(entries: list[LocustStatEntry], fraction: float) -> float:
|
||||
"""The response time at `fraction` of the merged histograms, in seconds.
|
||||
|
||||
Locust buckets response times by millisecond, so this reads the first bucket whose
|
||||
running count reaches the rank, the same lower-sample convention locust's own
|
||||
percentiles use.
|
||||
"""
|
||||
samples = sorted((milliseconds, count) for entry in entries for milliseconds, count in entry.response_times.items())
|
||||
total = sum(count for _, count in samples)
|
||||
if total == 0:
|
||||
return 0.0
|
||||
running = accumulate(count for _, count in samples)
|
||||
return next(
|
||||
milliseconds for (milliseconds, _), seen in zip(samples, running) if seen >= total / 2
|
||||
) / 1000.0
|
||||
rank: Final = total * fraction
|
||||
return next(milliseconds for (milliseconds, _), seen in zip(samples, running) if seen >= rank) / 1000.0
|
||||
|
||||
|
||||
def aggregate_stats(
|
||||
|
|
@ -79,7 +94,9 @@ def aggregate_stats(
|
|||
requests=requests,
|
||||
failures=failures,
|
||||
requests_per_second=0.0,
|
||||
median_response_seconds=0.0,
|
||||
p50_seconds=0.0,
|
||||
p90_seconds=0.0,
|
||||
p99_seconds=0.0,
|
||||
errors=errors,
|
||||
generator_warnings=generator_warnings,
|
||||
)
|
||||
|
|
@ -88,7 +105,9 @@ def aggregate_stats(
|
|||
requests=requests,
|
||||
failures=failures,
|
||||
requests_per_second=requests / elapsed if elapsed > 0 else 0.0,
|
||||
median_response_seconds=median_seconds(entries),
|
||||
p50_seconds=percentile_seconds(entries, 0.5),
|
||||
p90_seconds=percentile_seconds(entries, 0.9),
|
||||
p99_seconds=percentile_seconds(entries, 0.99),
|
||||
errors=errors,
|
||||
generator_warnings=generator_warnings,
|
||||
)
|
||||
|
|
@ -121,3 +140,67 @@ def read_generator_warnings(stderr: str) -> tuple[str, ...]:
|
|||
if _GENERATOR_SATURATION_MARKER in line
|
||||
)
|
||||
return tuple(dict.fromkeys(saturated))
|
||||
|
||||
|
||||
def run_chat_load(
|
||||
*,
|
||||
base_url: str,
|
||||
api_keys: tuple[str, ...],
|
||||
model: str,
|
||||
users: int,
|
||||
spawn_rate: float,
|
||||
duration_seconds: float,
|
||||
) -> LoadResult:
|
||||
"""Drive /chat/completions from headless locust and aggregate what it reported.
|
||||
|
||||
Each simulated user picks one of `api_keys`, so auth and budget lookups spread over a
|
||||
pool of virtual keys instead of keeping one key's cache entry permanently warm.
|
||||
"""
|
||||
with tempfile.TemporaryDirectory(prefix="e2e-load-") as report_dir:
|
||||
csv_prefix = Path(report_dir) / _CSV_PREFIX
|
||||
completed = subprocess.run(
|
||||
[
|
||||
sys.executable,
|
||||
"-m",
|
||||
"locust",
|
||||
"--headless",
|
||||
"--json",
|
||||
"--csv",
|
||||
str(csv_prefix),
|
||||
"--locustfile",
|
||||
str(_LOCUSTFILE),
|
||||
"--host",
|
||||
base_url,
|
||||
"--users",
|
||||
str(users),
|
||||
"--spawn-rate",
|
||||
str(spawn_rate),
|
||||
"--run-time",
|
||||
f"{int(duration_seconds)}s",
|
||||
"--exit-code-on-error",
|
||||
"0",
|
||||
],
|
||||
env={**os.environ, "LOAD_API_KEYS": ",".join(api_keys), "LOAD_MODEL": model},
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=duration_seconds + 120,
|
||||
check=False,
|
||||
)
|
||||
if completed.returncode != 0:
|
||||
raise RuntimeError(
|
||||
f"locust exited {completed.returncode} before it could report throughput "
|
||||
f"(a startup failure, not request failures, which are folded into the JSON summary via "
|
||||
f"--exit-code-on-error 0):\n{completed.stderr}"
|
||||
)
|
||||
try:
|
||||
entries = _STATS_ADAPTER.validate_json(completed.stdout)
|
||||
except ValueError as exc:
|
||||
raise RuntimeError(
|
||||
f"locust exited 0 but did not print a parseable --json throughput summary on stdout; "
|
||||
f"got stdout={completed.stdout!r}, stderr={completed.stderr!r}"
|
||||
) from exc
|
||||
return aggregate_stats(
|
||||
entries,
|
||||
read_errors(csv_prefix.with_name(f"{_CSV_PREFIX}_failures.csv")),
|
||||
read_generator_warnings(completed.stderr),
|
||||
)
|
||||
|
|
|
|||
36
tests/e2e/load/locustfile.py
Normal file
36
tests/e2e/load/locustfile.py
Normal file
|
|
@ -0,0 +1,36 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import random
|
||||
import uuid
|
||||
from typing import Final
|
||||
|
||||
from locust import FastHttpUser, constant, task
|
||||
|
||||
_MODEL: Final = os.environ["LOAD_MODEL"]
|
||||
_API_KEYS: Final = tuple(os.environ["LOAD_API_KEYS"].split(","))
|
||||
|
||||
|
||||
def _payload() -> dict[str, object]:
|
||||
"""A prompt no other request sent, so the response cache never answers for the deployment."""
|
||||
return {
|
||||
"model": _MODEL,
|
||||
"messages": [{"role": "user", "content": f"load test ping {uuid.uuid4().hex}"}],
|
||||
"max_tokens": 16,
|
||||
}
|
||||
|
||||
|
||||
class ChatUser(FastHttpUser):
|
||||
wait_time = constant(0)
|
||||
|
||||
def on_start(self) -> None:
|
||||
self.headers = {"Authorization": f"Bearer {random.choice(_API_KEYS)}"}
|
||||
|
||||
@task
|
||||
def chat(self) -> None:
|
||||
self.client.post( # pyright: ignore[reportUnknownMemberType] # locust FastHttpSession.post types json/**kwargs as Any
|
||||
"/chat/completions",
|
||||
json=_payload(),
|
||||
headers=self.headers,
|
||||
name="/chat/completions",
|
||||
)
|
||||
156
tests/e2e/load/proxy_usage.py
Normal file
156
tests/e2e/load/proxy_usage.py
Normal file
|
|
@ -0,0 +1,156 @@
|
|||
"""Resident memory and CPU of the proxy process tree, sampled on a background thread.
|
||||
|
||||
The proxy under load runs several worker processes, and `/metrics` cannot report their
|
||||
memory: litellm sets PROMETHEUS_MULTIPROC_DIR when num_workers > 1, and the multiprocess
|
||||
collector drops the process collector's `process_resident_memory_bytes` /
|
||||
`process_cpu_seconds_total` entirely. So the test measures the tree itself through psutil,
|
||||
which needs the proxy to run on the same host as the test.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
import threading
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from typing import Final
|
||||
|
||||
import psutil
|
||||
from pydantic import BaseModel, ConfigDict
|
||||
|
||||
|
||||
class _MemoryInfo(BaseModel):
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
rss: int
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class UsageSample:
|
||||
elapsed_seconds: float
|
||||
rss_bytes: int
|
||||
cpu_seconds: float
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class UsageWindow:
|
||||
"""The samples taken across one phase, plus what they say about that phase."""
|
||||
|
||||
samples: tuple[UsageSample, ...]
|
||||
|
||||
def rss_percentile(self, fraction: float) -> int:
|
||||
if not self.samples:
|
||||
return 0
|
||||
ordered: Final = sorted(sample.rss_bytes for sample in self.samples)
|
||||
return ordered[_rank(len(ordered), fraction)]
|
||||
|
||||
def cpu_seconds_consumed(self) -> float:
|
||||
"""CPU seconds the tree burned across the window, from its monotonic counter."""
|
||||
if len(self.samples) < 2:
|
||||
return 0.0
|
||||
return self.samples[-1].cpu_seconds - self.samples[0].cpu_seconds
|
||||
|
||||
def cpu_utilization_percentiles(self) -> tuple[float, float, float]:
|
||||
"""Per-interval CPU utilization (cores busy) at p50, p90 and p99.
|
||||
|
||||
Derived from consecutive samples of the cumulative counter rather than
|
||||
psutil's own cpu_percent, so it covers every process in the tree including
|
||||
workers that came and went between samples.
|
||||
"""
|
||||
rates: Final = sorted(
|
||||
(later.cpu_seconds - earlier.cpu_seconds) / (later.elapsed_seconds - earlier.elapsed_seconds)
|
||||
for earlier, later in zip(self.samples, self.samples[1:])
|
||||
if later.elapsed_seconds > earlier.elapsed_seconds
|
||||
)
|
||||
if not rates:
|
||||
return 0.0, 0.0, 0.0
|
||||
return (
|
||||
rates[_rank(len(rates), 0.5)],
|
||||
rates[_rank(len(rates), 0.9)],
|
||||
rates[_rank(len(rates), 0.99)],
|
||||
)
|
||||
|
||||
def summary(self) -> str:
|
||||
p50_cpu, p90_cpu, p99_cpu = self.cpu_utilization_percentiles()
|
||||
return (
|
||||
f"RSS p50 {self.rss_percentile(0.5) / 2**20:.0f} MB, "
|
||||
f"p90 {self.rss_percentile(0.9) / 2**20:.0f} MB, "
|
||||
f"p99 {self.rss_percentile(0.99) / 2**20:.0f} MB; "
|
||||
f"CPU cores busy p50 {p50_cpu:.2f}, p90 {p90_cpu:.2f}, p99 {p99_cpu:.2f}; "
|
||||
f"{self.cpu_seconds_consumed():.1f} CPU seconds consumed"
|
||||
)
|
||||
|
||||
|
||||
def _rank(count: int, fraction: float) -> int:
|
||||
"""Index of the sample at `fraction`, the same lower-sample convention as locust's percentiles."""
|
||||
return min(count - 1, max(0, math.ceil(count * fraction) - 1))
|
||||
|
||||
|
||||
def _read_process(process: psutil.Process) -> tuple[int, float] | None:
|
||||
try:
|
||||
with process.oneshot():
|
||||
memory: Final = _MemoryInfo.model_validate(process.memory_info())
|
||||
times: Final = process.cpu_times()
|
||||
return memory.rss, times.user + times.system
|
||||
except (psutil.NoSuchProcess, psutil.AccessDenied):
|
||||
return None
|
||||
|
||||
|
||||
class ProxyUsageSampler:
|
||||
"""Samples the proxy process tree every `interval_seconds` until stopped.
|
||||
|
||||
`split()` returns the samples taken so far and starts a new window, so one sampler
|
||||
covers a baseline phase and a chaos phase without a gap between them.
|
||||
"""
|
||||
|
||||
def __init__(self, pid: int, interval_seconds: float = 1.0) -> None:
|
||||
self._process: Final = psutil.Process(pid)
|
||||
self._interval: Final = interval_seconds
|
||||
self._stop: Final = threading.Event()
|
||||
self._lock: Final = threading.Lock()
|
||||
self._samples: list[UsageSample] = [] # mutable-ok: a sampling buffer the reader drains under a lock
|
||||
self._started: Final = time.monotonic()
|
||||
self._thread: Final = threading.Thread(target=self._run, name="proxy-usage-sampler", daemon=True)
|
||||
|
||||
def __enter__(self) -> ProxyUsageSampler:
|
||||
self._thread.start()
|
||||
return self
|
||||
|
||||
def __exit__(self, *_: object) -> None:
|
||||
self._stop.set()
|
||||
self._thread.join(timeout=self._interval * 5)
|
||||
|
||||
def _tree(self) -> tuple[psutil.Process, ...]:
|
||||
try:
|
||||
return (self._process, *self._process.children(recursive=True))
|
||||
except psutil.NoSuchProcess:
|
||||
return ()
|
||||
|
||||
def _sample(self) -> UsageSample | None:
|
||||
readings: Final = tuple(reading for process in self._tree() if (reading := _read_process(process)) is not None)
|
||||
if not readings:
|
||||
return None
|
||||
return UsageSample(
|
||||
elapsed_seconds=time.monotonic() - self._started,
|
||||
rss_bytes=sum(rss for rss, _ in readings),
|
||||
cpu_seconds=sum(cpu for _, cpu in readings),
|
||||
)
|
||||
|
||||
def _run(self) -> None:
|
||||
while not self._stop.is_set():
|
||||
sample = self._sample()
|
||||
if sample is not None:
|
||||
with self._lock:
|
||||
self._samples.append(sample)
|
||||
self._stop.wait(self._interval)
|
||||
|
||||
def split(self) -> UsageWindow:
|
||||
"""The window that ends now; the next one starts from this window's last sample.
|
||||
|
||||
The boundary sample is carried into the next window so its CPU counter has a
|
||||
starting point, which is what makes the two windows' utilization comparable.
|
||||
"""
|
||||
with self._lock:
|
||||
taken = tuple(self._samples)
|
||||
self._samples = [taken[-1]] if taken else [] # rebind-ok: drains the buffer under the lock
|
||||
return UsageWindow(samples=taken)
|
||||
|
|
@ -7,7 +7,7 @@ from locust_load import (
|
|||
LoadResult,
|
||||
LocustStatEntry,
|
||||
aggregate_stats,
|
||||
median_seconds,
|
||||
percentile_seconds,
|
||||
read_errors,
|
||||
read_generator_warnings,
|
||||
)
|
||||
|
|
@ -41,35 +41,47 @@ def _result(
|
|||
requests=10,
|
||||
failures=10,
|
||||
requests_per_second=1.0,
|
||||
median_response_seconds=0.05,
|
||||
p50_seconds=0.05,
|
||||
p90_seconds=0.08,
|
||||
p99_seconds=0.1,
|
||||
errors=errors,
|
||||
generator_warnings=generator_warnings,
|
||||
)
|
||||
|
||||
|
||||
class TestSerialLatency:
|
||||
class TestPercentiles:
|
||||
def test_median_is_the_middle_sample_not_the_mean_a_slow_tail_would_drag(self) -> None:
|
||||
# Nine fast requests and one very slow one: the mean is 1.99s, the median is 20ms.
|
||||
entry = _entry(num_requests=10, response_times={20: 9, 20000: 1})
|
||||
|
||||
assert median_seconds([entry]) == 0.02
|
||||
assert percentile_seconds([entry], 0.5) == 0.02
|
||||
|
||||
def test_median_merges_the_histograms_of_every_stats_entry(self) -> None:
|
||||
def test_the_tail_percentiles_reach_the_slow_samples_the_median_hides(self) -> None:
|
||||
# 100 samples: 89 fast, 10 slow, 1 very slow. p50 sits in the fast bucket, p90 in the
|
||||
# slow one, and p99 lands on the single very slow sample.
|
||||
entry = _entry(num_requests=100, response_times={20: 89, 500: 10, 20000: 1})
|
||||
|
||||
assert percentile_seconds([entry], 0.5) == 0.02
|
||||
assert percentile_seconds([entry], 0.9) == 0.5
|
||||
assert percentile_seconds([entry], 0.99) == 0.5
|
||||
assert percentile_seconds([entry], 1.0) == 20.0
|
||||
|
||||
def test_percentiles_merge_the_histograms_of_every_stats_entry(self) -> None:
|
||||
# Per entry the median would be 10ms and 90ms; merged, the middle of the five samples is 90ms.
|
||||
entries = [
|
||||
_entry(num_requests=2, response_times={10: 2}),
|
||||
_entry(num_requests=3, response_times={90: 3}),
|
||||
]
|
||||
|
||||
assert median_seconds(entries) == 0.09
|
||||
assert percentile_seconds(entries, 0.5) == 0.09
|
||||
|
||||
def test_an_even_split_takes_the_lower_middle_sample_as_locust_itself_does(self) -> None:
|
||||
entry = _entry(num_requests=4, response_times={10: 2, 90: 2})
|
||||
|
||||
assert median_seconds([entry]) == 0.01
|
||||
assert percentile_seconds([entry], 0.5) == 0.01
|
||||
|
||||
def test_no_samples_reports_zero_rather_than_dividing_by_an_empty_histogram(self) -> None:
|
||||
assert median_seconds([]) == 0.0
|
||||
assert percentile_seconds([], 0.5) == 0.0
|
||||
|
||||
|
||||
class TestAggregate:
|
||||
|
|
@ -84,9 +96,20 @@ class TestAggregate:
|
|||
result = aggregate_stats([entry], (), ())
|
||||
|
||||
assert result.requests_per_second == 3.0
|
||||
assert result.median_response_seconds == 0.057
|
||||
assert result.p50_seconds == 0.057
|
||||
assert result.p99_seconds == 0.057
|
||||
assert result.failure_ratio == 0.0
|
||||
|
||||
def test_tail_percentiles_come_from_the_slow_end_of_the_histogram(self) -> None:
|
||||
entry = _entry(num_requests=100, response_times={20: 89, 500: 10, 3000: 1})
|
||||
|
||||
result = aggregate_stats([entry], (), ())
|
||||
|
||||
assert result.p50_seconds == 0.02
|
||||
assert result.p90_seconds == 0.5
|
||||
assert result.p99_seconds == 0.5
|
||||
assert result.latency_summary() == "p50 0.020s, p90 0.500s, p99 0.500s"
|
||||
|
||||
def test_throughput_spans_from_the_earliest_start_when_locust_reports_several_entries(self) -> None:
|
||||
entries = [
|
||||
_entry(num_requests=60, start_time=1000.0, last_request_timestamp=1030.0),
|
||||
|
|
@ -133,8 +156,7 @@ class TestErrorBreakdown:
|
|||
def test_diagnosis_caps_the_list_and_says_how_many_it_left_out(self) -> None:
|
||||
result = _result(
|
||||
errors=tuple(
|
||||
LoadError(name="/chat/completions", error=f"error-{index}", occurrences=index)
|
||||
for index in range(1, 9)
|
||||
LoadError(name="/chat/completions", error=f"error-{index}", occurrences=index) for index in range(1, 9)
|
||||
)
|
||||
)
|
||||
|
||||
|
|
|
|||
59
tests/e2e/load/test_proxy_usage.py
Normal file
59
tests/e2e/load/test_proxy_usage.py
Normal file
|
|
@ -0,0 +1,59 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from typing import Final
|
||||
|
||||
from proxy_usage import UsageSample, UsageWindow
|
||||
|
||||
_MB: Final = 2**20
|
||||
|
||||
|
||||
def _window(*points: tuple[float, int, float]) -> UsageWindow:
|
||||
return UsageWindow(
|
||||
samples=tuple(
|
||||
UsageSample(elapsed_seconds=elapsed, rss_bytes=rss, cpu_seconds=cpu) for elapsed, rss, cpu in points
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
class TestRssPercentiles:
|
||||
def test_the_tail_percentiles_reach_the_peak_the_median_hides(self) -> None:
|
||||
# 100 one-second samples: 89 flat, 10 elevated, 1 spike. The median stays flat, p90 sees the
|
||||
# elevated plateau, and only the max reaches the spike.
|
||||
window: Final = _window(
|
||||
*((float(i), 100 * _MB, float(i)) for i in range(89)),
|
||||
*((float(89 + i), 300 * _MB, float(89 + i)) for i in range(10)),
|
||||
(99.0, 900 * _MB, 99.0),
|
||||
)
|
||||
|
||||
assert window.rss_percentile(0.5) == 100 * _MB
|
||||
assert window.rss_percentile(0.9) == 300 * _MB
|
||||
assert window.rss_percentile(0.99) == 300 * _MB
|
||||
assert window.rss_percentile(1.0) == 900 * _MB
|
||||
|
||||
def test_an_empty_window_reports_zero_rather_than_indexing_nothing(self) -> None:
|
||||
assert _window().rss_percentile(0.5) == 0
|
||||
|
||||
|
||||
class TestCpuUtilization:
|
||||
def test_utilization_is_the_counter_delta_over_the_interval_not_the_counter_itself(self) -> None:
|
||||
# The counter climbs 0.5 CPU seconds per second, then 4.0 per second: half a core, then four.
|
||||
window: Final = _window((0.0, _MB, 0.0), (1.0, _MB, 0.5), (2.0, _MB, 1.0), (3.0, _MB, 5.0))
|
||||
|
||||
p50, p90, p99 = window.cpu_utilization_percentiles()
|
||||
|
||||
assert (p50, p90, p99) == (0.5, 4.0, 4.0)
|
||||
assert window.cpu_seconds_consumed() == 5.0
|
||||
|
||||
def test_a_single_sample_has_no_interval_and_reports_zero(self) -> None:
|
||||
window: Final = _window((0.0, _MB, 3.0))
|
||||
|
||||
assert window.cpu_utilization_percentiles() == (0.0, 0.0, 0.0)
|
||||
assert window.cpu_seconds_consumed() == 0.0
|
||||
|
||||
def test_summary_reports_every_percentile_in_human_units(self) -> None:
|
||||
window: Final = _window((0.0, 200 * _MB, 0.0), (1.0, 200 * _MB, 1.5), (2.0, 200 * _MB, 3.0))
|
||||
|
||||
assert window.summary() == (
|
||||
"RSS p50 200 MB, p90 200 MB, p99 200 MB; "
|
||||
"CPU cores busy p50 1.50, p90 1.50, p99 1.50; 3.0 CPU seconds consumed"
|
||||
)
|
||||
258
tests/e2e/load/test_redis_chaos_e2e.py
Normal file
258
tests/e2e/load/test_redis_chaos_e2e.py
Normal file
|
|
@ -0,0 +1,258 @@
|
|||
"""Live e2e: the proxy under load keeps serving every request while Redis writes time out.
|
||||
|
||||
Runs against a proxy booted from tests/e2e/gateway/redis_chaos_ci_config.yml, which points
|
||||
cache_params at a real Redis with litellm's default socket_timeout. That one client backs all
|
||||
three Redis touchpoints on the request path: the virtual-key auth cache, the response cache,
|
||||
and the cross-pod spend counter the cost-tracking callback awaits.
|
||||
|
||||
The load runs in two phases against one model group of three mock deployments. The two at
|
||||
order 1 raise InternalServerError and the one at order 2 serves, so every request burns its
|
||||
retries on the failing pair (a 500 is retryable, so retries keep re-picking inside the lowest
|
||||
order) and the router's order-based fallback then re-targets order 2. Every request is expected
|
||||
to succeed, and each one carries retry breadcrumbs into cost tracking.
|
||||
|
||||
Phase A is a baseline with Redis healthy; phase B holds Redis in CLIENT PAUSE WRITE, so the
|
||||
spend counter increment times out and the callback stringifies the request metadata,
|
||||
breadcrumbs included, into a failed-tracking alert. On v1.100.0 that string doubled per request
|
||||
until the worker hung (LIT-6780), which is what the per-phase RSS and CPU percentiles are here
|
||||
to catch.
|
||||
|
||||
Needs the proxy on the same host, since RSS and CPU come from psutil on its process tree:
|
||||
a multi-worker proxy serves /metrics from the prometheus multiprocess collector, which drops
|
||||
the process collector's memory and CPU series. Deselected unless E2E_REDIS_CHAOS is set.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
from collections.abc import Iterator
|
||||
from dataclasses import dataclass
|
||||
from typing import Final
|
||||
|
||||
import pytest
|
||||
import redis
|
||||
from e2e_config import PROXY_BASE_URL, unique_marker
|
||||
from e2e_http import NoBody
|
||||
from lifecycle import ResourceManager
|
||||
from load_client import LoadClient
|
||||
from locust_load import LoadResult, run_chat_load
|
||||
from models import KeyGenerateBody, LiteLLMParamsBody
|
||||
from proxy_client import ProxyClient
|
||||
from proxy_usage import ProxyUsageSampler, UsageWindow
|
||||
|
||||
pytestmark = [pytest.mark.e2e, pytest.mark.redis_chaos]
|
||||
|
||||
MODEL_GROUP: Final = "redis-chaos-fable"
|
||||
MOCK_MODEL: Final = "anthropic/claude-fable-5-1"
|
||||
FAILING_DEPLOYMENTS: Final = 2
|
||||
SERVING_DEPLOYMENTS: Final = 1
|
||||
FAILING_ORDER: Final = 1
|
||||
SERVING_ORDER: Final = 2
|
||||
KEY_POOL_SIZE: Final = 8
|
||||
LOCUST_USERS: Final = 50
|
||||
LOCUST_SPAWN_RATE: Final = 50.0
|
||||
BASELINE_SECONDS: Final = 60.0
|
||||
CHAOS_SECONDS: Final = 90.0
|
||||
REDIS_PAUSE_MS: Final = 600_000
|
||||
BASELINE_TIMEOUT_RATE_CEILING: Final = 0.05
|
||||
CHAOS_TIMEOUT_RATE_FLOOR: Final = 0.20
|
||||
|
||||
TIMEOUT_FAILURES_RE: Final = re.compile(
|
||||
r'^litellm_redis_circuit_breaker_failures_total\{failure_class="timeout"\} ([0-9.e+]+)$', re.M
|
||||
)
|
||||
BREAKER_OPEN_RE: Final = re.compile(r'^litellm_redis_circuit_breaker_state\{state="open"\} ([0-9.e+]+)$', re.M)
|
||||
BREAKER_TRANSITIONS_RE: Final = re.compile(
|
||||
r'^litellm_redis_circuit_breaker_transitions_total\{state="[a-z_]+"\} ([0-9.e+]+)$', re.M
|
||||
)
|
||||
RETRIES_RE: Final = re.compile(r"^litellm_deployment_failure_responses_total\{[^}]*\} ([0-9.e+]+)$", re.M)
|
||||
COOLDOWN_RE: Final = re.compile(r"^litellm_deployment_cooled_down_total\{[^}]*\} ([0-9.e+]+)$", re.M)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Phase:
|
||||
"""One load phase's traffic and what the proxy's process tree did during it."""
|
||||
|
||||
name: str
|
||||
load: LoadResult
|
||||
usage: UsageWindow
|
||||
|
||||
def report(self) -> str:
|
||||
return (
|
||||
f"{self.name}: {self.load.requests} requests, {self.load.failures} failures, "
|
||||
f"{self.load.requests_per_second:.0f} rps, {self.load.latency_summary()}; {self.usage.summary()}"
|
||||
)
|
||||
|
||||
|
||||
def _failing_params() -> LiteLLMParamsBody:
|
||||
return LiteLLMParamsBody(
|
||||
model=MOCK_MODEL,
|
||||
api_key="sk-redis-chaos-not-used",
|
||||
mock_response="litellm.InternalServerError",
|
||||
order=FAILING_ORDER,
|
||||
)
|
||||
|
||||
|
||||
def _serving_params() -> LiteLLMParamsBody:
|
||||
return LiteLLMParamsBody(
|
||||
model=MOCK_MODEL,
|
||||
api_key="sk-redis-chaos-not-used",
|
||||
mock_response="redis chaos ok",
|
||||
order=SERVING_ORDER,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def proxy_pid() -> int:
|
||||
"""The proxy's PID, which the workflow exports after starting it.
|
||||
|
||||
Required rather than discovered: picking a process out of the table by name would be
|
||||
ambiguous on a developer machine running more than one proxy.
|
||||
"""
|
||||
pid: Final = os.environ.get("E2E_PROXY_PID")
|
||||
assert pid and pid.isdigit(), (
|
||||
"E2E_PROXY_PID must hold the PID of the proxy under test; RSS and CPU are read from "
|
||||
"its process tree because a multi-worker proxy does not report them on /metrics"
|
||||
)
|
||||
return int(pid)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def redis_control() -> Iterator[redis.Redis[bytes]]:
|
||||
"""A control connection to the proxy's Redis, which unpauses writes in teardown.
|
||||
|
||||
Only writes are paused: CLIENT PAUSE ALL would freeze this connection too, leaving
|
||||
nothing able to lift the pause.
|
||||
"""
|
||||
host: Final = os.environ.get("REDIS_HOST")
|
||||
port: Final = os.environ.get("REDIS_PORT")
|
||||
assert host and port, "REDIS_HOST and REDIS_PORT must name the Redis the proxy under test uses"
|
||||
control: Final = redis.Redis(host=host, port=int(port), socket_timeout=5)
|
||||
try:
|
||||
yield control
|
||||
finally:
|
||||
control.client_unpause() # pyright: ignore[reportUnknownMemberType] # redis-py stubs return Any
|
||||
control.close()
|
||||
|
||||
|
||||
def _metric(proxy: ProxyClient, pattern: re.Pattern[str]) -> float:
|
||||
body: Final = proxy.probe("/metrics", params=NoBody()).body
|
||||
return sum(float(match.group(1)) for match in pattern.finditer(body))
|
||||
|
||||
|
||||
def _register_deployments(proxy: ProxyClient, resources: ResourceManager) -> None:
|
||||
for _ in range(FAILING_DEPLOYMENTS):
|
||||
failing_id = proxy.create_model(MODEL_GROUP, _failing_params())
|
||||
resources.defer(lambda model_id=failing_id: proxy.delete_model(model_id))
|
||||
for _ in range(SERVING_DEPLOYMENTS):
|
||||
serving_id = proxy.create_model(MODEL_GROUP, _serving_params())
|
||||
resources.defer(lambda model_id=serving_id: proxy.delete_model(model_id))
|
||||
|
||||
|
||||
def _generate_key_pool(proxy: ProxyClient, resources: ResourceManager) -> tuple[str, ...]:
|
||||
"""A pool of virtual keys so auth and budget lookups are not one permanently warm
|
||||
cache entry; each locust user picks one, so Redis auth reads actually happen."""
|
||||
keys: Final = tuple(
|
||||
proxy.generate_key(
|
||||
KeyGenerateBody(models=[MODEL_GROUP], key_alias=f"e2e-redis-chaos-{unique_marker()}-{index}")
|
||||
)
|
||||
for index in range(KEY_POOL_SIZE)
|
||||
)
|
||||
for key in keys:
|
||||
resources.defer(lambda doomed=key: proxy.delete_key(doomed))
|
||||
return keys
|
||||
|
||||
|
||||
def _drive(keys: tuple[str, ...], seconds: float) -> LoadResult:
|
||||
return run_chat_load(
|
||||
base_url=PROXY_BASE_URL,
|
||||
api_keys=keys,
|
||||
model=MODEL_GROUP,
|
||||
users=LOCUST_USERS,
|
||||
spawn_rate=LOCUST_SPAWN_RATE,
|
||||
duration_seconds=seconds,
|
||||
)
|
||||
|
||||
|
||||
class TestRedisChaos:
|
||||
@pytest.mark.covers(
|
||||
"reliability.circuit_breaker.redis_timeout.stays_responsive",
|
||||
exercised_on=["chat_completions"],
|
||||
)
|
||||
def test_load_survives_redis_write_timeouts(
|
||||
self,
|
||||
client: LoadClient,
|
||||
resources: ResourceManager,
|
||||
proxy_pid: int,
|
||||
redis_control: redis.Redis[bytes],
|
||||
) -> None:
|
||||
proxy: Final = client.proxy
|
||||
_register_deployments(proxy, resources)
|
||||
keys: Final = _generate_key_pool(proxy, resources)
|
||||
|
||||
timeouts_at_start: Final = _metric(proxy, TIMEOUT_FAILURES_RE)
|
||||
retries_before: Final = _metric(proxy, RETRIES_RE)
|
||||
cooldowns_before: Final = _metric(proxy, COOLDOWN_RE)
|
||||
|
||||
with ProxyUsageSampler(proxy_pid) as sampler:
|
||||
baseline: Final = Phase(name="baseline", load=_drive(keys, BASELINE_SECONDS), usage=sampler.split())
|
||||
timeouts_after_baseline: Final = _metric(proxy, TIMEOUT_FAILURES_RE)
|
||||
|
||||
redis_control.client_pause(REDIS_PAUSE_MS, all=False) # pyright: ignore[reportUnknownMemberType] # redis-py stubs return Any
|
||||
chaos: Final = Phase(name="chaos", load=_drive(keys, CHAOS_SECONDS), usage=sampler.split())
|
||||
|
||||
report: Final = f"{baseline.report()} | {chaos.report()}"
|
||||
|
||||
for phase in (baseline, chaos):
|
||||
assert phase.load.requests > 0, (
|
||||
f"{phase.name} drove no traffic at all, so it proved nothing: {phase.load.diagnosis()}. {report}"
|
||||
)
|
||||
assert phase.load.failures == 0, (
|
||||
f"{phase.name} had {phase.load.failures} of {phase.load.requests} requests fail. Every request "
|
||||
f"must succeed: the failing deployments sit at order {FAILING_ORDER} and the serving one at order "
|
||||
f"{SERVING_ORDER}, so once the retries on order {FAILING_ORDER} are spent the order-based fallback "
|
||||
f"lands on the serving deployment. Failures mean it was cooled down, the fallback did not run, or "
|
||||
f"a Redis failure reached the response path. {phase.load.diagnosis()}. {report}"
|
||||
)
|
||||
|
||||
cooldowns: Final = _metric(proxy, COOLDOWN_RE) - cooldowns_before
|
||||
assert cooldowns == 0, (
|
||||
f"{cooldowns:.0f} deployments were cooled down during the run; the failing deployments are supposed "
|
||||
f"to stay in rotation so every request keeps exercising the retry path. {report}"
|
||||
)
|
||||
|
||||
retries: Final = _metric(proxy, RETRIES_RE) - retries_before
|
||||
assert retries >= baseline.load.requests + chaos.load.requests, (
|
||||
f"only {retries:.0f} deployment failures were counted across "
|
||||
f"{baseline.load.requests + chaos.load.requests} requests; the mock deployments did not fail, so no "
|
||||
f"request carried retry breadcrumbs into cost tracking and the regression path was never entered. "
|
||||
f"{report}"
|
||||
)
|
||||
|
||||
baseline_timeout_rate: Final = (timeouts_after_baseline - timeouts_at_start) / baseline.load.requests
|
||||
assert baseline_timeout_rate <= BASELINE_TIMEOUT_RATE_CEILING, (
|
||||
f"a healthy Redis timed out on {baseline_timeout_rate:.1%} of baseline requests, over the "
|
||||
f"{BASELINE_TIMEOUT_RATE_CEILING:.0%} this test tolerates; at litellm's default socket_timeout a "
|
||||
f"loaded Redis does time out occasionally, but this much means the baseline is already degraded and "
|
||||
f"the two phases are not comparable. {report}"
|
||||
)
|
||||
|
||||
chaos_timeouts: Final = _metric(proxy, TIMEOUT_FAILURES_RE) - timeouts_after_baseline
|
||||
chaos_timeout_rate: Final = chaos_timeouts / chaos.load.requests
|
||||
transitions: Final = _metric(proxy, BREAKER_TRANSITIONS_RE)
|
||||
breaker_open: Final = _metric(proxy, BREAKER_OPEN_RE) >= 1
|
||||
assert chaos_timeout_rate >= CHAOS_TIMEOUT_RATE_FLOOR or transitions >= 1 or breaker_open, (
|
||||
f"with writes paused the breaker saw Redis time out on only {chaos_timeout_rate:.1%} of requests "
|
||||
f"against {baseline_timeout_rate:.1%} at baseline, under the {CHAOS_TIMEOUT_RATE_FLOOR:.0%} a real "
|
||||
f"outage produces, and it counted {transitions:.0f} state transitions and ended "
|
||||
f"{'open' if breaker_open else 'closed'}. The spend counter increment never failed, so this run "
|
||||
f"proved nothing. {report}"
|
||||
)
|
||||
|
||||
rows: Final = proxy.poll_logs_for_key(keys[0], min_rows=1)
|
||||
assert rows, (
|
||||
f"no spend rows landed for the first key in the pool; a Redis outage must not cost the proxy its "
|
||||
f"spend logs, which are written to Postgres through a queue rather than through Redis. {report}"
|
||||
)
|
||||
|
||||
print(f"\nredis chaos load: {report}") # noqa: T201 # the numbers this test exists to report, read off the CI log
|
||||
|
|
@ -924,6 +924,7 @@ class LiteLLMParamsBody(BaseModel):
|
|||
timeout: float | None = None
|
||||
tpm: int | None = None
|
||||
weight: int | None = None
|
||||
order: int | None = None
|
||||
|
||||
|
||||
ModelMode = Literal["batch", "realtime", "image_generation"]
|
||||
|
|
|
|||
|
|
@ -9,4 +9,4 @@ markers =
|
|||
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
|
||||
managed_files: needs a proxy running with require_managed_files enabled; deselected unless E2E_MANAGED_FILES_STACK is set
|
||||
redis_timeout: needs a proxy booted from gateway/redis_timeout_ci_config.yml whose Redis times out on every command; deselected unless E2E_REDIS_TIMEOUT is set
|
||||
redis_chaos: load test that pauses the proxy's Redis writes mid-run; needs a proxy booted from gateway/redis_chaos_ci_config.yml on the same host, and is deselected unless E2E_REDIS_CHAOS is set
|
||||
|
|
|
|||
|
|
@ -10,12 +10,10 @@ proxy does not already list it (compose has it in static config; stage does not)
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from collections.abc import Iterator
|
||||
|
||||
import pytest
|
||||
from complexity_router_client import ComplexityRouterClient, build_client
|
||||
from e2e_config import REDIS_TIMEOUT_OPT_IN_ENV
|
||||
from e2e_http import NoBody, Success
|
||||
from lifecycle import ResourceManager
|
||||
from models import (
|
||||
|
|
@ -46,16 +44,6 @@ ROUTER_PARAMS = LiteLLMParamsBody(
|
|||
ROUTER_KEY_MODELS = [ROUTER_MODEL, "gpt-5.5", "claude-haiku-4-5"]
|
||||
|
||||
|
||||
def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item]) -> None:
|
||||
if os.environ.get(REDIS_TIMEOUT_OPT_IN_ENV):
|
||||
return
|
||||
deselected = [item for item in items if item.get_closest_marker("redis_timeout") is not None]
|
||||
if not deselected:
|
||||
return
|
||||
config.hook.pytest_deselected(items=deselected)
|
||||
items[:] = [item for item in items if item.get_closest_marker("redis_timeout") is None]
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def client(proxy: ProxyClient) -> ComplexityRouterClient:
|
||||
return build_client(proxy)
|
||||
|
|
|
|||
|
|
@ -1,282 +0,0 @@
|
|||
"""Live e2e: the proxy keeps answering while every Redis command times out.
|
||||
|
||||
Runs only against a proxy booted from tests/e2e/gateway/redis_timeout_ci_config.yml, which
|
||||
points cache_params at a real Redis with socket_timeout 0.001. The test holds that Redis in
|
||||
CLIENT PAUSE WRITE for its duration, so every write the proxy sends, the spend counter increment
|
||||
included, hangs past the timeout. For each endpoint it registers two deployments through
|
||||
/model/new: a primary whose api_base is a closed port and a backup that answers with a mock. Each request fails the primary,
|
||||
retries, falls back and succeeds, so it carries retry breadcrumbs; its cost tracking then fails on
|
||||
the spend counter increment and stringifies the request metadata into a failed-tracking alert. On
|
||||
v1.100.0 that string doubled per request until the worker hung (LIT-6780). Deselected unless
|
||||
E2E_REDIS_TIMEOUT is set, since it needs that dedicated proxy.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
import time
|
||||
from collections.abc import Callable, Iterator
|
||||
from dataclasses import dataclass
|
||||
from typing import Final
|
||||
|
||||
import pytest
|
||||
import redis
|
||||
from complexity_router_client import ComplexityRouterClient
|
||||
from e2e_config import unique_marker
|
||||
from e2e_http import NoBody, Result, Success
|
||||
from lifecycle import ResourceManager
|
||||
from models import ChatBody, ChatMessage, ChatResponse, EmbedBody, KeyGenerateBody, LiteLLMParamsBody
|
||||
from proxy_client import ProxyClient
|
||||
from pydantic import BaseModel
|
||||
|
||||
pytestmark = [pytest.mark.e2e, pytest.mark.redis_timeout]
|
||||
|
||||
REQUESTS: Final = 20
|
||||
MAX_SECONDS_PER_REQUEST: Final = 10.0
|
||||
MAX_LATENCY_GROWTH_RATIO: Final = 3.0
|
||||
MAX_LIVELINESS_SECONDS: Final = 2.0
|
||||
MAX_RSS_GROWTH_BYTES: Final = 200 * 1024 * 1024
|
||||
REDIS_PAUSE_MS: Final = 600_000
|
||||
BREAKER_FAILURE_THRESHOLD: Final = 5
|
||||
CLOSED_PORT_API_BASE: Final = "http://127.0.0.1:1"
|
||||
TIMEOUT_FAILURES_RE: Final = re.compile(
|
||||
r'^litellm_redis_circuit_breaker_failures_total\{failure_class="timeout"\} ([0-9.e+]+)$', re.M
|
||||
)
|
||||
BREAKER_OPEN_RE: Final = re.compile(r'^litellm_redis_circuit_breaker_state\{state="open"\} ([0-9.e+]+)$', re.M)
|
||||
BREAKER_TRANSITIONS_RE: Final = re.compile(
|
||||
r'^litellm_redis_circuit_breaker_transitions_total\{state="[a-z_]+"\} ([0-9.e+]+)$', re.M
|
||||
)
|
||||
FALLBACKS_RE: Final = re.compile(r"^litellm_deployment_successful_fallbacks_total\{[^}]*\} ([0-9.e+]+)$", re.M)
|
||||
RSS_RE: Final = re.compile(r"^process_resident_memory_bytes ([0-9.e+]+)$", re.M)
|
||||
|
||||
|
||||
class ResponsesBody(BaseModel):
|
||||
model: str
|
||||
input: str
|
||||
max_output_tokens: int = 5
|
||||
|
||||
|
||||
class ResponsesObject(BaseModel):
|
||||
id: str | None = None
|
||||
status: str | None = None
|
||||
output: list[object] = []
|
||||
|
||||
|
||||
class EmbeddingsObject(BaseModel):
|
||||
model: str | None = None
|
||||
data: list[object] = []
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Endpoint:
|
||||
"""One endpoint's deployments and request shape."""
|
||||
|
||||
name: str
|
||||
primary: str
|
||||
backup: str
|
||||
primary_params: LiteLLMParamsBody
|
||||
backup_params: LiteLLMParamsBody
|
||||
send: Callable[[ProxyClient, str, str, str], Result[BaseModel]]
|
||||
served: Callable[[BaseModel], bool]
|
||||
|
||||
|
||||
def _send_chat(proxy: ProxyClient, key: str, model: str, marker: str) -> Result[BaseModel]:
|
||||
return proxy.transport.post(
|
||||
"/chat/completions",
|
||||
headers=proxy.transport.bearer(key),
|
||||
json=ChatBody(model=model, messages=[ChatMessage(role="user", content=marker)], max_tokens=5),
|
||||
response_type=ChatResponse,
|
||||
timeout=MAX_SECONDS_PER_REQUEST,
|
||||
)
|
||||
|
||||
|
||||
def _send_responses(proxy: ProxyClient, key: str, model: str, marker: str) -> Result[BaseModel]:
|
||||
return proxy.transport.post(
|
||||
"/v1/responses",
|
||||
headers=proxy.transport.bearer(key),
|
||||
json=ResponsesBody(model=model, input=marker),
|
||||
response_type=ResponsesObject,
|
||||
timeout=MAX_SECONDS_PER_REQUEST,
|
||||
)
|
||||
|
||||
|
||||
def _send_embeddings(proxy: ProxyClient, key: str, model: str, marker: str) -> Result[BaseModel]:
|
||||
return proxy.transport.post(
|
||||
"/embeddings",
|
||||
headers=proxy.transport.bearer(key),
|
||||
json=EmbedBody(model=model, input=marker),
|
||||
response_type=EmbeddingsObject,
|
||||
timeout=MAX_SECONDS_PER_REQUEST,
|
||||
)
|
||||
|
||||
|
||||
_CHAT_PRIMARY: Final = LiteLLMParamsBody(
|
||||
model="openai/gpt-5-mini", api_key="sk-redis-timeout-primary-not-used", api_base=CLOSED_PORT_API_BASE
|
||||
)
|
||||
_CHAT_BACKUP: Final = LiteLLMParamsBody(
|
||||
model="openai/gpt-5-nano", api_key="sk-redis-timeout-backup-not-used", mock_response="ok"
|
||||
)
|
||||
_EMBED_PRIMARY: Final = LiteLLMParamsBody(
|
||||
model="openai/text-embedding-3-small", api_key="sk-redis-timeout-primary-not-used", api_base=CLOSED_PORT_API_BASE
|
||||
)
|
||||
_EMBED_BACKUP: Final = LiteLLMParamsBody(
|
||||
model="openai/text-embedding-3-large", api_key="sk-redis-timeout-backup-not-used", mock_response=[0.1, 0.2, 0.3]
|
||||
)
|
||||
|
||||
ENDPOINTS: Final = (
|
||||
Endpoint(
|
||||
name="chat_completions",
|
||||
primary="redis-timeout-primary",
|
||||
backup="redis-timeout-backup",
|
||||
primary_params=_CHAT_PRIMARY,
|
||||
backup_params=_CHAT_BACKUP,
|
||||
send=_send_chat,
|
||||
served=lambda data: isinstance(data, ChatResponse) and bool(data.choices),
|
||||
),
|
||||
Endpoint(
|
||||
name="responses",
|
||||
primary="redis-timeout-primary",
|
||||
backup="redis-timeout-backup",
|
||||
primary_params=_CHAT_PRIMARY,
|
||||
backup_params=_CHAT_BACKUP,
|
||||
send=_send_responses,
|
||||
served=lambda data: isinstance(data, ResponsesObject) and bool(data.output),
|
||||
),
|
||||
Endpoint(
|
||||
name="embeddings",
|
||||
primary="redis-timeout-embed-primary",
|
||||
backup="redis-timeout-embed-backup",
|
||||
primary_params=_EMBED_PRIMARY,
|
||||
backup_params=_EMBED_BACKUP,
|
||||
send=_send_embeddings,
|
||||
served=lambda data: isinstance(data, EmbeddingsObject) and bool(data.data),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def paused_redis() -> Iterator[None]:
|
||||
"""Hold the proxy's Redis in CLIENT PAUSE WRITE so every write it sends outlives the 1 ms socket
|
||||
timeout. A loopback Redis otherwise answers many commands inside that budget. Reads stay live so
|
||||
this control connection can lift the pause in teardown."""
|
||||
host = os.environ.get("REDIS_HOST")
|
||||
port = os.environ.get("REDIS_PORT")
|
||||
assert host and port, "REDIS_HOST and REDIS_PORT must name the Redis the proxy under test uses"
|
||||
control = redis.Redis(host=host, port=int(port), socket_timeout=5)
|
||||
control.client_pause(REDIS_PAUSE_MS, all=False) # pyright: ignore[reportUnknownMemberType] # redis-py stubs return Any
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
control.client_unpause() # pyright: ignore[reportUnknownMemberType] # redis-py stubs return Any
|
||||
control.close()
|
||||
|
||||
|
||||
def _metric(proxy: ProxyClient, pattern: re.Pattern[str]) -> float:
|
||||
body = proxy.probe("/metrics", params=NoBody()).body
|
||||
return sum(float(match.group(1)) for match in pattern.finditer(body))
|
||||
|
||||
|
||||
def _rss_bytes(proxy: ProxyClient) -> float | None:
|
||||
"""The proxy's resident memory from the Prometheus process collector, which reads /proc and
|
||||
so reports on Linux only; None where the metric is absent."""
|
||||
body = proxy.probe("/metrics", params=NoBody()).body
|
||||
match = RSS_RE.search(body)
|
||||
return float(match.group(1)) if match else None
|
||||
|
||||
|
||||
class TestRedisTimeout:
|
||||
@pytest.mark.parametrize("endpoint", ENDPOINTS, ids=[endpoint.name for endpoint in ENDPOINTS])
|
||||
@pytest.mark.covers(
|
||||
"reliability.circuit_breaker.redis_timeout.stays_responsive",
|
||||
exercised_on=["chat_completions", "responses", "embeddings"],
|
||||
)
|
||||
def test_retries_under_redis_timeouts_keep_answering(
|
||||
self, client: ComplexityRouterClient, resources: ResourceManager, endpoint: Endpoint, paused_redis: None
|
||||
) -> None:
|
||||
proxy = client.proxy
|
||||
primary_id = proxy.create_model(endpoint.primary, endpoint.primary_params)
|
||||
resources.defer(lambda: proxy.delete_model(primary_id))
|
||||
backup_id = proxy.create_model(endpoint.backup, endpoint.backup_params)
|
||||
resources.defer(lambda: proxy.delete_model(backup_id))
|
||||
key = proxy.generate_key(
|
||||
KeyGenerateBody(
|
||||
models=[endpoint.primary, endpoint.backup],
|
||||
key_alias=f"e2e-redis-timeout-{endpoint.name}-{unique_marker()}",
|
||||
)
|
||||
)
|
||||
resources.defer(lambda: proxy.delete_key(key))
|
||||
timeouts_before = _metric(proxy, TIMEOUT_FAILURES_RE)
|
||||
transitions_before = _metric(proxy, BREAKER_TRANSITIONS_RE)
|
||||
fallbacks_before = _metric(proxy, FALLBACKS_RE)
|
||||
rss_before = _rss_bytes(proxy)
|
||||
|
||||
latencies: list[float] = []
|
||||
for request_number in range(1, REQUESTS + 1):
|
||||
started = time.monotonic()
|
||||
result = endpoint.send(proxy, key, endpoint.primary, f"redis timeout {unique_marker()} {request_number}")
|
||||
elapsed = time.monotonic() - started
|
||||
assert isinstance(result, Success), (
|
||||
f"{endpoint.name} request {request_number} failed after {elapsed:.1f}s with Redis timing out: {result}; "
|
||||
f"earlier requests took {[round(seconds, 2) for seconds in latencies]}"
|
||||
)
|
||||
assert endpoint.served(result.data), (
|
||||
f"{endpoint.name} request {request_number}: fallback to {endpoint.backup} returned no output"
|
||||
)
|
||||
assert elapsed < MAX_SECONDS_PER_REQUEST, (
|
||||
f"{endpoint.name} request {request_number} took {elapsed:.1f}s with Redis timing out; "
|
||||
f"earlier requests took {[round(seconds, 2) for seconds in latencies]}"
|
||||
)
|
||||
latencies.append(elapsed)
|
||||
|
||||
third = REQUESTS // 3
|
||||
early = sum(latencies[:third]) / third
|
||||
late = sum(latencies[-third:]) / third
|
||||
assert late <= max(early * MAX_LATENCY_GROWTH_RATIO, 0.5), (
|
||||
f"{endpoint.name} per-request latency grew from {early:.2f}s to {late:.2f}s across {REQUESTS} requests "
|
||||
"while Redis timed out; the proxy is paying more for each failed request"
|
||||
)
|
||||
|
||||
started = time.monotonic()
|
||||
probe = proxy.transport.probe("/health/liveliness", params=NoBody())
|
||||
liveliness_seconds = time.monotonic() - started
|
||||
assert probe.healthy, f"/health/liveliness returned {probe.status_code} after the Redis timeout loop"
|
||||
assert liveliness_seconds < MAX_LIVELINESS_SECONDS, (
|
||||
f"/health/liveliness took {liveliness_seconds:.1f}s after the loop; the worker is stalled"
|
||||
)
|
||||
|
||||
if rss_before is not None:
|
||||
rss_after = _rss_bytes(proxy)
|
||||
assert rss_after is not None
|
||||
assert rss_after - rss_before <= MAX_RSS_GROWTH_BYTES, (
|
||||
f"proxy RSS grew {(rss_after - rss_before) / 2**20:.0f} MB across {REQUESTS} {endpoint.name} requests "
|
||||
"with Redis timing out; on v1.100.0 this path grew by gigabytes"
|
||||
)
|
||||
|
||||
fallbacks = _metric(proxy, FALLBACKS_RE) - fallbacks_before
|
||||
assert fallbacks >= REQUESTS, (
|
||||
f"only {fallbacks:.0f} successful fallbacks were counted across {REQUESTS} {endpoint.name} requests; "
|
||||
"the closed-port primary did not fail every request, so the retry path was not exercised"
|
||||
)
|
||||
|
||||
timeouts_total = _metric(proxy, TIMEOUT_FAILURES_RE)
|
||||
timeouts = timeouts_total - timeouts_before
|
||||
transitions = _metric(proxy, BREAKER_TRANSITIONS_RE) - transitions_before
|
||||
breaker_open = _metric(proxy, BREAKER_OPEN_RE) >= 1
|
||||
assert timeouts_total >= BREAKER_FAILURE_THRESHOLD, (
|
||||
f"the proxy counted only {timeouts_total:.0f} Redis timeouts in its lifetime; the write-paused Redis "
|
||||
"never made its spend counter writes time out, so this run proved nothing"
|
||||
)
|
||||
assert timeouts >= REQUESTS or transitions >= 1 or breaker_open, (
|
||||
f"during {REQUESTS} {endpoint.name} requests the breaker counted {timeouts:.0f} new timeouts, "
|
||||
f"{transitions:.0f} state transitions, and ended {'open' if breaker_open else 'closed'}; "
|
||||
"Redis was healthy for this case, so it proved nothing"
|
||||
)
|
||||
|
||||
rows = proxy.poll_logs_for_key(key, min_rows=REQUESTS)
|
||||
assert len(rows) >= REQUESTS, (
|
||||
f"only {len(rows)} of {REQUESTS} {endpoint.name} requests reached the spend log; "
|
||||
"a Redis outage must not lose spend rows"
|
||||
)
|
||||
failed_rows = [row.status for row in rows if row.status not in (None, "success")]
|
||||
assert not failed_rows, f"{len(failed_rows)} {endpoint.name} spend rows are not successes: {failed_rows[:3]}"
|
||||
4
uv.lock
generated
4
uv.lock
generated
|
|
@ -10,7 +10,7 @@ resolution-markers = [
|
|||
]
|
||||
|
||||
[options]
|
||||
exclude-newer = "2026-09-06T00:40:30.433549Z"
|
||||
exclude-newer = "2026-09-08T03:56:24.358378Z"
|
||||
exclude-newer-span = "P3D"
|
||||
|
||||
[manifest]
|
||||
|
|
@ -4559,6 +4559,7 @@ e2e-dev = [
|
|||
{ name = "locust" },
|
||||
{ name = "mcp" },
|
||||
{ name = "playwright" },
|
||||
{ name = "psutil" },
|
||||
{ name = "websockets" },
|
||||
]
|
||||
healthcheck = [
|
||||
|
|
@ -4747,6 +4748,7 @@ e2e-dev = [
|
|||
{ name = "locust", specifier = "==2.45.0" },
|
||||
{ name = "mcp", specifier = ">=1.28.1,<2.0" },
|
||||
{ name = "playwright", specifier = "==1.61.0" },
|
||||
{ name = "psutil", specifier = "==7.2.2" },
|
||||
{ name = "websockets", specifier = ">=15.0.1,<16.0" },
|
||||
]
|
||||
healthcheck = [
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue