Ilker ManapandClaude Opus 5
05bdc7d6d4
`ctl bench` measures what a replica can actually serve, so weight, target_waiting and kv_ceiling stop being numbers somebody typed. Concurrency climbs until the SLO breaks and the last level that met it is the capacity: peak throughput arrives long after latency has become unusable, so throughput alone overstates what the hardware can serve. Queue depth and KV usage are read from the engine while it is still meeting the target, which is exactly what admission should treat as healthy. Prompts are unique per request by default. With one repeated prompt the engine's cache answers most of the prefill and the measurement describes the cache rather than the hardware; --shared-prefix asks that question deliberately, and it now produces byte-identical prompts, which the first version did not — it dropped the random header and still picked words at random, so the cache missed and the flag quietly measured the thing it exists to avoid. Running it against the cluster turned up two real defects: A replica stuck in Starting never timed out. Every other way out of that state needs the agent to report something, and when it never does the replica sits there holding GPUs and counting towards the model's replica count while serving nothing. Two of them had been like that for eight minutes past a two-minute startup timeout, and nothing would ever have moved them. They were stuck because adoption lost to a replica we had only placed. Replica ids come from the model name and a counter, so a control plane that restarts and places before its agents reconnect builds a different replica under the same id on a freshly allocated port; the agent's report of the engine that was actually serving was then discarded as already-known. The cluster pointed at port 20004 while the engine answered on 20000. The fleet is authoritative about what is running, so what we merely placed now loses to what an agent reports — unless our copy has a shard that has actually come up, in which case two things really are running and the stray is the one we did not ask for. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
109 lines
2.3 KiB
Go
109 lines
2.3 KiB
Go
package bench
|
|
|
|
import (
|
|
"context"
|
|
"net/http"
|
|
"sync"
|
|
"time"
|
|
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/router/scrape"
|
|
)
|
|
|
|
// sampler reads an engine's metrics during a step.
|
|
//
|
|
// It is what turns a latency measurement into something admission can use: the
|
|
// queue depth and cache pressure observed while the SLO was still being met are
|
|
// the values target_waiting and kv_ceiling should have, rather than the round
|
|
// numbers somebody typed.
|
|
//
|
|
// Optional, because a run pointed at the gateway has no single engine to read.
|
|
type sampler struct {
|
|
url string
|
|
engine string
|
|
client *http.Client
|
|
cancel context.CancelFunc
|
|
|
|
mu sync.Mutex
|
|
wg sync.WaitGroup
|
|
waiting float64
|
|
running float64
|
|
kv float64
|
|
n int
|
|
}
|
|
|
|
func newSampler(metricsURL, engine string) *sampler {
|
|
return &sampler{
|
|
url: metricsURL,
|
|
engine: engine,
|
|
client: &http.Client{Timeout: 2 * time.Second},
|
|
}
|
|
}
|
|
|
|
func (s *sampler) start(ctx context.Context) {
|
|
if s.url == "" {
|
|
return
|
|
}
|
|
ctx, s.cancel = context.WithCancel(ctx)
|
|
s.wg.Add(1)
|
|
go func() {
|
|
defer s.wg.Done()
|
|
t := time.NewTicker(time.Second)
|
|
defer t.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-t.C:
|
|
s.sample(ctx)
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
|
|
func (s *sampler) sample(ctx context.Context) {
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodGet, s.url, nil)
|
|
if err != nil {
|
|
return
|
|
}
|
|
resp, err := s.client.Do(req)
|
|
if err != nil {
|
|
return
|
|
}
|
|
defer resp.Body.Close()
|
|
if resp.StatusCode != http.StatusOK {
|
|
return
|
|
}
|
|
|
|
// The same mapping the router reads with, so a bench result cannot disagree
|
|
// with what the scheduler sees. The engine has to be named: reading a
|
|
// llama.cpp endpoint with vLLM's names finds nothing, and the run would
|
|
// report a queue depth of zero and suggest thresholds built on it.
|
|
stats, err := scrape.Parse(resp.Body, scrape.MappingFor(s.engine))
|
|
if err != nil {
|
|
return
|
|
}
|
|
s.mu.Lock()
|
|
s.waiting += float64(stats.Waiting)
|
|
s.running += float64(stats.Running)
|
|
s.kv += stats.KVCacheUsage
|
|
s.n++
|
|
s.mu.Unlock()
|
|
}
|
|
|
|
func (s *sampler) stop() {
|
|
if s.cancel != nil {
|
|
s.cancel()
|
|
}
|
|
s.wg.Wait()
|
|
}
|
|
|
|
func (s *sampler) averages() (waiting, running, kv float64) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
if s.n == 0 {
|
|
return 0, 0, 0
|
|
}
|
|
n := float64(s.n)
|
|
return s.waiting / n, s.running / n, s.kv / n
|
|
}
|