mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-27 06:13:07 +00:00
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 <kerry@berri.ai> Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Kerry Lu
GitHub
parent
159b5b2f72
commit
b99d2cc06e
18 changed files with 677 additions and 44 deletions
File diff suppressed because one or more lines are too long
24
tests/e2e/perf/cases/mixed_size_requests_ramp.py
Normal file
24
tests/e2e/perf/cases/mixed_size_requests_ramp.py
Normal file
|
|
@ -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),),
|
||||
)
|
||||
24
tests/e2e/perf/cases/small_bodies_ramp.py
Normal file
24
tests/e2e/perf/cases/small_bodies_ramp.py
Normal file
|
|
@ -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),),
|
||||
)
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
116
tests/e2e/perf/setup/ramp.py
Normal file
116
tests/e2e/perf/setup/ramp.py
Normal file
|
|
@ -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_<rps>_<reading>`, 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'}"
|
||||
)
|
||||
|
|
@ -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,
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
48
tests/e2e/perf/setup/series.py
Normal file
48
tests/e2e/perf/setup/series.py
Normal file
|
|
@ -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)
|
||||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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):
|
||||
|
|
|
|||
|
|
@ -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))
|
||||
|
|
|
|||
168
tests/e2e_harness/perf/setup/test_ramp.py
Normal file
168
tests/e2e_harness/perf/setup/test_ramp.py
Normal file
|
|
@ -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",
|
||||
}
|
||||
|
|
@ -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
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue