From 33ec56ed759a3d2876e821d970198cc30563f8d6 Mon Sep 17 00:00:00 2001 From: Kerry Lu Date: Thu, 10 Sep 2026 21:55:23 -0700 Subject: [PATCH] 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 --- .github/e2e-stack/select_tests.py | 1 - ...s-timeout.yml => test-e2e-redis-chaos.yml} | 21 +- pyproject.toml | 1 + tests/e2e/CLAUDE.md | 4 +- tests/e2e/CONTRIBUTING.md | 2 +- tests/e2e/conftest.py | 4 +- tests/e2e/coverage_registry/reliability.yaml | 2 +- tests/e2e/e2e_config.py | 2 +- ...i_config.yml => redis_chaos_ci_config.yml} | 11 +- tests/e2e/load/conftest.py | 22 +- tests/e2e/load/locust_load.py | 103 ++++++- tests/e2e/load/locustfile.py | 36 +++ tests/e2e/load/proxy_usage.py | 156 ++++++++++ tests/e2e/load/test_locust_load.py | 44 ++- tests/e2e/load/test_proxy_usage.py | 59 ++++ tests/e2e/load/test_redis_chaos_e2e.py | 258 ++++++++++++++++ tests/e2e/models.py | 1 + tests/e2e/pytest.ini | 2 +- tests/e2e/router/conftest.py | 12 - tests/e2e/router/test_redis_timeout_e2e.py | 282 ------------------ uv.lock | 4 +- 21 files changed, 673 insertions(+), 354 deletions(-) rename .github/workflows/{test-e2e-redis-timeout.yml => test-e2e-redis-chaos.yml} (82%) rename tests/e2e/gateway/{redis_timeout_ci_config.yml => redis_chaos_ci_config.yml} (59%) create mode 100644 tests/e2e/load/locustfile.py create mode 100644 tests/e2e/load/proxy_usage.py create mode 100644 tests/e2e/load/test_proxy_usage.py create mode 100644 tests/e2e/load/test_redis_chaos_e2e.py delete mode 100644 tests/e2e/router/test_redis_timeout_e2e.py diff --git a/.github/e2e-stack/select_tests.py b/.github/e2e-stack/select_tests.py index 10f26bd3b8a..a62358f81ff 100644 --- a/.github/e2e-stack/select_tests.py +++ b/.github/e2e-stack/select_tests.py @@ -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)$" diff --git a/.github/workflows/test-e2e-redis-timeout.yml b/.github/workflows/test-e2e-redis-chaos.yml similarity index 82% rename from .github/workflows/test-e2e-redis-timeout.yml rename to .github/workflows/test-e2e-redis-chaos.yml index 501af0f240f..2ce20836db7 100644 --- a/.github/workflows/test-e2e-redis-timeout.yml +++ b/.github/workflows/test-e2e-redis-chaos.yml @@ -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() diff --git a/pyproject.toml b/pyproject.toml index 04f2f3fd1dd..ae1698feb5f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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 = [ diff --git a/tests/e2e/CLAUDE.md b/tests/e2e/CLAUDE.md index 76e8550cb9a..dc197996ac2 100644 --- a/tests/e2e/CLAUDE.md +++ b/tests/e2e/CLAUDE.md @@ -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 diff --git a/tests/e2e/CONTRIBUTING.md b/tests/e2e/CONTRIBUTING.md index 7a859965c80..b1331de190a 100644 --- a/tests/e2e/CONTRIBUTING.md +++ b/tests/e2e/CONTRIBUTING.md @@ -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 diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py index ff3d2dcc6bd..99a3f4967c3 100644 --- a/tests/e2e/conftest.py +++ b/tests/e2e/conftest.py @@ -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", ) diff --git a/tests/e2e/coverage_registry/reliability.yaml b/tests/e2e/coverage_registry/reliability.yaml index a42e4b2779a..23163ab666c 100644 --- a/tests/e2e/coverage_registry/reliability.yaml +++ b/tests/e2e/coverage_registry/reliability.yaml @@ -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"} diff --git a/tests/e2e/e2e_config.py b/tests/e2e/e2e_config.py index 6cb59ea4e90..7344a2b6cec 100644 --- a/tests/e2e/e2e_config.py +++ b/tests/e2e/e2e_config.py @@ -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")) diff --git a/tests/e2e/gateway/redis_timeout_ci_config.yml b/tests/e2e/gateway/redis_chaos_ci_config.yml similarity index 59% rename from tests/e2e/gateway/redis_timeout_ci_config.yml rename to tests/e2e/gateway/redis_chaos_ci_config.yml index fd283dbc7df..37b024502f7 100644 --- a/tests/e2e/gateway/redis_timeout_ci_config.yml +++ b/tests/e2e/gateway/redis_chaos_ci_config.yml @@ -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 diff --git a/tests/e2e/load/conftest.py b/tests/e2e/load/conftest.py index 3a926ef2a61..e6f3c7aa538 100644 --- a/tests/e2e/load/conftest.py +++ b/tests/e2e/load/conftest.py @@ -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") diff --git a/tests/e2e/load/locust_load.py b/tests/e2e/load/locust_load.py index e0da8ba70c2..a1668c5bfc5 100644 --- a/tests/e2e/load/locust_load.py +++ b/tests/e2e/load/locust_load.py @@ -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), + ) diff --git a/tests/e2e/load/locustfile.py b/tests/e2e/load/locustfile.py new file mode 100644 index 00000000000..3d9ef7c83ea --- /dev/null +++ b/tests/e2e/load/locustfile.py @@ -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", + ) diff --git a/tests/e2e/load/proxy_usage.py b/tests/e2e/load/proxy_usage.py new file mode 100644 index 00000000000..b48129f0099 --- /dev/null +++ b/tests/e2e/load/proxy_usage.py @@ -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) diff --git a/tests/e2e/load/test_locust_load.py b/tests/e2e/load/test_locust_load.py index e3cc6f3efd5..40a238ef624 100644 --- a/tests/e2e/load/test_locust_load.py +++ b/tests/e2e/load/test_locust_load.py @@ -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) ) ) diff --git a/tests/e2e/load/test_proxy_usage.py b/tests/e2e/load/test_proxy_usage.py new file mode 100644 index 00000000000..8d9f848825f --- /dev/null +++ b/tests/e2e/load/test_proxy_usage.py @@ -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" + ) diff --git a/tests/e2e/load/test_redis_chaos_e2e.py b/tests/e2e/load/test_redis_chaos_e2e.py new file mode 100644 index 00000000000..f91b9c2fa09 --- /dev/null +++ b/tests/e2e/load/test_redis_chaos_e2e.py @@ -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 diff --git a/tests/e2e/models.py b/tests/e2e/models.py index f3c111bdf06..31eb6e31401 100644 --- a/tests/e2e/models.py +++ b/tests/e2e/models.py @@ -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"] diff --git a/tests/e2e/pytest.ini b/tests/e2e/pytest.ini index f69d5d058f3..1dcb50a6fab 100644 --- a/tests/e2e/pytest.ini +++ b/tests/e2e/pytest.ini @@ -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 diff --git a/tests/e2e/router/conftest.py b/tests/e2e/router/conftest.py index 57fde729f41..2d5e546dc6c 100644 --- a/tests/e2e/router/conftest.py +++ b/tests/e2e/router/conftest.py @@ -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) diff --git a/tests/e2e/router/test_redis_timeout_e2e.py b/tests/e2e/router/test_redis_timeout_e2e.py deleted file mode 100644 index 3f1984c0696..00000000000 --- a/tests/e2e/router/test_redis_timeout_e2e.py +++ /dev/null @@ -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]}" diff --git a/uv.lock b/uv.lock index 0fe787645a2..be124e2ead7 100644 --- a/uv.lock +++ b/uv.lock @@ -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 = [