From 967026feefc086ed0618c8a4b0328ecad5f4953b Mon Sep 17 00:00:00 2001 From: Nate Armstrong Date: Tue, 20 Oct 2026 19:36:28 -0700 Subject: [PATCH] feat(loadtest): sustained-rate workloads at 100, 500 and 1,000 rps (#47719) Evenly spaced arrivals ("arrivals": "constant"), per-request lag_ms so a run shows when the generator falls behind, and an s3stub that stores objects compressed and lists from a sorted index, so a 15-minute run at 1,000 rps per gateway fits in memory and verify can list it. --- litellm-rust/loadtest/README.md | 9 ++- litellm-rust/loadtest/cmd/loadgen/main.go | 16 +++- litellm-rust/loadtest/cmd/loadgen/workload.go | 9 ++- litellm-rust/loadtest/cmd/s3stub/main.go | 76 +++++++++++++------ litellm-rust/loadtest/cmd/s3stub/main_test.go | 73 +++++++++++++++++- .../loadtest/workloads/sustained-100.json | 20 +++++ .../loadtest/workloads/sustained-1000.json | 20 +++++ .../loadtest/workloads/sustained-500.json | 20 +++++ .../loadtest/workloads/sustained-check.json | 20 +++++ 9 files changed, 231 insertions(+), 32 deletions(-) create mode 100644 litellm-rust/loadtest/workloads/sustained-100.json create mode 100644 litellm-rust/loadtest/workloads/sustained-1000.json create mode 100644 litellm-rust/loadtest/workloads/sustained-500.json create mode 100644 litellm-rust/loadtest/workloads/sustained-check.json diff --git a/litellm-rust/loadtest/README.md b/litellm-rust/loadtest/README.md index 9c96c0ce4d0..52e3e10157b 100644 --- a/litellm-rust/loadtest/README.md +++ b/litellm-rust/loadtest/README.md @@ -7,16 +7,19 @@ machines. | Tool | What it does | | --- | --- | | `mockllm` | Fake OpenAI chat completions, Anthropic messages and Bedrock invoke upstream. Streams or not, and takes its timing from a directive in the prompt (`[[mock ttft_ms=800 dur_ms=300000 out=2000]]`). Counts hits per deployment (`/d//` path prefix) at `GET /stats`; `-faults "chat-0-3=429:1.0"` makes chosen deployments fail. | -| `loadgen` | Open-loop Poisson arrivals from a workload file: weighted request classes with their own endpoint, stream share, context size (10k to 1M tokens), time to first token and total time. Writes one JSON line per request. Shard across machines with `-shard i -shards n`. | +| `loadgen` | Open-loop arrivals from a workload file, Poisson or evenly spaced (`"arrivals": "constant"`): weighted request classes with their own endpoint, stream share, context size (10k to 1M tokens), time to first token and total time. Writes one JSON line per request, with `lag_ms`, how late it left against its planned arrival (when that grows, the generator is the limit). Shard across machines with `-shard i -shards n`. | | `genconfig` | Gateway `config.yaml` with many model groups and deployments pointing at the mocks, plus `keys.txt` and `keys.sql` (rows for `LiteLLM_VerificationToken`). `-key-seed` makes every machine generate the same keys. The `bifrost` key set is written as a Bifrost config; see [Bifrost](#bifrost). | | `summarize` | Merges result files and prints proxy overhead (observed minus planned time) by stage, class and context size. | | `procmon` | Samples the gateway process's CPU, RSS, fds and TCP connections from `/proc`. | | `verify` | Checks what each gateway recorded for a run against what loadgen sent: spend logs, key/team/user/member spend, daily tables, Prometheus counters, S3 request logs and the headers the provider received. See [Full-feature runs](#full-feature-runs). | | `chatparity` | Sends the same chat completions requests through the Python proxy and the gateway, both pointed at its recording mock provider, and reports every difference in what the provider was sent and what the client got. See [Chat parity](#chat-parity). | -| `s3stub` | In-memory path-style S3 (PUT, GET, ListObjectsV2) for the `s3_v2` logs of a full-feature run. | +| `s3stub` | In-memory path-style S3 (PUT, GET, ListObjectsV2) for the `s3_v2` logs of a full-feature run. Objects are stored compressed and listed from a sorted index, so millions of logs fit. | Workloads live in `workloads/`: `smoke`, `realistic-ramp`, `long-streams`, `large-context`, -`key-churn` and `soak`. +`key-churn` and `soak`. `sustained-100`, `sustained-500` and `sustained-1000` hold one rate for +15 minutes with evenly spaced arrivals and smoke's request mix over every model group; +`sustained-check` runs the same mix at 100, 500 and 1,000 rps for 2 minutes each, to check that the +client, mock and stack hold 1,000 rps before the long runs. ## Build the images diff --git a/litellm-rust/loadtest/cmd/loadgen/main.go b/litellm-rust/loadtest/cmd/loadgen/main.go index 8b2322deb90..0b308484b52 100644 --- a/litellm-rust/loadtest/cmd/loadgen/main.go +++ b/litellm-rust/loadtest/cmd/loadgen/main.go @@ -1,8 +1,11 @@ // loadgen is an open-loop load generator for the LiteLLM gateway. Requests arrive as a Poisson -// process (so a slow proxy cannot slow the arrivals down, unlike a closed loop), each drawn from a +// process, or evenly spaced with "arrivals": "constant" (either way a slow proxy cannot slow the +// arrivals down, unlike a closed loop), each drawn from a // weighted class with its own endpoint, stream share, context size and simulated model time. The // simulated timing travels to mockllm as a directive in the prompt, so every result line carries // both what the mock was told to take and what the client saw; the difference is proxy overhead. +// Each line also has lag_ms, how late the request left against its planned arrival: when it grows, +// the generator, not the gateway, is the limit. // // Run several copies on separate machines with -shard i -shards n; each sends rate/n. package main @@ -52,6 +55,9 @@ type result struct { MaxGapMs float64 `json:"max_gap_ms,omitempty"` Chunks int `json:"chunks,omitempty"` Err string `json:"err,omitempty"` + LagMs float64 `json:"lag_ms"` + + planned time.Time } type counters struct { @@ -177,7 +183,11 @@ schedule: next = stageEnd continue } - next = next.Add(seconds(rng.ExpFloat64() / rate)) + gap := 1 / rate + if w.Arrivals != "constant" { + gap = rng.ExpFloat64() / rate + } + next = next.Add(seconds(gap)) select { case <-ctx.Done(): break schedule @@ -214,6 +224,7 @@ schedule: Stream: stream, NewConn: rng.Float64() < w.NewConnFraction, CtxTokens: int(cl.ContextTokens.Sample(rng)), TTFTPlanMs: ttftMs, DurPlanMs: durMs, + planned: next, } outTokens := max(int(float64(durMs-ttftMs)/1000*tps), 1) res.OutTokens = outTokens @@ -304,6 +315,7 @@ func do(ctx context.Context, transport *http.Transport, target string, p paths, } t0 := time.Now() + res.LagMs = ms(t0.Sub(res.planned)) resp, err := transport.RoundTrip(req) if err != nil { res.Err = trim(err.Error()) diff --git a/litellm-rust/loadtest/cmd/loadgen/workload.go b/litellm-rust/loadtest/cmd/loadgen/workload.go index 2b8909177c0..8e96426ac57 100644 --- a/litellm-rust/loadtest/cmd/loadgen/workload.go +++ b/litellm-rust/loadtest/cmd/loadgen/workload.go @@ -8,11 +8,13 @@ import ( "os" ) -// Workload describes an open-loop test: requests arrive as a Poisson process at the stage's -// rate, each one drawn from a weighted class. +// Workload describes an open-loop test: requests arrive at the stage's rate, as a Poisson process +// or evenly spaced, each one drawn from a weighted class. type Workload struct { Name string `json:"name"` Stages []Stage `json:"stages"` + // Arrivals is "poisson" (the default) or "constant", one request every 1/rps seconds. + Arrivals string `json:"arrivals"` // WarmupS seconds at the first stage's rate whose requests are marked warmup. WarmupS float64 `json:"warmup_s"` // MaxInFlight caps open requests per load generator; arrivals above it are counted as shed. @@ -132,6 +134,9 @@ func loadWorkload(path string) (*Workload, error) { return nil, fmt.Errorf("class %s: no models for endpoint %s", c.Name, c.Endpoint) } } + if w.Arrivals != "" && w.Arrivals != "poisson" && w.Arrivals != "constant" { + return nil, fmt.Errorf("%s: arrivals must be poisson or constant, got %q", path, w.Arrivals) + } if w.MaxInFlight == 0 { w.MaxInFlight = 100000 } diff --git a/litellm-rust/loadtest/cmd/s3stub/main.go b/litellm-rust/loadtest/cmd/s3stub/main.go index 679cb94f4f6..ed2ba728028 100644 --- a/litellm-rust/loadtest/cmd/s3stub/main.go +++ b/litellm-rust/loadtest/cmd/s3stub/main.go @@ -7,11 +7,16 @@ // bytes, so request logs keep their fields and numbers but not whole prompts, and a long run's logs // fit in memory. Objects that are not JSON are stored as sent. // +// Objects are kept deflate-compressed and listed from a sorted index that is rebuilt only after +// new keys arrive, so a 15-minute run at 1,000 requests/s per gateway (millions of logs) fits in a +// few GiB and verify's listing stays linear. +// // s3stub -addr :9100 -max-string 256 package main import ( "bytes" + "compress/flate" "encoding/json" "encoding/xml" "flag" @@ -26,10 +31,17 @@ import ( type store struct { mu sync.RWMutex - objects map[string][]byte // "/" + objects map[string]object // "/" + sorted []string // every key of objects, sorted; nil after a new key arrives maxString int } +// object is a stored body, deflate-compressed, and its size as sent. +type object struct { + data []byte + size int +} + type listResult struct { XMLName xml.Name `xml:"ListBucketResult"` Name string `xml:"Name"` @@ -51,7 +63,7 @@ func main() { addr := flag.String("addr", ":9100", "listen address") maxString := flag.Int("max-string", 0, "shorten JSON strings longer than this many bytes (0 keeps objects as sent)") flag.Parse() - s := &store{objects: map[string][]byte{}, maxString: *maxString} + s := &store{objects: map[string]object{}, maxString: *maxString} http.HandleFunc("GET /health", func(w http.ResponseWriter, r *http.Request) { io.WriteString(w, "ok") }) http.HandleFunc("/", s.serve) log.Printf("s3stub listening on %s", *addr) @@ -75,8 +87,12 @@ func (s *store) serve(w http.ResponseWriter, r *http.Request) { if s.maxString > 0 { body = shorten(body, s.maxString) } + obj := object{data: compress(body), size: len(body)} s.mu.Lock() - s.objects[bucket+"/"+key] = body + if _, ok := s.objects[bucket+"/"+key]; !ok { + s.sorted = nil + } + s.objects[bucket+"/"+key] = obj s.mu.Unlock() w.Header().Set("ETag", `"stub"`) case r.Method == http.MethodGet && key == "": @@ -85,15 +101,15 @@ func (s *store) serve(w http.ResponseWriter, r *http.Request) { // HeadBucket: every bucket exists. Bifrost checks its log bucket at startup. case (r.Method == http.MethodGet || r.Method == http.MethodHead) && key != "": s.mu.RLock() - body, ok := s.objects[bucket+"/"+key] + obj, ok := s.objects[bucket+"/"+key] s.mu.RUnlock() if !ok { http.Error(w, "NoSuchKey", http.StatusNotFound) return } - w.Header().Set("content-length", strconv.Itoa(len(body))) + w.Header().Set("content-length", strconv.Itoa(obj.size)) if r.Method == http.MethodGet { - w.Write(body) + io.Copy(w, flate.NewReader(bytes.NewReader(obj.data))) } default: http.Error(w, "unsupported", http.StatusMethodNotAllowed) @@ -103,33 +119,45 @@ func (s *store) serve(w http.ResponseWriter, r *http.Request) { func (s *store) list(w http.ResponseWriter, r *http.Request, bucket string) { prefix := r.URL.Query().Get("prefix") after := r.URL.Query().Get("continuation-token") - s.mu.RLock() - var keys []string - sizes := map[string]int{} - for full, body := range s.objects { - b, key, _ := strings.Cut(full, "/") - if b == bucket && strings.HasPrefix(key, prefix) && key > after { - keys = append(keys, key) - sizes[key] = len(body) - } - } - s.mu.RUnlock() - sort.Strings(keys) res := listResult{Name: bucket, Prefix: prefix} - if len(keys) > page { - res.IsTruncated = true - res.NextContinuationToken = keys[page-1] - keys = keys[:page] + s.mu.Lock() + if s.sorted == nil { + s.sorted = make([]string, 0, len(s.objects)) + for full := range s.objects { + s.sorted = append(s.sorted, full) + } + sort.Strings(s.sorted) } - for _, k := range keys { - res.Contents = append(res.Contents, entry{Key: k, Size: sizes[k]}) + from := bucket + "/" + prefix + i := sort.SearchStrings(s.sorted, from) + if after != "" { + // Past every key up to and including the token. + i = max(i, sort.Search(len(s.sorted), func(j int) bool { return s.sorted[j] > bucket+"/"+after })) } + for ; i < len(s.sorted) && strings.HasPrefix(s.sorted[i], from); i++ { + if len(res.Contents) == page { + res.IsTruncated = true + res.NextContinuationToken = res.Contents[page-1].Key + break + } + key := strings.TrimPrefix(s.sorted[i], bucket+"/") + res.Contents = append(res.Contents, entry{Key: key, Size: s.objects[s.sorted[i]].size}) + } + s.mu.Unlock() res.KeyCount = len(res.Contents) w.Header().Set("content-type", "application/xml") io.WriteString(w, xml.Header) xml.NewEncoder(w).Encode(res) } +func compress(body []byte) []byte { + var b bytes.Buffer + zw, _ := flate.NewWriter(&b, flate.BestSpeed) + zw.Write(body) + zw.Close() + return b.Bytes() +} + // shorten re-encodes a JSON object with every string cut to max bytes; numbers keep their text. func shorten(body []byte, max int) []byte { dec := json.NewDecoder(bytes.NewReader(body)) diff --git a/litellm-rust/loadtest/cmd/s3stub/main_test.go b/litellm-rust/loadtest/cmd/s3stub/main_test.go index 0f1644b3009..3aba9e09fac 100644 --- a/litellm-rust/loadtest/cmd/s3stub/main_test.go +++ b/litellm-rust/loadtest/cmd/s3stub/main_test.go @@ -1,6 +1,15 @@ package main -import "testing" +import ( + "encoding/xml" + "fmt" + "io" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "testing" +) func TestShortenKeepsFieldsAndNumbers(t *testing.T) { in := []byte(`{"messages":[{"content":"abcdefgh"}],"response_cost":0.10262000000000001,"n":3,"ok":true}`) @@ -13,3 +22,65 @@ func TestShortenKeepsFieldsAndNumbers(t *testing.T) { t.Fatalf("non-JSON changed: %s", got) } } + +func TestPutGetAndListPages(t *testing.T) { + s := &store{objects: map[string]object{}} + srv := httptest.NewServer(http.HandlerFunc(s.serve)) + defer srv.Close() + put := func(key, body string) { + req, _ := http.NewRequest(http.MethodPut, srv.URL+"/logs/"+key, strings.NewReader(body)) + resp, err := http.DefaultClient.Do(req) + if err != nil || resp.StatusCode != http.StatusOK { + t.Fatalf("put %s: %v %v", key, err, resp) + } + resp.Body.Close() + } + const n = 2*page + 7 + for i := 0; i < n; i++ { + put(fmt.Sprintf("rust/%05d.json", i), fmt.Sprintf(`{"i":%d}`, i)) + } + put("python/00000.json", `{"i":-1}`) + + resp, err := http.Get(srv.URL + "/logs/rust/00042.json") + if err != nil { + t.Fatal(err) + } + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if string(body) != `{"i":42}` || resp.Header.Get("content-length") != "8" { + t.Fatalf("get: %q %s", body, resp.Header.Get("content-length")) + } + + var keys []string + token := "" + for { + q := url.Values{"list-type": {"2"}, "prefix": {"rust/"}} + if token != "" { + q.Set("continuation-token", token) + } + resp, err := http.Get(srv.URL + "/logs?" + q.Encode()) + if err != nil { + t.Fatal(err) + } + var l listResult + if err := xml.NewDecoder(resp.Body).Decode(&l); err != nil { + t.Fatal(err) + } + resp.Body.Close() + for _, c := range l.Contents { + keys = append(keys, c.Key) + } + if !l.IsTruncated { + break + } + token = l.NextContinuationToken + } + if len(keys) != n || keys[0] != "rust/00000.json" || keys[n-1] != fmt.Sprintf("rust/%05d.json", n-1) { + t.Fatalf("listed %d keys, first %s last %s", len(keys), keys[0], keys[len(keys)-1]) + } + for i := 1; i < len(keys); i++ { + if keys[i] <= keys[i-1] { + t.Fatalf("keys out of order at %d: %s after %s", i, keys[i], keys[i-1]) + } + } +} diff --git a/litellm-rust/loadtest/workloads/sustained-100.json b/litellm-rust/loadtest/workloads/sustained-100.json new file mode 100644 index 00000000000..055b6ba4f96 --- /dev/null +++ b/litellm-rust/loadtest/workloads/sustained-100.json @@ -0,0 +1,20 @@ +{ + "name": "sustained 100 rps for 15 minutes, evenly spaced arrivals, smoke's request mix over every model group", + "stages": [{"duration_s": 900, "rps": 100}], + "arrivals": "constant", + "nonstream_max_s": 360, + "warmup_s": 60, + "max_in_flight": 30000, + "keys": {"file": "out/keys.txt", "zipf_s": 0}, + "models": {"chat": {"prefix": "gpt-6-astra-", "count": 10}, "messages": {"prefix": "claude-opus-5-5-", "count": 10}}, + "classes": [ + {"name": "chat", "weight": 1, "endpoint": "chat", "stream_fraction": 0.5, + "context_tokens": {"dist": "loguniform", "min": 10000, "max": 50000}, + "ttft_ms": {"dist": "uniform", "min": 200, "max": 800}, + "duration_s": {"dist": "uniform", "min": 1, "max": 5}}, + {"name": "messages", "weight": 1, "endpoint": "messages", "stream_fraction": 0.5, + "context_tokens": {"dist": "loguniform", "min": 10000, "max": 50000}, + "ttft_ms": {"dist": "uniform", "min": 200, "max": 800}, + "duration_s": {"dist": "uniform", "min": 1, "max": 5}} + ] +} diff --git a/litellm-rust/loadtest/workloads/sustained-1000.json b/litellm-rust/loadtest/workloads/sustained-1000.json new file mode 100644 index 00000000000..d277b240c7c --- /dev/null +++ b/litellm-rust/loadtest/workloads/sustained-1000.json @@ -0,0 +1,20 @@ +{ + "name": "sustained 1000 rps for 15 minutes, evenly spaced arrivals, smoke's request mix over every model group", + "stages": [{"duration_s": 900, "rps": 1000}], + "arrivals": "constant", + "nonstream_max_s": 360, + "warmup_s": 60, + "max_in_flight": 30000, + "keys": {"file": "out/keys.txt", "zipf_s": 0}, + "models": {"chat": {"prefix": "gpt-6-astra-", "count": 10}, "messages": {"prefix": "claude-opus-5-5-", "count": 10}}, + "classes": [ + {"name": "chat", "weight": 1, "endpoint": "chat", "stream_fraction": 0.5, + "context_tokens": {"dist": "loguniform", "min": 10000, "max": 50000}, + "ttft_ms": {"dist": "uniform", "min": 200, "max": 800}, + "duration_s": {"dist": "uniform", "min": 1, "max": 5}}, + {"name": "messages", "weight": 1, "endpoint": "messages", "stream_fraction": 0.5, + "context_tokens": {"dist": "loguniform", "min": 10000, "max": 50000}, + "ttft_ms": {"dist": "uniform", "min": 200, "max": 800}, + "duration_s": {"dist": "uniform", "min": 1, "max": 5}} + ] +} diff --git a/litellm-rust/loadtest/workloads/sustained-500.json b/litellm-rust/loadtest/workloads/sustained-500.json new file mode 100644 index 00000000000..f296955d7cf --- /dev/null +++ b/litellm-rust/loadtest/workloads/sustained-500.json @@ -0,0 +1,20 @@ +{ + "name": "sustained 500 rps for 15 minutes, evenly spaced arrivals, smoke's request mix over every model group", + "stages": [{"duration_s": 900, "rps": 500}], + "arrivals": "constant", + "nonstream_max_s": 360, + "warmup_s": 60, + "max_in_flight": 30000, + "keys": {"file": "out/keys.txt", "zipf_s": 0}, + "models": {"chat": {"prefix": "gpt-6-astra-", "count": 10}, "messages": {"prefix": "claude-opus-5-5-", "count": 10}}, + "classes": [ + {"name": "chat", "weight": 1, "endpoint": "chat", "stream_fraction": 0.5, + "context_tokens": {"dist": "loguniform", "min": 10000, "max": 50000}, + "ttft_ms": {"dist": "uniform", "min": 200, "max": 800}, + "duration_s": {"dist": "uniform", "min": 1, "max": 5}}, + {"name": "messages", "weight": 1, "endpoint": "messages", "stream_fraction": 0.5, + "context_tokens": {"dist": "loguniform", "min": 10000, "max": 50000}, + "ttft_ms": {"dist": "uniform", "min": 200, "max": 800}, + "duration_s": {"dist": "uniform", "min": 1, "max": 5}} + ] +} diff --git a/litellm-rust/loadtest/workloads/sustained-check.json b/litellm-rust/loadtest/workloads/sustained-check.json new file mode 100644 index 00000000000..3b9d3c78f20 --- /dev/null +++ b/litellm-rust/loadtest/workloads/sustained-check.json @@ -0,0 +1,20 @@ +{ + "name": "sustained-rate check: 100, 500 and 1,000 rps for 2 minutes each with the sustained workloads' mix, to confirm the client, mock and stack hold 1,000 rps before the 15-minute runs", + "stages": [{"duration_s": 120, "rps": 100}, {"duration_s": 120, "rps": 500}, {"duration_s": 120, "rps": 1000}], + "arrivals": "constant", + "nonstream_max_s": 360, + "warmup_s": 15, + "max_in_flight": 30000, + "keys": {"file": "out/keys.txt", "zipf_s": 0}, + "models": {"chat": {"prefix": "gpt-6-astra-", "count": 10}, "messages": {"prefix": "claude-opus-5-5-", "count": 10}}, + "classes": [ + {"name": "chat", "weight": 1, "endpoint": "chat", "stream_fraction": 0.5, + "context_tokens": {"dist": "loguniform", "min": 10000, "max": 50000}, + "ttft_ms": {"dist": "uniform", "min": 200, "max": 800}, + "duration_s": {"dist": "uniform", "min": 1, "max": 5}}, + {"name": "messages", "weight": 1, "endpoint": "messages", "stream_fraction": 0.5, + "context_tokens": {"dist": "loguniform", "min": 10000, "max": 50000}, + "ttft_ms": {"dist": "uniform", "min": 200, "max": 800}, + "duration_s": {"dist": "uniform", "min": 1, "max": 5}} + ] +}