diff --git a/tests/e2e/load/conftest.py b/tests/e2e/load/conftest.py index e9fba02680d..b7b9c957a62 100644 --- a/tests/e2e/load/conftest.py +++ b/tests/e2e/load/conftest.py @@ -3,64 +3,41 @@ from __future__ import annotations from collections.abc import Iterator import pytest -from requests import RequestException -from e2e_http import NoBody, Success +from e2e_config import unique_marker from load_client import LoadClient, build_client -from load_constants import LOAD_MODEL -from models import KeyGenerateBody, LiteLLMParamsBody, ModelsListResponse +from load_constants import LOAD_MOCK_PARAMS from lifecycle import ResourceManager +from models import KeyGenerateBody from proxy_client import ProxyClient -LOAD_MODEL_PARAMS = LiteLLMParamsBody( - model="openai/load-mock", - mock_response="This is a mock response for the throughput load test.", -) - @pytest.fixture(scope="session") def client(proxy: ProxyClient) -> LoadClient: return build_client(proxy) -def _model_is_servable(proxy: ProxyClient, model_name: str) -> bool: - result = proxy.transport.get( - "/v1/models", - headers=proxy.transport.master, - params=NoBody(), - response_type=ModelsListResponse, - ) - return isinstance(result, Success) and any(entry.id == model_name for entry in result.data.data) - - -@pytest.fixture(scope="session", autouse=True) -def _ensure_load_model( # pyright: ignore[reportUnusedFunction] # pytest autouse session fixture, wired by name - client: LoadClient, -) -> Iterator[None]: - proxy = client.proxy - if _model_is_servable(proxy, LOAD_MODEL): - yield - return +@pytest.fixture(scope="session") +def load_model(client: LoadClient) -> Iterator[str]: + """Register a fresh mock deployment for this session and delete it after. + A fixed name like ``load-mock`` is unsafe on a shared stage proxy: a prior + run (or a hand-registered row) can leave a deployment without mock_response, + so Locust would hit real OpenAI with an invalid model and fail ~all requests + while the fixture skipped /model/new because the name was already listed. + """ + model_name = f"load-mock-{unique_marker()}" + model_id = client.proxy.create_model(model_name, LOAD_MOCK_PARAMS) try: - model_id = proxy.create_model(LOAD_MODEL, LOAD_MODEL_PARAMS) - except (AssertionError, RequestException) as exc: - if _model_is_servable(proxy, LOAD_MODEL): - yield - return - raise AssertionError( - f"failed to register {LOAD_MODEL!r} for the throughput load test " - f"(not listed on the data plane and /model/new failed): {exc}" - ) from exc - - try: - yield + yield model_name finally: - proxy.delete_model(model_id) + client.proxy.delete_model(model_id) @pytest.fixture -def load_key(resources: ResourceManager, client: LoadClient) -> str: - key = client.proxy.generate_key(KeyGenerateBody(models=[LOAD_MODEL], user_id="e2e-load")) +def load_key(resources: ResourceManager, client: LoadClient, load_model: str) -> str: + key = client.proxy.generate_key( + KeyGenerateBody(models=[load_model], user_id=f"e2e-load-{unique_marker()}") + ) resources.defer(lambda: client.proxy.delete_key(key)) return key diff --git a/tests/e2e/load/load_client.py b/tests/e2e/load/load_client.py index c3ce3890e8a..41b6e0a5d7f 100644 --- a/tests/e2e/load/load_client.py +++ b/tests/e2e/load/load_client.py @@ -2,6 +2,9 @@ from __future__ import annotations from dataclasses import dataclass +from e2e_http import unwrap +from load_constants import LOAD_MOCK_BODY_SNIPPET +from models import ChatBody, ChatMessage from proxy_client import ProxyClient @@ -9,6 +12,27 @@ from proxy_client import ProxyClient class LoadClient: proxy: ProxyClient + def preflight_mock_chat(self, *, key: str, model: str) -> None: + """One real chat before Locust; fail with the body if mock routing is broken.""" + response = unwrap( + self.proxy.chat( + key, + ChatBody( + model=model, + messages=[ChatMessage(role="user", content="preflight load mock")], + max_tokens=16, + ), + ) + ) + content = "" + if response.choices and response.choices[0].message is not None: + content = response.choices[0].message.content or "" + assert LOAD_MOCK_BODY_SNIPPET in content, ( + f"load preflight to {model!r} did not return the mock body " + f"(mock_response not applied or wrong deployment). content={content!r} " + f"full={response!r}" + ) + def build_client(proxy: ProxyClient) -> LoadClient: return LoadClient(proxy=proxy) diff --git a/tests/e2e/load/load_constants.py b/tests/e2e/load/load_constants.py index fd97f1398f4..0092944facb 100644 --- a/tests/e2e/load/load_constants.py +++ b/tests/e2e/load/load_constants.py @@ -1,3 +1,13 @@ from __future__ import annotations -LOAD_MODEL = "load-mock" +from models import LiteLLMParamsBody + +# Backend id is never called: mock_response short-circuits before OpenAI. +# model_name on the proxy is unique per session (see conftest) so a stale +# stage deployment without mock_response cannot be reused across runs. +LOAD_MOCK_PARAMS = LiteLLMParamsBody( + model="openai/load-mock", + mock_response="This is a mock response for the throughput load test.", +) + +LOAD_MOCK_BODY_SNIPPET = "mock response for the throughput load test" diff --git a/tests/e2e/load/locust_load.py b/tests/e2e/load/locust_load.py index 990e24ee064..3ac7e97cd5c 100644 --- a/tests/e2e/load/locust_load.py +++ b/tests/e2e/load/locust_load.py @@ -1,8 +1,10 @@ from __future__ import annotations +import json import os import subprocess import sys +import tempfile from dataclasses import dataclass from pathlib import Path @@ -26,20 +28,59 @@ class LoadResult: requests: int failures: int requests_per_second: float + failure_reasons: tuple[tuple[str, int], ...] = () @property def failure_ratio(self) -> float: return self.failures / self.requests if self.requests else 1.0 + def format_failure_reasons(self, *, limit: int = 10) -> str: + if not self.failure_reasons: + return "(no per-status breakdown; Locust did not write a failure report)" + lines = [f" {reason}: {count}" for reason, count in self.failure_reasons[:limit]] + return "\n".join(lines) -def _aggregate(entries: list[_LocustStatEntry]) -> LoadResult: + +def _aggregate( + entries: list[_LocustStatEntry], + failure_reasons: tuple[tuple[str, int], ...], +) -> LoadResult: requests = sum(entry.num_requests for entry in entries) failures = sum(entry.num_failures for entry in entries) if not entries or requests == 0: - return LoadResult(requests=requests, failures=failures, requests_per_second=0.0) - elapsed = max(entry.last_request_timestamp for entry in entries) - min(entry.start_time for entry in entries) + return LoadResult( + requests=requests, + failures=failures, + requests_per_second=0.0, + failure_reasons=failure_reasons, + ) + elapsed = max(entry.last_request_timestamp for entry in entries) - min( + entry.start_time for entry in entries + ) rps = requests / elapsed if elapsed > 0 else 0.0 - return LoadResult(requests=requests, failures=failures, requests_per_second=rps) + return LoadResult( + requests=requests, + failures=failures, + requests_per_second=rps, + failure_reasons=failure_reasons, + ) + + +def _read_failure_report(path: Path) -> tuple[tuple[str, int], ...]: + if not path.is_file(): + return () + try: + raw = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return () + if not isinstance(raw, dict): + return () + pairs: list[tuple[str, int]] = [] + for key, value in raw.items(): + if isinstance(key, str) and isinstance(value, int): + pairs.append((key, value)) + pairs.sort(key=lambda item: item[1], reverse=True) + return tuple(pairs) def run_chat_load( @@ -51,43 +92,50 @@ def run_chat_load( spawn_rate: float, duration_seconds: float, ) -> LoadResult: - completed = subprocess.run( - [ - sys.executable, - "-m", - "locust", - "--headless", - "--json", - "--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_KEY": api_key, "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}" + with tempfile.TemporaryDirectory(prefix="e2e-load-") as tmp: + report_path = Path(tmp) / "failure_reasons.json" + completed = subprocess.run( + [ + sys.executable, + "-m", + "locust", + "--headless", + "--json", + "--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_KEY": api_key, + "LOAD_MODEL": model, + "LOAD_FAILURE_REPORT_PATH": str(report_path), + }, + capture_output=True, + text=True, + timeout=duration_seconds + 120, + check=False, ) - 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(entries) + 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 " + f"summary via --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 " + f"on stdout; got stdout={completed.stdout!r}, stderr={completed.stderr!r}" + ) from exc + return _aggregate(entries, _read_failure_report(report_path)) diff --git a/tests/e2e/load/locustfile.py b/tests/e2e/load/locustfile.py index 4aa7517ca0b..44c7033c9c3 100644 --- a/tests/e2e/load/locustfile.py +++ b/tests/e2e/load/locustfile.py @@ -1,8 +1,11 @@ from __future__ import annotations +import json import os +from collections import Counter +from typing import Any -from locust import FastHttpUser, constant, task +from locust import FastHttpUser, constant, events, task _MODEL = os.environ["LOAD_MODEL"] _HEADERS = {"Authorization": f"Bearer {os.environ['LOAD_API_KEY']}"} @@ -12,16 +15,57 @@ _PAYLOAD = { "temperature": 0, "max_tokens": 16, } +_FAILURE_REPORT_PATH = os.environ.get("LOAD_FAILURE_REPORT_PATH", "") + +_failure_reasons: Counter[str] = Counter() + + +@events.request.add_listener +def _record_failure( # pyright: ignore[reportUnusedFunction] # locust event hook + request_type: str, + name: str, + response_time: float, + response_length: int, + response: Any, + context: dict[str, object], + exception: BaseException | None, + start_time: float, + url: str, + **kwargs: object, +) -> None: + if exception is not None: + _failure_reasons[f"exception:{type(exception).__name__}"] += 1 + return + status = getattr(response, "status_code", None) + if isinstance(status, int) and status >= 400: + _failure_reasons[f"http:{status}"] += 1 + + +@events.quitting.add_listener +def _write_failure_report(environment: object, **kwargs: object) -> None: # pyright: ignore[reportUnusedFunction] + if not _FAILURE_REPORT_PATH: + return + path = _FAILURE_REPORT_PATH + payload = dict(_failure_reasons.most_common(20)) + with open(path, "w", encoding="utf-8") as handle: + json.dump(payload, handle) class ChatUser(FastHttpUser): wait_time = constant(0) + connection_timeout = 10.0 + network_timeout = 60.0 @task def chat(self) -> None: - self.client.post( # pyright: ignore[reportUnknownMemberType] # locust FastHttpSession.post types json/**kwargs as Any - "/chat/completions", + with self.client.post( # pyright: ignore[reportUnknownMemberType] + "/v1/chat/completions", json=_PAYLOAD, headers=_HEADERS, - name="/chat/completions", - ) + name="/v1/chat/completions", + catch_response=True, + ) as response: + if response.status_code != 200: + response.failure(f"status={response.status_code}") + return + response.success() diff --git a/tests/e2e/load/test_chat_completions_throughput_e2e.py b/tests/e2e/load/test_chat_completions_throughput_e2e.py index 6dd5fff971a..0e7e527e28b 100644 --- a/tests/e2e/load/test_chat_completions_throughput_e2e.py +++ b/tests/e2e/load/test_chat_completions_throughput_e2e.py @@ -9,7 +9,6 @@ from e2e_config import ( PROXY_BASE_URL, ) from load_client import LoadClient -from load_constants import LOAD_MODEL from locust_load import run_chat_load pytestmark = [pytest.mark.e2e, pytest.mark.load] @@ -17,11 +16,15 @@ pytestmark = [pytest.mark.e2e, pytest.mark.load] class TestChatCompletionsThroughput: @pytest.mark.covers("reliability.perf.throughput.under_slo") - def test_sustains_throughput_slo_under_load(self, client: LoadClient, load_key: str) -> None: + def test_sustains_throughput_slo_under_load( + self, client: LoadClient, load_key: str, load_model: str + ) -> None: + client.preflight_mock_chat(key=load_key, model=load_model) + result = run_chat_load( base_url=PROXY_BASE_URL, api_key=load_key, - model=LOAD_MODEL, + model=load_model, users=LOAD_USERS, spawn_rate=LOAD_SPAWN_RATE, duration_seconds=LOAD_DURATION_SECONDS, @@ -32,11 +35,16 @@ class TestChatCompletionsThroughput: f"the load generator never drove traffic (proxy unreachable or model unservable)" ) assert result.failure_ratio <= LOAD_MAX_FAILURE_RATIO, ( - f"{result.failures}/{result.requests} requests failed " - f"({result.failure_ratio:.1%} > {LOAD_MAX_FAILURE_RATIO:.1%} allowed); " - f"throughput of {result.requests_per_second:.1f} RPS is not a clean read under this error rate" + f"load failure ratio {result.failure_ratio:.1%} exceeds " + f"{LOAD_MAX_FAILURE_RATIO:.1%} allowed " + f"({result.failures}/{result.requests} failed, " + f"{result.requests_per_second:.1f} RPS observed; not a clean SLO read).\n" + f"failure breakdown:\n{result.format_failure_reasons()}\n" + f"users={LOAD_USERS} spawn_rate={LOAD_SPAWN_RATE} " + f"duration={LOAD_DURATION_SECONDS}s model={load_model!r}" ) assert result.requests_per_second >= LOAD_MIN_RPS, ( f"sustained {result.requests_per_second:.1f} RPS over {LOAD_DURATION_SECONDS}s with " - f"{LOAD_USERS} users, below the {LOAD_MIN_RPS} RPS SLO; the proxy request path regressed under load" + f"{LOAD_USERS} users, below the {LOAD_MIN_RPS} RPS SLO; the proxy request path " + f"regressed under load (failure_ratio={result.failure_ratio:.1%})" )