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

@@ -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/<id>/` 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
 

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())

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
 	}

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 // "<bucket>/<key>"
+	objects   map[string]object // "<bucket>/<key>"
+	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))

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])
+		}
+	}
+}

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}}
+  ]
+}

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}}
+  ]
+}

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}}
+  ]
+}

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}}
+  ]
+}