fix(e2e): make load throughput suite use a fresh mock model and report failures

The stage load SLO was failing at ~95% errors while RPS looked fine. A fixed load-mock name could reuse a stale deployment without mock_response, so Locust hit real OpenAI. Always register a unique mock model, preflight one chat, hit /v1/chat/completions, and attach a status/exception failure breakdown to the assert so the next red run is diagnosable.
2026-09-16 23:41:43 +00:00 · 2026-07-18 14:08:18 -07:00 · 2026-07-18 14:08:18 -07:00 · a150a3f85f
commit a150a3f85f
parent 4f8d83ca85
6 changed files with 209 additions and 98 deletions
--- 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
--- 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)
--- 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"
--- 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))
--- 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()
--- 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%})"
        )