From b99d2cc06e9f5a05765228b85bc64d2a090ffd00 Mon Sep 17 00:00:00 2001 From: Kerry Lu Date: Tue, 20 Oct 2026 22:31:08 -0700 Subject: [PATCH] test(perf): ramp the load cases until the pod stops keeping up and floor sustained_rps (#47758) * test(perf): add small_bodies_30rps and mixed_size_requests_4rps load cases Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(perf): run the load cases at 50 rps small and 5 rps mixed, the pod's knee Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(perf): ramp the load cases until the pod stops keeping up and report sustained_rps Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(perf): hand the ramp to the observation and widen the mixed ramp to 14 rps Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(perf): floor the small ramp at 20 rps and emit the per-step readings Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(perf): emit the ramp's hold bound and each step's unexpected statuses Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(perf): read each ramp step's cpu, redis latency and connections from the samplers' windows Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(perf): floor mixed_size_requests_ramp at 8 rps from the lane's 10 rps sustained step Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(perf): compare step readings with math.isclose so the e2e basedpyright gate stays clean Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: kerry Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- tests/e2e/AGENTS.md | 2 +- .../perf/cases/mixed_size_requests_ramp.py | 24 +++ tests/e2e/perf/cases/small_bodies_ramp.py | 24 +++ tests/e2e/perf/setup/datastore.py | 18 +- tests/e2e/perf/setup/locustfile.py | 15 +- tests/e2e/perf/setup/observe.py | 129 ++++++++++++-- tests/e2e/perf/setup/performance_test_case.py | 78 +++++++- tests/e2e/perf/setup/plan.py | 4 + tests/e2e/perf/setup/podcpu.py | 18 +- tests/e2e/perf/setup/ramp.py | 116 ++++++++++++ tests/e2e/perf/setup/run.py | 1 + tests/e2e/perf/setup/series.py | 48 +++++ tests/e2e/perf/setup/telemetry.py | 51 +++++- tests/e2e/perf/setup/upgrade.py | 4 +- tests/e2e/perf/setup/wire.py | 2 + tests/e2e/perf/test_perf_cases.py | 15 +- tests/e2e_harness/perf/setup/test_ramp.py | 168 ++++++++++++++++++ tests/e2e_harness/perf/test_upgrade.py | 4 +- 18 files changed, 677 insertions(+), 44 deletions(-) create mode 100644 tests/e2e/perf/cases/mixed_size_requests_ramp.py create mode 100644 tests/e2e/perf/cases/small_bodies_ramp.py create mode 100644 tests/e2e/perf/setup/ramp.py create mode 100644 tests/e2e/perf/setup/series.py create mode 100644 tests/e2e_harness/perf/setup/test_ramp.py diff --git a/tests/e2e/AGENTS.md b/tests/e2e/AGENTS.md index d976a2b5ee3..ef57135a4d2 100644 --- a/tests/e2e/AGENTS.md +++ b/tests/e2e/AGENTS.md @@ -44,7 +44,7 @@ Each subdirectory under `tests/e2e/` is one suite, scoped to an endpoint family - `logging/` - logging-integration delivery (datadog and friends) - `security/` - secret handling and log-leak protection - `router/` - routing and reliability behavior (fallbacks, cooldowns) plus the memory tests (`test_reliability_memory_e2e.py`: every worker's RSS as read at collection time, before any test traffic, must sit under a fixed idle budget, the release-gate check for a DB-backed boot that idles near the pod limit the way v1.100.x did, and a few hundred failing requests with retries and fallbacks must not grow proxy RSS past a loose ceiling over the measured batches nor store a request snapshot past a fixed size, the release-gate check for the v1.100.0 retry-breadcrumb leak) -- `perf/` - 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 split round robin over `/chat/completions` and `/v1/messages`, one endpoint per simulated user, with `CLIENT PAUSE ALL` on the proxy's Redis mid-run to simulate it being down outright, asserting zero failed requests on every endpoint, budgeting RSS and CPU-per-request as ratios against the same run's healthy phase, and holding p50/p90/p99 latency and log-bytes-per-request to flat ceilings (a ratio cannot bound those two: an open breaker skips Redis instead of waiting on it, so the chaos phase can measure cheaper than baseline while still being far slower than a user should see), needs a proxy booted from `gateway/redis_chaos_ci_config.yml` on the same host with `E2E_PROXY_PID` and `E2E_PROXY_LOG` set, marked `redis_chaos`, deselected unless `E2E_PERF_TESTS` is set and excluded from the per-PR selector like the rest of `perf/`, driven by `.github/workflows/test-e2e-redis-chaos.yml` and by the Buildkite `e2e-redis-chaos` step in project-releaser, which runs the proxy, Postgres and Valkey co-located with pytest in one pod and sets the opt-in). New perf tests go here too, marked `perf`, behind the same `E2E_PERF_TESTS` opt-in. The perf cases (`test_perf_cases.py`, one `PerformanceTestCase` per module in `perf/cases/`, no `_e2e` suffix) are the request-performance runs. A case names what the gateway was deployed with (model groups and guardrails, written out in the gateway's config: the `proxy_config` block of project-releaser's perf helm values in the lane, `gateway/perf_ci_config.yml` locally, with each deployment's mock knobs in an `x-mock-knobs` extra header so the model ids are real ones like `gpt-6-luna` and `anthropic.claude-sonnet-5`) and lists what the test creates and sends: teams, Keycloak JWT users, virtual keys, `RequestShape`s (a coding-agent session from `perf/setup/corpus/` whose input-token size is drawn per request from the shape's range, with or without tools, and the output cap), a `Traffic` plan (seed, rps, request count, endpoint, shape and stream shares) and `Limit`s (one metric at one percentile under one ceiling). The infra side lives in `perf/setup/`: `provision.py` checks the deployed config matches the case and builds the callers, `plan.py` draws the request list from the seed, `run.py` replays it from a separate Locust process (`locustfile.py`) that builds each body right before sending and records status, header, first-byte, first-token and total time per request, and `observe.py` joins all of that with what the mocks received and the workers' RSS to score every `Limit`. Limits are about what the caller sees and what the gateway costs: every non-blocked request is a 200 (`unexpected_status_count`), the seconds the gateway added per request at p50 and p99 (`gateway_added_seconds`, to the first token on streams), the client's time to the first token of a stream at p50 and p99 (`time_to_first_token_seconds`, mock delay included), worker memory growth (`memory_growth_megabytes`), and the Postgres statements and Redis commands a request waits on before it is forwarded (`postgres_calls_per_request`, `redis_calls_per_request`: auth and routing reads, the rate-limiter script), with the work the gateway does after the response or on a timer (spend writes, daily aggregates, config reloads) reported separately as `postgres_background_calls_per_request` and `redis_background_calls_per_request`; the same overhead as the gateway itself reports it, read from the `x-litellm-overhead-duration-ms` header per request (`litellm_overhead_seconds`, gated next to `gateway_added_seconds`, the outside reading of the same thing, so the two can be charted together) and the gateway's own end-to-end time from `x-litellm-response-duration-ms` (`litellm_response_seconds`), the run mean of the gateway's `litellm_redis_latency` histogram from a `/metrics` scrape before and after the run (`telemetry.py`, `prometheus_redis_latency_mean_seconds`), and the database's own view of the run from the same `pg_stat_statements` rows and a once-a-second `pg_stat_activity` sample: execution seconds per wall second (`postgres_active_sessions`, the average-active-sessions number RDS reports as DB load), the slowest statement's mean (`postgres_slowest_statement_seconds`, named in the detail) and the peak client connection count (`postgres_peak_connections`), each under a ceiling set from the first local runs and due a retune once the lane has its ten-run baseline; `datastore.py` reads `pg_stat_statements` before and after the run and holds a Redis `MONITOR` connection during it, sorts each statement by verb and table and each command by key, and the run output lists the top statements and commands per bucket so a red run names what grew. Whether a body, a guardrail call or a spend row is right is printed as a diagnostic, never a limit, that is the functional suites' job. The providers and the guardrail are out-of-process servers in `perf/setup/mocks/` (`server.py openai|anthropic|bedrock|guardrail`, each answering with a reply generated for the request in that provider's wire format, sized from the prompt and output cap, a tool call when tools were offered, waiting the case's `mock_latency` first, request logs at `/mock/requests` and `/mock/calls`), reached through `E2E_PERF_MOCK_*_URL` by both the gateway config and the test. Adding a perf test is adding a module with a `CASE` to `perf/cases/`; a new deployment or guardrail is a config change; a new metric is a name in `Metric` plus a branch in `observe.observe` +- `perf/` - 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 split round robin over `/chat/completions` and `/v1/messages`, one endpoint per simulated user, with `CLIENT PAUSE ALL` on the proxy's Redis mid-run to simulate it being down outright, asserting zero failed requests on every endpoint, budgeting RSS and CPU-per-request as ratios against the same run's healthy phase, and holding p50/p90/p99 latency and log-bytes-per-request to flat ceilings (a ratio cannot bound those two: an open breaker skips Redis instead of waiting on it, so the chaos phase can measure cheaper than baseline while still being far slower than a user should see), needs a proxy booted from `gateway/redis_chaos_ci_config.yml` on the same host with `E2E_PROXY_PID` and `E2E_PROXY_LOG` set, marked `redis_chaos`, deselected unless `E2E_PERF_TESTS` is set and excluded from the per-PR selector like the rest of `perf/`, driven by `.github/workflows/test-e2e-redis-chaos.yml` and by the Buildkite `e2e-redis-chaos` step in project-releaser, which runs the proxy, Postgres and Valkey co-located with pytest in one pod and sets the opt-in). New perf tests go here too, marked `perf`, behind the same `E2E_PERF_TESTS` opt-in. The perf cases (`test_perf_cases.py`, one `PerformanceTestCase` per module in `perf/cases/`, no `_e2e` suffix) are the request-performance runs. A case names what the gateway was deployed with (model groups and guardrails, written out in the gateway's config: the `proxy_config` block of project-releaser's perf helm values in the lane, `gateway/perf_ci_config.yml` locally, with each deployment's mock knobs in an `x-mock-knobs` extra header so the model ids are real ones like `gpt-6-luna` and `anthropic.claude-sonnet-5`) and lists what the test creates and sends: teams, Keycloak JWT users, virtual keys, `RequestShape`s (a coding-agent session from `perf/setup/corpus/` whose input-token size is drawn per request from the shape's range, with or without tools, and the output cap), a `Traffic` plan (seed, rps, request count, endpoint, shape and stream shares, or a `Ramp` that steps the rate up through the run, built with `ramped()`, whose plan carries each request's send second and whose run reports `sustained_rps`, the last step the pod held, scored in `perf/setup/ramp.py`, with every step's latency, completed rate, non-200s, CPU per request, Redis latency and Postgres connections cut from the samplers' time-stamped `Series` (`perf/setup/series.py`) to the step's window and emitted as `perf_ramp_step__` JUnit properties for the dashboard) and `Limit`s (one metric at one percentile under one ceiling, or over one floor for `sustained_rps`). The infra side lives in `perf/setup/`: `provision.py` checks the deployed config matches the case and builds the callers, `plan.py` draws the request list from the seed, `run.py` replays it from a separate Locust process (`locustfile.py`) that builds each body right before sending and records status, header, first-byte, first-token and total time per request, and `observe.py` joins all of that with what the mocks received and the workers' RSS to score every `Limit`. Limits are about what the caller sees and what the gateway costs: every non-blocked request is a 200 (`unexpected_status_count`), the seconds the gateway added per request at p50 and p99 (`gateway_added_seconds`, to the first token on streams), the client's time to the first token of a stream at p50 and p99 (`time_to_first_token_seconds`, mock delay included), worker memory growth (`memory_growth_megabytes`), and the Postgres statements and Redis commands a request waits on before it is forwarded (`postgres_calls_per_request`, `redis_calls_per_request`: auth and routing reads, the rate-limiter script), with the work the gateway does after the response or on a timer (spend writes, daily aggregates, config reloads) reported separately as `postgres_background_calls_per_request` and `redis_background_calls_per_request`; the same overhead as the gateway itself reports it, read from the `x-litellm-overhead-duration-ms` header per request (`litellm_overhead_seconds`, gated next to `gateway_added_seconds`, the outside reading of the same thing, so the two can be charted together) and the gateway's own end-to-end time from `x-litellm-response-duration-ms` (`litellm_response_seconds`), the run mean of the gateway's `litellm_redis_latency` histogram from a `/metrics` scrape before and after the run (`telemetry.py`, `prometheus_redis_latency_mean_seconds`), and the database's own view of the run from the same `pg_stat_statements` rows and a once-a-second `pg_stat_activity` sample: execution seconds per wall second (`postgres_active_sessions`, the average-active-sessions number RDS reports as DB load), the slowest statement's mean (`postgres_slowest_statement_seconds`, named in the detail) and the peak client connection count (`postgres_peak_connections`), each under a ceiling set from the first local runs and due a retune once the lane has its ten-run baseline; `datastore.py` reads `pg_stat_statements` before and after the run and holds a Redis `MONITOR` connection during it, sorts each statement by verb and table and each command by key, and the run output lists the top statements and commands per bucket so a red run names what grew. Whether a body, a guardrail call or a spend row is right is printed as a diagnostic, never a limit, that is the functional suites' job. The providers and the guardrail are out-of-process servers in `perf/setup/mocks/` (`server.py openai|anthropic|bedrock|guardrail`, each answering with a reply generated for the request in that provider's wire format, sized from the prompt and output cap, a tool call when tools were offered, waiting the case's `mock_latency` first, request logs at `/mock/requests` and `/mock/calls`), reached through `E2E_PERF_MOCK_*_URL` by both the gateway config and the test. Adding a perf test is adding a module with a `CASE` to `perf/cases/`; a new deployment or guardrail is a config change; a new metric is a name in `Metric` plus a branch in `observe.observe` - `other/` - the holding-pen suite for the `other.*` registry cluster with no home of its own yet: the master-key auth gate, JWT auth (access tokens issued by a real Keycloak realm, `idp.py` plus `idp_realm.json`, whose JWKS the proxy's `JWT_PUBLIC_KEY_URL` points at, see CONTRIBUTING.md for the start command and config block), 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 - `secret_manager/` - the gateway's `key_management_system` against a real secret manager: deployment keys resolved from it (`os.environ/` where the name exists only in the manager) and virtual keys written to and deleted from it. The tests are backend-agnostic and each backend is its own lane, because the setting is global to the proxy: `E2E_SECRET_MANAGER=` opts in and picks the backend from `secret_backends.BACKENDS`, the proxy is booted from `gateway/secret_manager__ci_config.yml` against the live manager, and the tests reach that manager through the backend's `SecretStore` (`secret_store_.py`). A test needing something not every backend does carries `requires_capability(...)` and is deselected on lanes that lack it. `secret_manager/backend.sh up ` runs a backend in Docker and writes the proxy's and the tests' env. Marked `secret_manager`, deselected unless `E2E_SECRET_MANAGER` is set, and kept out of the per-PR selector. Backends today: `hashicorp_vault` and `cyberark` (CyberArk Conjur, which cannot delete, so the delete test is Vault-only) - `gateway/` - proxy configuration only (`litellm-config.yml`). No tests diff --git a/tests/e2e/perf/cases/mixed_size_requests_ramp.py b/tests/e2e/perf/cases/mixed_size_requests_ramp.py new file mode 100644 index 00000000000..d7db545c031 --- /dev/null +++ b/tests/e2e/perf/cases/mixed_size_requests_ramp.py @@ -0,0 +1,24 @@ +"""The mixed-size traffic stepped up until the pod stops keeping up: the same callers, guardrails, mocks and body +mix (short chats, 300k to 1M-token contexts, tool calls, half streamed) as `mixed_size_requests_2rps`, 2 rps more +every 90 s from 2 to 14 on the same one 4-worker pod. Where the 2 rps case prices one big body on its own, this one +reports `sustained_rps`, the last step the pod held with big bodies overlapping in every worker, so a handler that +re-walks a 1M-token body on the event loop or a worker that holds several of them in memory at once moves the knee +down and fails the floor.""" + +from __future__ import annotations + +from dataclasses import replace +from typing import Final + +from setup.performance_test_case import Limit, Ramp, ramped + +from cases.mixed_size_requests_2rps import CASE as MIXED + +RAMP: Final = Ramp(first_rps=2.0, last_rps=14.0, step_rps=2.0, step_seconds=90.0, held_p99_seconds=5.0) + +CASE: Final = replace( + MIXED, + name="mixed_size_requests_ramp", + traffic=ramped(MIXED.traffic, RAMP, seed=9, concurrency=300), + limits=(Limit("sustained_rps", min=8),), +) diff --git a/tests/e2e/perf/cases/small_bodies_ramp.py b/tests/e2e/perf/cases/small_bodies_ramp.py new file mode 100644 index 00000000000..64308168cee --- /dev/null +++ b/tests/e2e/perf/cases/small_bodies_ramp.py @@ -0,0 +1,24 @@ +"""The small-bodies traffic stepped up until the pod stops keeping up: the same callers, guardrails, mocks and +2.5k to 4k-token bodies as `small_bodies_5rps`, 10 rps more every 90 s from 10 to 70 on the same one 4-worker pod. +The 5 rps case prices one request with the pod idle; this one reports `sustained_rps`, the last step the pod +held (gateway added p99 under 5 s, the queue still draining), so a change that moves the knee down, a lock held +across an await, a pool that fills, a flush that stalls the loop while dozens of requests are in flight, fails the +floor instead of hiding under a fixed rate the pod still manages.""" + +from __future__ import annotations + +from dataclasses import replace +from typing import Final + +from setup.performance_test_case import Limit, Ramp, ramped + +from cases.small_bodies_5rps import CASE as SMALL + +RAMP: Final = Ramp(first_rps=10.0, last_rps=70.0, step_rps=10.0, step_seconds=90.0, held_p99_seconds=5.0) + +CASE: Final = replace( + SMALL, + name="small_bodies_ramp", + traffic=ramped(SMALL.traffic, RAMP, seed=13, concurrency=400), + limits=(Limit("sustained_rps", min=20),), +) diff --git a/tests/e2e/perf/setup/datastore.py b/tests/e2e/perf/setup/datastore.py index 89739b6d13e..6ce37da3554 100644 --- a/tests/e2e/perf/setup/datastore.py +++ b/tests/e2e/perf/setup/datastore.py @@ -22,6 +22,8 @@ from typing import TYPE_CHECKING, Final, Literal from pydantic import TypeAdapter +from setup.series import Recorder, Series + if TYPE_CHECKING: import redis @@ -182,13 +184,22 @@ def slowest_statement(statements: tuple[StatementCalls, ...] | None) -> Statemen class ConnectionsSampler: """`pg_stat_activity` client backends on the gateway's database once a second on a background thread, minus - this sampler's own connection, so the peak is what the gateway (and the test's own reads) held open.""" + this sampler's own connection, so the peak is what the gateway (and the test's own reads) held open. + `series` keeps every count with its time for a ramped run's per-step peaks.""" def __init__(self, url: str) -> None: self._url: Final = url self._stop: Final = threading.Event() self._thread: Final = threading.Thread(target=self._run, name="postgres-connections", daemon=True) - self.peak: int | None = None + self._counts: Final[Recorder[int]] = Recorder() + + @property + def peak(self) -> int | None: + return max((sample.value for sample in self._counts.series.samples), default=None) + + @property + def series(self) -> Series[int]: + return self._counts.series def start(self) -> None: self._thread.start() @@ -203,12 +214,13 @@ class ConnectionsSampler: with psycopg.connect(self._url, autocommit=True) as conn: while not self._stop.is_set(): (count,) = _COMMITS_ROW.validate_python(conn.execute(_CONNECTIONS_QUERY).fetchone()) - self.peak = count if self.peak is None else max(self.peak, count) + self._counts.record(count) _ = self._stop.wait(_CONNECTIONS_SAMPLE_SECONDS) class NoConnectionsSampler: peak: None = None + series: None = None def start(self) -> None: return None diff --git a/tests/e2e/perf/setup/locustfile.py b/tests/e2e/perf/setup/locustfile.py index 1c38eae22a1..82a5749becf 100644 --- a/tests/e2e/perf/setup/locustfile.py +++ b/tests/e2e/perf/setup/locustfile.py @@ -4,7 +4,8 @@ time to headers, time to first byte, time to first token (streams), total time, and whether it matched. Runs in its own process (`setup/run.py` starts it), never in the gateway's. Reads `PERF_PLAN_FILE` (JSON list of wire requests), writes `PERF_RECORDS_FILE` (one JSON object per -line), paces to `PERF_RPS` over `PERF_USERS` users, and quits once the last request is recorded. +line), paces to `PERF_RPS` over `PERF_USERS` users (or, when the plan carries `send_at_s`, sends each +request at its scheduled second), and quits once the last request is recorded. """ from __future__ import annotations @@ -35,6 +36,8 @@ _JITTER: Final = 0.25 _RNG: Final = random.Random(_PLAN[0].marker if _PLAN else "empty-plan") _RECORDS: Final = _RECORDS_PATH.open("a", encoding="utf-8") _PROGRESS: Final = PlanProgress(len(_PLAN)) +_SCHEDULED: Final = bool(_PLAN) and _PLAN[0].send_at_s is not None +_RUN_STARTED: Final = time.monotonic() _STOP_FILE: Final = os.environ.get("PERF_STOP_FILE") BLOCK_REASON: Final = "perf-test-block" _TOKEN_MAX_AGE_S: Final = 240.0 @@ -211,7 +214,8 @@ def _authorization(user: FastHttpUser, item: WireRequest) -> str: class PlanUser(FastHttpUser): """One simulated client; all users share the plan and together hold the planned rate. Each user starts at a random point of its first interval and jitters every later interval by up to `_JITTER` either way, so the - users never fire in lockstep; the mean interval is unchanged and `rps` stays the total rate.""" + users never fire in lockstep; the mean interval is unchanged and `rps` stays the total rate. A scheduled + plan (a ramp) is paced by the plan instead: a user takes the next request and sleeps until its second.""" network_timeout = float(os.environ.get("PERF_REQUEST_TIMEOUT", "120")) connection_timeout = 30.0 @@ -219,9 +223,12 @@ class PlanUser(FastHttpUser): def on_start(self) -> None: self._last_run = 0.0 self._last_wait = 0.0 - gevent.sleep(_RNG.uniform(0.0, _INTERVAL_S)) + if not _SCHEDULED: + gevent.sleep(_RNG.uniform(0.0, _INTERVAL_S)) def wait_time(self) -> float: + if _SCHEDULED: + return 0.1 target: Final = _INTERVAL_S * _RNG.uniform(1.0 - _JITTER, 1.0 + _JITTER) now: Final = time.monotonic() run_s: Final = now - self._last_run - self._last_wait if self._last_run else 0.0 @@ -237,6 +244,8 @@ class PlanUser(FastHttpUser): if _PROGRESS.drained: _quit(self) return + if item.send_at_s is not None: + gevent.sleep(max(0.0, _RUN_STARTED + item.send_at_s - time.monotonic())) started_at: Final = time.time() started: Final = time.perf_counter() first_byte_s: float | None = None diff --git a/tests/e2e/perf/setup/observe.py b/tests/e2e/perf/setup/observe.py index 1532a78606e..dccc6e272b3 100644 --- a/tests/e2e/perf/setup/observe.py +++ b/tests/e2e/perf/setup/observe.py @@ -7,14 +7,16 @@ import hashlib import json import statistics from collections import Counter, defaultdict -from collections.abc import Iterable +from collections.abc import Iterable, Mapping from dataclasses import asdict, dataclass -from itertools import chain +from itertools import chain, groupby +from types import MappingProxyType from typing import Final, Literal from e2e_http import NoBody, get_external, post_external, unwrap from memory_readings import RssCapture, RssReading, WorkerKey from pydantic import BaseModel +from typing_extensions import ReadOnly, TypedDict from setup.datastore import ( DbCounters, @@ -27,7 +29,9 @@ from setup.datastore import ( statement_deltas, ) from setup.mocks.reply_common import Provider -from setup.performance_test_case import Limit, Metric, MockLatency, PerformanceTestCase +from setup.performance_test_case import Limit, Metric, MockLatency, PerformanceTestCase, Ramp +from setup.ramp import StepReading, render_steps, sustained_rps +from setup.series import Series from setup.telemetry import GatewayTelemetry from setup.wire import WireRecord @@ -122,6 +126,10 @@ class Observation: telemetry_before: GatewayTelemetry | None = None telemetry_after: GatewayTelemetry | None = None peak_connections: int | None = None + ramp: Ramp | None = None + cpu_series: Series[float] | None = None + connection_series: Series[int] | None = None + telemetry_series: Series[GatewayTelemetry] | None = None @property def telemetry(self) -> GatewayTelemetry | None: @@ -239,6 +247,67 @@ class Observation: post: Final = Counter(call.marker for call in self.guardrail_calls if call.input_type == "response") return sum(1 for record in self.served if pre[record.marker] != 1 or post[record.marker] != 1) + def steps(self) -> tuple[StepReading, ...]: + """One reading per step of a ramped run: the gateway's added latency over the step's requests, the rate + at which requests completed inside the step's window, the step's unexpected statuses, and what the pod + was doing in that window (CPU per completed request, Redis latency, Postgres connections).""" + ramp: Final = self.ramp + if ramp is None or not self.records: + return () + started: Final = min(record.started_at for record in self.records) + added: Final = _by_step(ramp, self.latency_pairs("gateway_added_seconds")) + first_token: Final = _by_step(ramp, self.latency_pairs("time_to_first_token_seconds")) + completed: Final = Counter( + int((record.started_at + record.total_s - started) // ramp.step_seconds) for record in self.records + ) + unexpected: Final = Counter( + ramp.step_of(record.seq) for record in self.records if not record.blocked and record.status != 200 + ) + return tuple( + StepReading( + rps=rate, + planned=count, + completed_rps=completed[index] / ramp.step_seconds, + gateway_added_p99=percentile(added.get(index, ()), 99), + unexpected=unexpected[index], + held_p99_seconds=ramp.held_p99_seconds, + gateway_added_p50=percentile(added.get(index, ()), 50), + time_to_first_token_p99=percentile(first_token.get(index, ()), 99), + **self._window_readings(started + index * ramp.step_seconds, ramp.step_seconds, completed[index]), + ) + for index, (rate, count) in enumerate(zip(ramp.rates, ramp.counts, strict=True)) + ) + + def _window_readings(self, start: float, seconds: float, completed: int) -> WindowReadings: + end: Final = start + seconds + cpu: Final = None if self.cpu_series is None else sum(self.cpu_series.between(start, end)) + conns: Final = None if self.connection_series is None else self.connection_series.between(start, end) + before: Final = None if self.telemetry_series is None else self.telemetry_series.last_before(start) + after: Final = None if self.telemetry_series is None else self.telemetry_series.last_before(end) + return WindowReadings( + cpu_seconds_per_request=None if cpu is None or completed == 0 else cpu / completed, + redis_latency_mean_seconds=None + if before is None or after is None + else after.minus(before).redis_latency.mean, + postgres_peak_connections=max(conns, default=None) if conns else None, + ) + + +class WindowReadings(TypedDict): + cpu_seconds_per_request: ReadOnly[float | None] + redis_latency_mean_seconds: ReadOnly[float | None] + postgres_peak_connections: ReadOnly[int | None] + + +def _by_step(ramp: Ramp, pairs: Iterable[tuple[WireRecord, float]]) -> Mapping[int, tuple[float, ...]]: + ordered: Final = sorted(pairs, key=lambda pair: pair[0].seq) + return MappingProxyType( + { + step: tuple(value for _, value in group) + for step, group in groupby(ordered, key=lambda pair: ramp.step_of(pair[0].seq)) + } + ) + def _text_sha1(text: str, *, intervened: bool) -> str: original: Final = text.removesuffix(" [post]") if intervened else text @@ -415,6 +484,8 @@ def observe( if metric == "cpu_seconds_per_request": cpu: Final = observation.gateway_cpu_seconds return None if cpu is None or not observation.records else cpu / len(observation.records) + if metric == "sustained_rps": + return None if observation.ramp is None else sustained_rps(observation.steps()) return worst_worker_growth_mb(observation.rss_before, observation.rss_after) @@ -424,6 +495,7 @@ class Row: percentile: int | None observed: float | None limit: float | None + floor: float | None = None detail: str = "" breakdown: tuple[Contributor, ...] = () @@ -437,11 +509,22 @@ class Row: @property def over(self) -> bool: - return self.limit is not None and self.observed is not None and self.observed > self.limit + """Past the limit: above a max, or below a floor.""" + if self.observed is None: + return False + return (self.limit is not None and self.observed > self.limit) or ( + self.floor is not None and self.observed < self.floor + ) @property def missing(self) -> bool: - return self.limit is not None and self.observed is None + return (self.limit is not None or self.floor is not None) and self.observed is None + + @property + def bound(self) -> str: + if self.limit is not None: + return f"{self.limit:.4f}" + return "" if self.floor is None else f">= {self.floor:.4f}" REPORTED: Final[tuple[tuple[Metric, int | None], ...]] = ( @@ -476,6 +559,8 @@ def evaluate(case: PerformanceTestCase, observation: Observation) -> tuple[list[ (limit.metric, limit.percentile): limit for limit in case.limits } keys: list[tuple[Metric, int | None]] = list(REPORTED) + if observation.ramp is not None: + keys.append(("sustained_rps", None)) keys.extend(key for key in limits if key not in keys) rows: list[Row] = [] for metric, rank in keys: @@ -486,20 +571,23 @@ def evaluate(case: PerformanceTestCase, observation: Observation) -> tuple[list[ percentile=rank, observed=observe(metric, rank, observation, case.background_cap_exempt_tables), limit=limit.max if limit else None, + floor=limit.min if limit else None, detail=_detail(metric, observation, case.background_cap_exempt_tables), breakdown=breakdown(metric, rank, observation, case.background_cap_exempt_tables), ) ) - exceeded: Final = [ - f"{row.label}: observed {row.observed:.4f} > limit {row.limit:.4f}{f' ({row.detail})' if row.detail else ''}" - if row.over and row.observed is not None and row.limit is not None - else f"{row.label}: limit set but nothing observed" - for row in rows - if row.over or row.missing - ] + exceeded: Final = [_breach(row) for row in rows if row.over or row.missing] return rows, exceeded +def _breach(row: Row) -> str: + if row.observed is None: + return f"{row.label}: limit set but nothing observed" + if row.floor is not None: + return f"{row.label}: observed {row.observed:.4f} < floor {row.floor:.4f}" + return f"{row.label}: observed {row.observed:.4f} > limit {row.bound}{f' ({row.detail})' if row.detail else ''}" + + def _detail(metric: Metric, observation: Observation, exempt_tables: frozenset[str]) -> str: if metric == "postgres_max_statement_per_request": busiest: Final = observation.busiest_awaited_statement @@ -516,12 +604,13 @@ def _detail(metric: Metric, observation: Observation, exempt_tables: frozenset[s def _row_properties(row: Row) -> tuple[tuple[str, float | str], ...]: observed: Final = ((row.property_name, row.observed),) if row.observed is not None else () limit: Final = ((f"{row.property_name}_limit", row.limit),) if row.limit is not None else () + floor: Final = ((f"{row.property_name}_floor", row.floor),) if row.floor is not None else () contributors: Final = ( ((f"{row.property_name}_breakdown", json.dumps([asdict(item) for item in row.breakdown])),) if row.breakdown else () ) - return observed + limit + contributors + return observed + limit + floor + contributors def junit_properties(rows: Iterable[Row]) -> tuple[tuple[str, float | str], ...]: @@ -546,9 +635,17 @@ def render(case: PerformanceTestCase, observation: Observation, rows: list[Row]) ] for row in rows: observed = "n/a" if row.observed is None else f"{row.observed:.4f}" - limit = "" if row.limit is None else f"{row.limit:.4f}" - flag = " OVER" if row.over else " MISSING" if row.missing else "" - lines.append(f" {row.label:32} {observed:>12} {limit:>12}{flag}") + flag = ( + " UNDER" + if row.over and row.floor is not None + else " OVER" + if row.over + else " MISSING" + if row.missing + else "" + ) + lines.append(f" {row.label:32} {observed:>12} {row.bound:>12}{flag}") + lines.extend(render_steps(observation.steps())) lines.extend(_diagnostics(observation)) lines.extend(_datastore_lines(observation)) for record in failures: diff --git a/tests/e2e/perf/setup/performance_test_case.py b/tests/e2e/perf/setup/performance_test_case.py index 3f39b4c38c8..512ef8b911f 100644 --- a/tests/e2e/perf/setup/performance_test_case.py +++ b/tests/e2e/perf/setup/performance_test_case.py @@ -9,8 +9,10 @@ so a case is a literal a reviewer can read top to bottom. from __future__ import annotations +import bisect from collections.abc import Mapping -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace +from itertools import accumulate from typing import Final, Literal from setup.event_loop import EventLoopStalls @@ -36,6 +38,7 @@ Metric = Literal[ "postgres_active_sessions", "postgres_slowest_statement_seconds", "postgres_peak_connections", + "sustained_rps", ] @@ -97,12 +100,50 @@ SHORT: Final = RequestShape("short", input_tokens=(5_000, 100_000), tools=True) LONG: Final = RequestShape("long", input_tokens=(300_000, 1_000_000), tools=True) +@dataclass(frozen=True, slots=True) +class Ramp: + """Rate stepped up through the run: `first_rps`, then `step_rps` more every `step_seconds`, up to `last_rps`. + Each step is scored on its own and the run reports `sustained_rps`, the last step the pod held before the + first one it did not: a step holds when its `gateway_added_seconds` p99 stayed under `held_p99_seconds`, it + completed at least `HELD_SHARE` of its rate and no more than `HELD_UNEXPECTED_SHARE` of its requests came back + with an unexpected status (`setup/ramp.py`). Requests are sent on a schedule drawn from the plan's seed, not + from per-user intervals, so the rate changes on the step boundary.""" + + first_rps: float + last_rps: float + step_rps: float + step_seconds: float + held_p99_seconds: float + + @property + def rates(self) -> tuple[float, ...]: + count: Final = round((self.last_rps - self.first_rps) / self.step_rps) + 1 + return tuple(self.first_rps + index * self.step_rps for index in range(count)) + + @property + def counts(self) -> tuple[int, ...]: + return tuple(round(rate * self.step_seconds) for rate in self.rates) + + @property + def total_requests(self) -> int: + return sum(self.counts) + + @property + def seconds(self) -> float: + return len(self.rates) * self.step_seconds + + def step_of(self, seq: int) -> int: + """Zero-based step of the request with 1-based plan number `seq`.""" + return bisect.bisect_left(tuple(accumulate(self.counts)), seq) + + @dataclass(frozen=True, slots=True) class Traffic: """The plan: `total_requests` drawn from `seed`, sent at `rps` by `concurrency` simulated - users. Shares are fractions of the requests: `endpoints` and `shapes` must each sum to 1, - `intervene_share` picks non-streaming requests the guardrail rewrites, `block_share` picks - requests it blocks.""" + users, or stepped up through a `ramp` (then `rps` is its first step and `total_requests` its + total, see `ramped`). Shares are fractions of the requests: `endpoints` and `shapes` must each + sum to 1, `intervene_share` picks non-streaming requests the guardrail rewrites, `block_share` + picks requests it blocks.""" seed: int rps: float @@ -113,20 +154,33 @@ class Traffic: intervene_share: float = 0.0 block_share: float = 0.0 concurrency: int = 20 + ramp: Ramp | None = None @property def expected_seconds(self) -> float: - return self.total_requests / self.rps + return self.total_requests / self.rps if self.ramp is None else self.ramp.seconds + + +def ramped(base: Traffic, ramp: Ramp, *, seed: int, concurrency: int) -> Traffic: + """`base`'s request mix sent along `ramp` instead of at one rate. `concurrency` caps the requests in + flight, so size it past the deepest step the ramp is expected to reach.""" + return replace( + base, seed=seed, rps=ramp.first_rps, total_requests=ramp.total_requests, concurrency=concurrency, ramp=ramp + ) @dataclass(frozen=True, slots=True) class Limit: - """One pass/fail rule: `metric` (at `percentile`, or the whole-run value) must not exceed `max`. - Metrics without a Limit are still reported.""" + """One pass/fail rule: `metric` (at `percentile`, or the whole-run value) must not exceed `max`, or must + reach `min` for a floor such as `sustained_rps`. Metrics without a Limit are still reported.""" metric: Metric - max: float + max: float | None = None percentile: int | None = None + min: float | None = None + + def __post_init__(self) -> None: + assert (self.max is None) != (self.min is None), f"{self.metric}: a Limit is either a max or a min" @property def label(self) -> str: @@ -201,6 +255,14 @@ class PerformanceTestCase: assert 0 <= self.traffic.stream_share <= 1, f"{self.name}: stream_share out of range" assert self.traffic.intervene_share + self.traffic.block_share <= 1, f"{self.name}: guardrail shares exceed 1" assert self.traffic.total_requests > 0 and self.traffic.rps > 0, f"{self.name}: empty traffic" + ramp: Final = self.traffic.ramp + assert ramp is None or (ramp.first_rps, ramp.total_requests) == ( + self.traffic.rps, + self.traffic.total_requests, + ), f"{self.name}: traffic rps and total_requests must be the ramp's, build it with ramped()" + assert ramp is None or (0 < ramp.first_rps <= ramp.last_rps and ramp.step_rps > 0 and ramp.step_seconds > 0), ( + f"{self.name}: ramp {ramp} does not step up" + ) assert self.model_groups, f"{self.name}: no model groups" assert 0 <= self.mock_latency.jitter_seconds <= self.mock_latency.seconds or self.mock_latency.seconds == 0, ( f"{self.name}: mock latency jitter {self.mock_latency.jitter_seconds} s exceeds {self.mock_latency.seconds} s" diff --git a/tests/e2e/perf/setup/plan.py b/tests/e2e/perf/setup/plan.py index 6b9a5a19662..4f28736a8e9 100644 --- a/tests/e2e/perf/setup/plan.py +++ b/tests/e2e/perf/setup/plan.py @@ -12,6 +12,7 @@ from dataclasses import dataclass from typing import Final, Literal, TypeVar from setup.performance_test_case import Endpoint, PerformanceTestCase, RequestShape, Team +from setup.ramp import send_offsets Flag = Literal["n", "i", "b"] CallerKind = Literal["jwt", "key"] @@ -39,6 +40,7 @@ class PlannedRequest: flag: Flag shape: RequestShape input_tokens: int + send_at_s: float | None = None @property def blocked(self) -> bool: @@ -98,6 +100,7 @@ def build_plan(case: PerformanceTestCase) -> tuple[PlannedRequest, ...]: case.validate() rng: Final = random.Random(case.traffic.seed) callers: Final = _callers_by_group(case) + offsets: Final = None if case.traffic.ramp is None else send_offsets(case.traffic.ramp, case.traffic.seed) planned: list[PlannedRequest] = [] for seq in range(1, case.traffic.total_requests + 1): endpoint = _pick(rng, case.traffic.endpoints) @@ -117,6 +120,7 @@ def build_plan(case: PerformanceTestCase) -> tuple[PlannedRequest, ...]: flag=flag, shape=shape, input_tokens=rng.randint(low, high), + send_at_s=None if offsets is None else offsets[seq - 1], ) ) return tuple(planned) diff --git a/tests/e2e/perf/setup/podcpu.py b/tests/e2e/perf/setup/podcpu.py index 400e5b7d38b..865fa14efe0 100644 --- a/tests/e2e/perf/setup/podcpu.py +++ b/tests/e2e/perf/setup/podcpu.py @@ -19,6 +19,8 @@ from urllib.parse import quote from e2e_http import AuthHeaders, Success, get_external from pydantic import BaseModel +from setup.series import Recorder, Series + SELECTOR_ENV: Final = "E2E_PERF_GATEWAY_POD_SELECTOR" _SERVICE_ACCOUNT: Final = Path("/var/run/secrets/kubernetes.io/serviceaccount") _INTERVAL_S: Final = 5.0 @@ -49,7 +51,8 @@ def cores(quantity: str) -> float: class GatewayCpuSampler: - """Integrates the matching pods' core use over the time it is running; `seconds` is CPU seconds.""" + """Integrates the matching pods' core use over the time it is running; `seconds` is CPU seconds and + `series` the CPU seconds of each sampling interval, stamped when the interval ended.""" def __init__(self, selector: str, namespace: str, token: str, ca_bundle: str) -> None: self._url: Final = ( @@ -60,12 +63,15 @@ class GatewayCpuSampler: self._ca_bundle: Final = ca_bundle self._stop: Final = threading.Event() self._thread: Final = threading.Thread(target=self._run, name="gateway-cpu-sampler", daemon=True) - self._cpu_seconds = 0.0 - self._samples = 0 + self._intervals: Final[Recorder[float]] = Recorder() @property def seconds(self) -> float | None: - return self._cpu_seconds if self._samples > 0 else None + return sum(sample.value for sample in self._intervals.series.samples) if len(self._intervals) else None + + @property + def series(self) -> Series[float]: + return self._intervals.series def _read_cores(self) -> float | None: result: Final = get_external( @@ -81,8 +87,7 @@ class GatewayCpuSampler: now: float = time.monotonic() current: float | None = self._read_cores() if current is not None: - self._cpu_seconds += current * (now - last) - self._samples += 1 + self._intervals.record(current * (now - last)) last = now # rebind-ok: the sampling clock advances every tick def start(self) -> None: @@ -95,6 +100,7 @@ class GatewayCpuSampler: class NoSampler: seconds: Final[None] = None + series: Final[None] = None def start(self) -> None: return None diff --git a/tests/e2e/perf/setup/ramp.py b/tests/e2e/perf/setup/ramp.py new file mode 100644 index 00000000000..217e140d470 --- /dev/null +++ b/tests/e2e/perf/setup/ramp.py @@ -0,0 +1,116 @@ +"""A ramped run step by step: the send schedule its plan carries (`send_offsets`), one `StepReading` per step +and `sustained_rps`, the last step the pod held before the first it did not. `Ramp` itself sits in +`performance_test_case.py` with the rest of the case schema.""" + +from __future__ import annotations + +import random +from collections.abc import Iterable, Iterator +from dataclasses import dataclass +from itertools import chain, takewhile +from typing import Final + +from setup.performance_test_case import Ramp + +HELD_SHARE: Final = 0.9 +"""A step that completed fewer than this share of its rate is a queue that stopped draining.""" +HELD_UNEXPECTED_SHARE: Final = 0.01 +"""A step with more than this share of unexpected statuses broke, whatever its latency.""" +_JITTER: Final = 0.25 + + +def _step_offsets(ramp: Ramp, index: int, count: int, rng: random.Random) -> Iterator[float]: + for slot in range(count): + yield (index + (slot + 0.5 + rng.uniform(-_JITTER, _JITTER)) / count) * ramp.step_seconds + + +def send_offsets(ramp: Ramp, seed: int) -> tuple[float, ...]: + """Seconds after the run starts to send each request of the plan, in plan order: a step's requests spread + evenly over its window, each moved by up to a quarter of a slot either way, drawn from `seed` so a rerun + sends at the same seconds.""" + rng: Final = random.Random(f"ramp-{seed}") + return tuple(chain.from_iterable(_step_offsets(ramp, index, count, rng) for index, count in enumerate(ramp.counts))) + + +@dataclass(frozen=True, slots=True) +class StepReading: + rps: float + planned: int + completed_rps: float + gateway_added_p99: float | None + unexpected: int + held_p99_seconds: float + gateway_added_p50: float | None = None + time_to_first_token_p99: float | None = None + cpu_seconds_per_request: float | None = None + redis_latency_mean_seconds: float | None = None + postgres_peak_connections: int | None = None + """The hold verdict reads the first six. The rest say what the pod was doing on that step: CPU per request + from the pod sampler, Redis latency from the gateway's `/metrics`, connections from `pg_stat_activity`, + all cut to the step's window, so a latency jump can be told apart as CPU, waiting or neither.""" + + @property + def held(self) -> bool: + return ( + self.gateway_added_p99 is not None + and self.gateway_added_p99 <= self.held_p99_seconds + and self.completed_rps >= HELD_SHARE * self.rps + and self.unexpected <= HELD_UNEXPECTED_SHARE * self.planned + ) + + +def sustained_rps(steps: Iterable[StepReading]) -> float: + """The rate of the last step that held before the first that did not, 0 when the first step did not hold.""" + held: Final = tuple(takewhile(lambda step: step.held, steps)) + return held[-1].rps if held else 0.0 + + +def step_properties(steps: Iterable[StepReading]) -> tuple[tuple[str, float], ...]: + """The per-step readings as JUnit properties (`perf_ramp_step__`, every reading the step has) + plus the hold bound, so the dashboard can draw the run's curves next to the sustained rate.""" + rows: Final = tuple(steps) + hold: Final = (("perf_ramp_hold_p99_seconds", rows[0].held_p99_seconds),) if rows else () + return (*hold, *chain.from_iterable(_step_properties(step) for step in rows)) + + +def _step_properties(step: StepReading) -> Iterator[tuple[str, float]]: + prefix: Final = f"perf_ramp_step_{step.rps:g}" + readings: Final[tuple[tuple[str, float | None], ...]] = ( + ("gateway_added_p99", step.gateway_added_p99), + ("completed_rps", step.completed_rps), + ("unexpected", float(step.unexpected)), + ("gateway_added_p50", step.gateway_added_p50), + ("time_to_first_token_p99", step.time_to_first_token_p99), + ("cpu_seconds_per_request", step.cpu_seconds_per_request), + ("redis_latency_mean_seconds", step.redis_latency_mean_seconds), + ( + "postgres_peak_connections", + None if step.postgres_peak_connections is None else float(step.postgres_peak_connections), + ), + ) + return ((f"{prefix}_{name}", value) for name, value in readings if value is not None) + + +def render_steps(steps: Iterable[StepReading]) -> tuple[str, ...]: + rows: Final = tuple(steps) + if not rows: + return () + header: Final = ( + f" steps: {'rps':>6} {'planned':>8} {'completed rps':>14} {'gateway added p50':>18} {'p99':>8}" + f" {'ttft p99':>9} {'unexpected':>11} {'cpu s/req':>10} {'redis mean':>11} {'pg conns':>9}" + ) + return (header, *(_step_line(step) for step in rows)) + + +def _seconds(value: float | None) -> str: + return "n/a" if value is None else f"{value:.4f}" + + +def _step_line(step: StepReading) -> str: + conns: Final = "n/a" if step.postgres_peak_connections is None else str(step.postgres_peak_connections) + return ( + f" {step.rps:6.1f} {step.planned:8d} {step.completed_rps:14.2f} {_seconds(step.gateway_added_p50):>18}" + f" {_seconds(step.gateway_added_p99):>8} {_seconds(step.time_to_first_token_p99):>9} {step.unexpected:11d}" + f" {_seconds(step.cpu_seconds_per_request):>10} {_seconds(step.redis_latency_mean_seconds):>11} {conns:>9}" + f" {'held' if step.held else 'broke'}" + ) diff --git a/tests/e2e/perf/setup/run.py b/tests/e2e/perf/setup/run.py index 272c64754c4..ec2927e90af 100644 --- a/tests/e2e/perf/setup/run.py +++ b/tests/e2e/perf/setup/run.py @@ -63,6 +63,7 @@ def wire_request(item: PlannedRequest, setup: CaseSetup) -> WireRequest: token_source=setup.token_source(item.caller), stream=item.stream, blocked=item.blocked, + send_at_s=item.send_at_s, ) diff --git a/tests/e2e/perf/setup/series.py b/tests/e2e/perf/setup/series.py new file mode 100644 index 00000000000..78e5cf272e1 --- /dev/null +++ b/tests/e2e/perf/setup/series.py @@ -0,0 +1,48 @@ +"""Readings a background sampler takes while the traffic runs, each stamped with the wall clock +(`time.time()`, the clock the Locust records carry in `started_at`), so a ramped run can cut them into its +steps' windows after the fact.""" + +from __future__ import annotations + +import time +from dataclasses import dataclass +from typing import Final, Generic, TypeVar + +T = TypeVar("T") + + +@dataclass(frozen=True, slots=True) +class Sample(Generic[T]): + at: float + value: T + + +@dataclass(frozen=True, slots=True) +class Series(Generic[T]): + samples: tuple[Sample[T], ...] + + def between(self, start: float, end: float) -> tuple[T, ...]: + """The values stamped inside `[start, end)`, in time order.""" + return tuple(sample.value for sample in self.samples if start <= sample.at < end) + + def last_before(self, at: float) -> T | None: + """The newest value stamped before `at`, None when the series starts later.""" + earlier: Final = tuple(sample.value for sample in self.samples if sample.at < at) + return earlier[-1] if earlier else None + + +class Recorder(Generic[T]): + """What a sampler thread appends to; `series` is a frozen copy for the observation.""" + + def __init__(self) -> None: + self._samples: list[Sample[T]] = [] + + def record(self, value: T) -> None: + self._samples.append(Sample(at=time.time(), value=value)) + + @property + def series(self) -> Series[T]: + return Series(samples=tuple(self._samples)) + + def __len__(self) -> int: + return len(self._samples) diff --git a/tests/e2e/perf/setup/telemetry.py b/tests/e2e/perf/setup/telemetry.py index e52bcabafda..185cc788601 100644 --- a/tests/e2e/perf/setup/telemetry.py +++ b/tests/e2e/perf/setup/telemetry.py @@ -1,5 +1,5 @@ -"""The gateway's own `/metrics`, read before and after a run for the numbers nothing outside the gateway -can see. Today that is `litellm_redis_latency`, the histogram customers' Grafana panels show when Redis +"""The gateway's own `/metrics`, read before and after a run (and every few seconds during a ramped run, see +`telemetry_sampler`) for the numbers nothing outside the gateway can see. Today that is `litellm_redis_latency`, the histogram customers' Grafana panels show when Redis slows the gateway down. The gateway's latency and request counts are read per request from its response headers instead (`setup/wire.py`), which gives percentiles where a histogram delta only gives a mean. @@ -10,7 +10,9 @@ on psutil (`proxy_usage.py`).""" from __future__ import annotations import math -from collections.abc import Iterable +import threading +from collections.abc import Generator, Iterable +from contextlib import contextmanager from dataclasses import dataclass from itertools import chain from typing import Final @@ -20,6 +22,10 @@ from prometheus_client.parser import text_string_to_metric_families from prometheus_client.samples import Sample from proxy_client import ProxyClient +from setup.series import Recorder, Series + +_SAMPLE_SECONDS: Final = 10.0 + @dataclass(frozen=True, slots=True) class HistogramTotals: @@ -63,3 +69,42 @@ def read_gateway_telemetry(proxy: ProxyClient) -> GatewayTelemetry | None: case every limit on a gateway-reported metric fails as missing rather than reading a silent zero.""" scrape: Final = proxy.probe("/metrics", params=NoBody()) return parse_gateway_telemetry(scrape.body) if scrape.status_code == 200 else None + + +class TelemetrySampler: + """`/metrics` every `_SAMPLE_SECONDS` on a background thread while traffic runs, so a ramped run can read + the Redis latency of each step's window as the difference between the scrapes around it.""" + + def __init__(self, proxy: ProxyClient) -> None: + self._proxy: Final = proxy + self._stop: Final = threading.Event() + self._thread: Final = threading.Thread(target=self._run, name="gateway-telemetry", daemon=True) + self._scrapes: Final[Recorder[GatewayTelemetry]] = Recorder() + + @property + def series(self) -> Series[GatewayTelemetry]: + return self._scrapes.series + + def _run(self) -> None: + while not self._stop.is_set(): + scrape: GatewayTelemetry | None = read_gateway_telemetry(self._proxy) + if scrape is not None: + self._scrapes.record(scrape) + _ = self._stop.wait(_SAMPLE_SECONDS) + + def start(self) -> None: + self._thread.start() + + def stop(self) -> None: + self._stop.set() + self._thread.join(timeout=30) + + +@contextmanager +def telemetry_sampler(proxy: ProxyClient) -> Generator[TelemetrySampler]: + sampler: Final = TelemetrySampler(proxy) + sampler.start() + try: + yield sampler + finally: + sampler.stop() diff --git a/tests/e2e/perf/setup/upgrade.py b/tests/e2e/perf/setup/upgrade.py index 36efc16e1fe..c9114ccbff2 100644 --- a/tests/e2e/perf/setup/upgrade.py +++ b/tests/e2e/perf/setup/upgrade.py @@ -145,7 +145,9 @@ class UpgradeMonitor: self.clock: Final = clock self.wall_clock: Final = wall_clock self.started: Final = clock() - self.status_limit: Final = next(limit.max for limit in CASE.limits if limit.metric == "unexpected_status_count") + self.status_limit: Final = next( + limit.max for limit in CASE.limits if limit.metric == "unexpected_status_count" and limit.max is not None + ) self.pending: tuple[WireRecord, ...] = () self.processed = 0 self.successes = 0 diff --git a/tests/e2e/perf/setup/wire.py b/tests/e2e/perf/setup/wire.py index 18351c18c6b..a1833fbd44b 100644 --- a/tests/e2e/perf/setup/wire.py +++ b/tests/e2e/perf/setup/wire.py @@ -51,6 +51,8 @@ class WireRequest(BaseModel): token_source: TokenSource | None = None stream: bool blocked: bool + send_at_s: float | None = None + """Seconds after the run starts to send this request, set when the case ramps; otherwise the users pace.""" class WireRecord(BaseModel): diff --git a/tests/e2e/perf/test_perf_cases.py b/tests/e2e/perf/test_perf_cases.py index 3e5c3c522c8..db657a04b00 100644 --- a/tests/e2e/perf/test_perf_cases.py +++ b/tests/e2e/perf/test_perf_cases.py @@ -44,8 +44,9 @@ from setup.performance_test_case import MockLatency, PerformanceTestCase from setup.plan import PlannedRequest, build_plan from setup.podcpu import gateway_cpu_sampler from setup.provision import MOCK_PROVIDERS, build_setup, guardrail_url, mock_url +from setup.ramp import step_properties from setup.run import run_plan -from setup.telemetry import read_gateway_telemetry +from setup.telemetry import read_gateway_telemetry, telemetry_sampler pytestmark = [pytest.mark.perf, pytest.mark.quiet_stack] _SPEND_SAMPLE: Final = 40 @@ -115,7 +116,12 @@ def test_perf_case( redis_before: Final = read_redis_counters() run_started: Final = time.monotonic() traffic_started_at: Final = datetime.now(tz=UTC) - with gateway_cpu_sampler() as cpu, redis_monitor() as monitor, connections_sampler() as connections: + with ( + gateway_cpu_sampler() as cpu, + redis_monitor() as monitor, + connections_sampler() as connections, + telemetry_sampler(proxy) as telemetry, + ): outcome: Final = run_plan(case, plan, setup) run_finished: Final = time.monotonic() db_after: Final = read_db_counters() @@ -151,6 +157,10 @@ def test_perf_case( telemetry_before=telemetry_before, telemetry_after=telemetry_after, peak_connections=connections.peak, + ramp=case.traffic.ramp, + cpu_series=cpu.series, + connection_series=connections.series, + telemetry_series=telemetry.series, ) rows, exceeded_limits = evaluate(case, observation) stall_rows: Final = _event_loop_rows(case, traffic_started_at) if case.event_loop is not None else () @@ -162,6 +172,7 @@ def test_perf_case( item.user_properties.extend( stall_properties(case.event_loop, stall_rows) if case.event_loop is not None else junit_properties(rows) ) + item.user_properties.extend(step_properties(observation.steps())) print(render(case, observation, rows)) if case.event_loop is not None: print(render_stalls(case.event_loop, stall_rows)) diff --git a/tests/e2e_harness/perf/setup/test_ramp.py b/tests/e2e_harness/perf/setup/test_ramp.py new file mode 100644 index 00000000000..6d6c1640b59 --- /dev/null +++ b/tests/e2e_harness/perf/setup/test_ramp.py @@ -0,0 +1,168 @@ +from __future__ import annotations + +import math +from dataclasses import replace +from typing import Final + +from cases.small_bodies_5rps import CASE +from memory_readings import RssCapture +from setup.observe import Observation, evaluate, junit_properties +from setup.performance_test_case import Limit, Ramp, ramped +from setup.plan import build_plan +from setup.ramp import StepReading, send_offsets, step_properties, sustained_rps +from setup.series import Sample, Series +from setup.telemetry import GatewayTelemetry, HistogramTotals +from setup.wire import WireRecord + +NO_RSS: Final = RssCapture(readings=(), failures=()) + + +def test_send_offsets_put_each_step_s_requests_inside_its_window_in_plan_order() -> None: + ramp: Final = Ramp(first_rps=10.0, last_rps=30.0, step_rps=10.0, step_seconds=10.0, held_p99_seconds=5.0) + offsets: Final = send_offsets(ramp, seed=13) + assert ramp.counts == (100, 200, 300) + assert len(offsets) == ramp.total_requests + assert offsets == tuple(sorted(offsets)) + for seq, offset in enumerate(offsets, start=1): + step = ramp.step_of(seq) + assert step * 10.0 <= offset < (step + 1) * 10.0, f"request {seq} at {offset:.2f}s is outside step {step}" + assert send_offsets(ramp, seed=13) == offsets + assert send_offsets(ramp, seed=14) != offsets + + +def test_build_plan_carries_the_schedule_only_for_a_ramped_case() -> None: + ramp: Final = Ramp(first_rps=2.0, last_rps=4.0, step_rps=2.0, step_seconds=5.0, held_p99_seconds=5.0) + ramped_case: Final = replace(CASE, traffic=ramped(CASE.traffic, ramp, seed=3, concurrency=20)) + assert ramped_case.traffic.expected_seconds == 10.0 + assert [item.send_at_s for item in build_plan(ramped_case)] == list(send_offsets(ramp, seed=3)) + assert all(item.send_at_s is None for item in build_plan(CASE)) + + +def _step(rps: float, p99: float, completed_rps: float | None = None, unexpected: int = 0) -> StepReading: + return StepReading( + rps=rps, + planned=int(rps * 90), + completed_rps=rps if completed_rps is None else completed_rps, + gateway_added_p99=p99, + unexpected=unexpected, + held_p99_seconds=5.0, + ) + + +def test_sustained_rps_is_the_last_step_held_before_the_first_that_broke() -> None: + assert sustained_rps((_step(10, 0.5), _step(20, 1.0), _step(30, 9.0), _step(40, 0.7))) == 20 + assert sustained_rps((_step(10, 0.5), _step(20, 1.0, completed_rps=17.0), _step(30, 1.0))) == 10 + assert sustained_rps((_step(10, 0.5), _step(20, 1.0, unexpected=19), _step(30, 1.0))) == 10 + assert sustained_rps((_step(10, 0.5), _step(20, 1.0, unexpected=18), _step(30, 1.0))) == 30 + assert sustained_rps((_step(10, 6.0), _step(20, 1.0))) == 0 + + +def _record(seq: int, started_at: float, total_s: float) -> WireRecord: + return WireRecord( + seq=seq, + marker=f"perfreq-{seq:06d}-n", + caller="key-a1", + endpoint="chat_completions", + stream=False, + blocked=False, + started_at=started_at, + status=200, + header_s=total_s, + first_byte_s=total_s, + first_token_s=None, + total_s=total_s, + text="", + ok=True, + mock_delay_s=1.0, + call_id=None, + error=None, + ) + + +def test_a_floor_fails_when_the_second_step_s_latency_breaks_it_and_the_properties_carry_the_floor() -> None: + ramp: Final = Ramp(first_rps=10.0, last_rps=20.0, step_rps=10.0, step_seconds=100.0, held_p99_seconds=2.0) + case: Final = replace( + CASE, + traffic=ramped(CASE.traffic, ramp, seed=1, concurrency=50), + limits=(Limit("sustained_rps", min=20),), + ) + offsets: Final = send_offsets(ramp, seed=1) + records: Final = tuple( + _record(seq, 1000.0 + offset, 1.5 if ramp.step_of(seq) == 0 else 4.0) + for seq, offset in enumerate(offsets, start=1) + ) + observation: Final = Observation( + records=records, + mock_requests=(), + guardrail_calls=(), + rss_before=NO_RSS, + rss_after=NO_RSS, + wall_seconds=200.0, + ramp=ramp, + ) + first, second = observation.steps() + assert first.held and first.gateway_added_p99 == 0.5 and first.completed_rps > 9.5 + assert not second.held and second.gateway_added_p99 == 3.0 + rows, exceeded = evaluate(case, observation) + assert exceeded == ["sustained_rps: observed 10.0000 < floor 20.0000"] + assert {("perf_sustained_rps", 10.0), ("perf_sustained_rps_floor", 20.0)} <= set(junit_properties(rows)) + assert dict(step_properties((first, second))) == { + "perf_ramp_hold_p99_seconds": 2.0, + "perf_ramp_step_10_gateway_added_p99": 0.5, + "perf_ramp_step_10_completed_rps": first.completed_rps, + "perf_ramp_step_10_unexpected": 0.0, + "perf_ramp_step_10_gateway_added_p50": 0.5, + "perf_ramp_step_20_gateway_added_p99": 3.0, + "perf_ramp_step_20_completed_rps": second.completed_rps, + "perf_ramp_step_20_unexpected": 0.0, + "perf_ramp_step_20_gateway_added_p50": 3.0, + } + assert not evaluate(replace(case, limits=(Limit("sustained_rps", min=10),)), observation)[1] + + +def test_each_step_reads_the_pod_s_cpu_redis_latency_and_connections_from_its_own_window() -> None: + ramp: Final = Ramp(first_rps=10.0, last_rps=20.0, step_rps=10.0, step_seconds=100.0, held_p99_seconds=2.0) + records: Final = tuple( + _record(seq, 1000.0 + offset, 1.5) for seq, offset in enumerate(send_offsets(ramp, seed=1), start=1) + ) + started: Final = min(record.started_at for record in records) + redis: Final = tuple( + Sample(at=started + at, value=GatewayTelemetry(redis_latency=HistogramTotals(total=total, count=count))) + for at, total, count in ((-1.0, 0.0, 0.0), (50.0, 1.0, 100.0), (99.0, 2.0, 200.0), (199.0, 42.0, 300.0)) + ) + observation: Final = Observation( + records=records, + mock_requests=(), + guardrail_calls=(), + rss_before=NO_RSS, + rss_after=NO_RSS, + wall_seconds=200.0, + ramp=ramp, + cpu_series=Series(samples=tuple(Sample(at=started + at, value=5.0) for at in (10.0, 60.0, 110.0, 160.0))), + connection_series=Series( + samples=tuple( + Sample(at=started + at, value=n) for at, n in ((1.0, 12), (50.0, 18), (150.0, 25), (199.0, 20)) + ) + ), + telemetry_series=Series(samples=redis), + ) + first, second = observation.steps() + first_completed: Final = round(first.completed_rps * ramp.step_seconds) + assert first.cpu_seconds_per_request is not None and math.isclose( + first.cpu_seconds_per_request, 10.0 / first_completed + ) + assert first.redis_latency_mean_seconds is not None and math.isclose(first.redis_latency_mean_seconds, 0.01) + assert first.postgres_peak_connections == 18 + assert second.redis_latency_mean_seconds is not None and math.isclose(second.redis_latency_mean_seconds, 0.4) + assert second.postgres_peak_connections == 25 + assert first.time_to_first_token_p99 is None + names: Final = {name for name, _ in step_properties((first, second)) if name.startswith("perf_ramp_step_10_")} + assert names == { + "perf_ramp_step_10_gateway_added_p99", + "perf_ramp_step_10_completed_rps", + "perf_ramp_step_10_unexpected", + "perf_ramp_step_10_gateway_added_p50", + "perf_ramp_step_10_cpu_seconds_per_request", + "perf_ramp_step_10_redis_latency_mean_seconds", + "perf_ramp_step_10_postgres_peak_connections", + } diff --git a/tests/e2e_harness/perf/test_upgrade.py b/tests/e2e_harness/perf/test_upgrade.py index 9a6c9d2af14..3414476ba4c 100644 --- a/tests/e2e_harness/perf/test_upgrade.py +++ b/tests/e2e_harness/perf/test_upgrade.py @@ -304,7 +304,9 @@ def test_invalid_200_is_not_an_allowed_non_200(failed: WireRecord) -> None: def _ceiling(metric: str, percentile: int | None = None) -> float: return next( - limit.max for limit in upgrade_case().limits if limit.metric == metric and limit.percentile == percentile + limit.max + for limit in upgrade_case().limits + if limit.metric == metric and limit.percentile == percentile and limit.max is not None )