Merge engine fixes: SGLang metrics and flags, bench reading guarded metrics
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
commit
cc4de36674
12 files changed
+655
-28
No files matched your search
@@ -597,6 +597,11 @@ engines:
|
||||
sglang: "0.5.20" # SGLang only when named
|
||||
```
|
||||
|
||||
SGLang is started with `--enable-metrics`, without which it serves no
|
||||
`/metrics`; the router reads `sglang:num_running_reqs`, `sglang:num_queue_reqs`
|
||||
and `sglang:token_usage` there. Check an SGLang against the flags llmgw builds
|
||||
with `llmgw ctl engine-check sglang --bin <its environment>/bin/python`.
|
||||
|
||||
A changed version reaches every node when it next connects. To use an engine
|
||||
you installed yourself on a node, name its path in that node's `agent.yaml`
|
||||
(`engines.llamacpp.bin`, `engines.vllm.bin`, `engines.sglang.python`,
|
||||
|
||||
@@ -366,6 +366,26 @@ func TestEnginesListenOnTheAddressTheyAreGiven(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestSGLangIsStartedServingItsMetrics(t *testing.T) {
|
||||
// Without --enable-metrics SGLang has no /metrics, and every scrape of
|
||||
// the replica fails as though it were unreachable.
|
||||
if cmd := buildOne(t, &SGLang{}, singleNodeSpec()); !slices.Contains(cmd.Args, "--enable-metrics") {
|
||||
t.Errorf("not started with its metrics: %v", cmd.Args)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSGLangIsAskedForItsHelpThroughItsModule(t *testing.T) {
|
||||
g := &SGLang{cfg: SGLangConfig{Python: "/opt/sglang/bin/python"}}
|
||||
g.UseInstalled("/opt/llmgw/engines/sglang/0.5.21-cu13/bin/python")
|
||||
path, args := g.HelpCommand(Command{})
|
||||
if path != "/opt/llmgw/engines/sglang/0.5.21-cu13/bin/python" {
|
||||
t.Errorf("asked %s, not the interpreter it was installed into", path)
|
||||
}
|
||||
if !slices.Equal(args, []string{"-m", "sglang.launch_server", "--help"}) {
|
||||
t.Errorf("asked %v", args)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSGLangSingleNodeOmitsMultiNodeFlags(t *testing.T) {
|
||||
cmd := buildOne(t, &SGLang{}, singleNodeSpec())
|
||||
for _, flag := range []string{"--nnodes", "--node-rank", "--dist-init-addr"} {
|
||||
|
||||
@@ -40,7 +40,15 @@ func (*SGLang) Name() string { return "sglang" }
|
||||
// HealthPath is SGLang's readiness endpoint.
|
||||
func (*SGLang) HealthPath() string { return "/health" }
|
||||
|
||||
// MetricsPath is SGLang's Prometheus endpoint.
|
||||
// HelpCommand is how to ask SGLang for its help: through its module, in the
|
||||
// environment it was installed into, since the command itself is llmgw's
|
||||
// launcher and the interpreter's own --help describes Python.
|
||||
func (g *SGLang) HelpCommand(Command) (string, []string) {
|
||||
return g.python(), []string{"-m", "sglang.launch_server", "--help"}
|
||||
}
|
||||
|
||||
// MetricsPath is SGLang's Prometheus endpoint, served only with
|
||||
// --enable-metrics.
|
||||
func (*SGLang) MetricsPath() string { return "/metrics" }
|
||||
|
||||
// Validate checks the profile against the GPUs it would run on.
|
||||
@@ -83,7 +91,13 @@ func (g *SGLang) serverCommand(s ShardSpec) (Command, error) {
|
||||
"--model-path", p.ModelPath,
|
||||
"--host", s.Host(),
|
||||
"--port", strconv.Itoa(s.HTTPPort),
|
||||
"--tp", strconv.Itoa(p.TP),
|
||||
// The flag's whole name: SGLang's documentation writes --tp, which
|
||||
// argparse accepts only as an abbreviation of --tp-size and only
|
||||
// while no other flag begins with it.
|
||||
"--tp-size", strconv.Itoa(p.TP),
|
||||
// Without it SGLang serves no /metrics at all, and the router has
|
||||
// nothing to schedule this replica by.
|
||||
"--enable-metrics",
|
||||
}
|
||||
if p.ServedModelName != "" {
|
||||
args = append(args, "--served-model-name", p.ServedModelName)
|
||||
|
||||
+24
-4
@@ -59,8 +59,13 @@ type Config struct {
|
||||
// is silent: nothing matches, every sample reads zero, and the suggested
|
||||
// thresholds are built on nothing.
|
||||
Engine string
|
||||
Model string
|
||||
APIKey string
|
||||
// MetricsKey is the engine key MetricsURL is read with. Not APIKey: that
|
||||
// is whatever Target takes, which for the gateway is a client's key, and
|
||||
// an engine that guards its metrics (llama.cpp, TensorRT-LLM) takes only
|
||||
// its own.
|
||||
MetricsKey string
|
||||
Model string
|
||||
APIKey string
|
||||
|
||||
// Concurrency is the ramp. Each level runs for Duration.
|
||||
Concurrency []int
|
||||
@@ -140,6 +145,10 @@ type Step struct {
|
||||
// FirstError is kept so a step that failed says why rather than just
|
||||
// reporting zeroes.
|
||||
FirstError string
|
||||
// MetricsUnread says why a step that was to read the engine's metrics
|
||||
// has none, so its zero queue depth and KV usage are not taken for an
|
||||
// idle engine.
|
||||
MetricsUnread string `json:",omitempty"`
|
||||
}
|
||||
|
||||
// Result is the whole ramp plus what it concluded.
|
||||
@@ -169,6 +178,17 @@ func Run(ctx context.Context, cfg Config, progress func(Step)) (Result, error) {
|
||||
return Result{}, err
|
||||
}
|
||||
res := Result{Model: cfg.Model, Target: cfg.Target, SLO: cfg.SLO, RanAt: time.Now()}
|
||||
|
||||
// Read once before any load. Metrics that cannot be read — refused for
|
||||
// want of the engine's key, or written in another engine's names — fail
|
||||
// every sample alike, and finding that out after the ramp has cost its
|
||||
// GPU time would leave a run with no thresholds to suggest.
|
||||
if cfg.MetricsURL != "" {
|
||||
if err := newSampler(cfg.MetricsURL, cfg.Engine, cfg.MetricsKey).read(ctx); err != nil {
|
||||
return Result{}, fmt.Errorf("bench: the engine's metrics cannot be read: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
start := time.Now()
|
||||
|
||||
client := &http.Client{Timeout: 10 * time.Minute}
|
||||
@@ -224,8 +244,7 @@ func runStep(ctx context.Context, client *http.Client, cfg Config, concurrency i
|
||||
firstErr string
|
||||
)
|
||||
|
||||
sampler := newSampler(cfg.MetricsURL, cfg.Engine)
|
||||
sampler.apiKey = cfg.APIKey
|
||||
sampler := newSampler(cfg.MetricsURL, cfg.Engine, cfg.MetricsKey)
|
||||
sampler.start(ctx)
|
||||
|
||||
start := time.Now()
|
||||
@@ -271,6 +290,7 @@ func runStep(ctx context.Context, client *http.Client, cfg Config, concurrency i
|
||||
step.TokensPerSec = float64(tokens) / elapsed.Seconds()
|
||||
}
|
||||
step.Waiting, step.Running, step.KVUsage = sampler.averages()
|
||||
step.MetricsUnread = sampler.unread()
|
||||
|
||||
// A step that could not complete anything has not met anything.
|
||||
step.MeetsSLO = step.Completed > 0 && step.Failed == 0 &&
|
||||
|
||||
@@ -2,6 +2,8 @@ package bench
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -30,12 +32,16 @@ type sampler struct {
|
||||
running float64
|
||||
kv float64
|
||||
n int
|
||||
// lastErr says why a step has no numbers: averages over no samples are
|
||||
// zeros, and zeros read as an idle engine.
|
||||
lastErr error
|
||||
}
|
||||
|
||||
func newSampler(metricsURL, engine string) *sampler {
|
||||
func newSampler(metricsURL, engine, key string) *sampler {
|
||||
return &sampler{
|
||||
url: metricsURL,
|
||||
engine: engine,
|
||||
apiKey: key,
|
||||
client: &http.Client{Timeout: 2 * time.Second},
|
||||
}
|
||||
}
|
||||
@@ -61,21 +67,41 @@ func (s *sampler) start(ctx context.Context) {
|
||||
}()
|
||||
}
|
||||
|
||||
func (s *sampler) sample(ctx context.Context) {
|
||||
// MetricsRefusedError is an engine refusing to show its metrics without its
|
||||
// key: llama.cpp guards /metrics with it, and llmgw's TensorRT-LLM launcher
|
||||
// guards everything but /health.
|
||||
type MetricsRefusedError struct {
|
||||
URL string
|
||||
Status string
|
||||
KeySent bool
|
||||
}
|
||||
|
||||
func (e *MetricsRefusedError) Error() string {
|
||||
if e.KeySent {
|
||||
return fmt.Sprintf("%s refused the key sent for its metrics (%s)", e.URL, e.Status)
|
||||
}
|
||||
return fmt.Sprintf("%s guards its metrics with the engine's key, and none was given (%s)", e.URL, e.Status)
|
||||
}
|
||||
|
||||
// read takes one sample.
|
||||
func (s *sampler) read(ctx context.Context) error {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, s.url, nil)
|
||||
if err != nil {
|
||||
return
|
||||
return err
|
||||
}
|
||||
if s.apiKey != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+s.apiKey)
|
||||
}
|
||||
resp, err := s.client.Do(req)
|
||||
if err != nil {
|
||||
return
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return
|
||||
switch {
|
||||
case resp.StatusCode == http.StatusUnauthorized || resp.StatusCode == http.StatusForbidden:
|
||||
return &MetricsRefusedError{URL: s.url, Status: resp.Status, KeySent: s.apiKey != ""}
|
||||
case resp.StatusCode != http.StatusOK:
|
||||
return fmt.Errorf("%s: %s", s.url, resp.Status)
|
||||
}
|
||||
|
||||
// The same mapping the router reads with, so a bench result cannot disagree
|
||||
@@ -84,7 +110,11 @@ func (s *sampler) sample(ctx context.Context) {
|
||||
// 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
|
||||
var missing *scrape.MissingError
|
||||
if errors.As(err, &missing) {
|
||||
return fmt.Errorf("%s does not serve %v: is %q the engine there?", s.url, missing.Missing, s.engine)
|
||||
}
|
||||
return err
|
||||
}
|
||||
s.mu.Lock()
|
||||
s.waiting += float64(stats.Waiting)
|
||||
@@ -92,6 +122,17 @@ func (s *sampler) sample(ctx context.Context) {
|
||||
s.kv += stats.KVCacheUsage
|
||||
s.n++
|
||||
s.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *sampler) sample(ctx context.Context) {
|
||||
err := s.read(ctx)
|
||||
if err == nil || ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
s.mu.Lock()
|
||||
s.lastErr = err
|
||||
s.mu.Unlock()
|
||||
}
|
||||
|
||||
func (s *sampler) stop() {
|
||||
@@ -110,3 +151,14 @@ func (s *sampler) averages() (waiting, running, kv float64) {
|
||||
n := float64(s.n)
|
||||
return s.waiting / n, s.running / n, s.kv / n
|
||||
}
|
||||
|
||||
// unread says why a step that tried to read the metrics has none of them, or
|
||||
// "" when it has some or never tried.
|
||||
func (s *sampler) unread() string {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if s.n > 0 || s.lastErr == nil {
|
||||
return ""
|
||||
}
|
||||
return s.lastErr.Error()
|
||||
}
|
||||
@@ -0,0 +1,133 @@
|
||||
package bench
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// llamaMetrics is the part of llama-server's /metrics the router reads.
|
||||
const llamaMetrics = `# TYPE llamacpp:requests_processing gauge
|
||||
llamacpp:requests_processing 1
|
||||
# TYPE llamacpp:requests_deferred gauge
|
||||
llamacpp:requests_deferred 2
|
||||
`
|
||||
|
||||
// guardedEngine is a llama.cpp-like engine: it streams completions to
|
||||
// clientKey and serves /metrics only to engineKey, as llama-server does with
|
||||
// --api-key. refuseAfter, when set, makes /metrics refuse every read after
|
||||
// that many.
|
||||
type guardedEngine struct {
|
||||
clientKey, engineKey string
|
||||
refuseAfter int64
|
||||
chats, reads atomic.Int64
|
||||
}
|
||||
|
||||
func (g *guardedEngine) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
switch r.URL.Path {
|
||||
case "/metrics":
|
||||
n := g.reads.Add(1)
|
||||
if r.Header.Get("Authorization") != "Bearer "+g.engineKey || (g.refuseAfter > 0 && n > g.refuseAfter) {
|
||||
http.Error(w, `{"error":"Invalid API Key"}`, http.StatusUnauthorized)
|
||||
return
|
||||
}
|
||||
_, _ = w.Write([]byte(llamaMetrics))
|
||||
case "/v1/chat/completions":
|
||||
if r.Header.Get("Authorization") != "Bearer "+g.clientKey {
|
||||
http.Error(w, "no", http.StatusUnauthorized)
|
||||
return
|
||||
}
|
||||
g.chats.Add(1)
|
||||
w.Header().Set("Content-Type", "text/event-stream")
|
||||
for i := range 3 {
|
||||
fmt.Fprintf(w, "data: {\"choices\":[{\"delta\":{\"content\":\"t%d\"}}]}\n\n", i)
|
||||
}
|
||||
fmt.Fprint(w, "data: [DONE]\n\n")
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
default:
|
||||
http.NotFound(w, r)
|
||||
}
|
||||
}
|
||||
|
||||
func guardedRun(t *testing.T, g *guardedEngine, metricsKey string) (Result, error) {
|
||||
t.Helper()
|
||||
srv := httptest.NewServer(g)
|
||||
t.Cleanup(srv.Close)
|
||||
return Run(context.Background(), Config{
|
||||
Target: srv.URL, APIKey: g.clientKey, Model: "m",
|
||||
MetricsURL: srv.URL + "/metrics", Engine: "llamacpp", MetricsKey: metricsKey,
|
||||
Concurrency: []int{1}, Duration: 1500 * time.Millisecond,
|
||||
SLO: SLO{TTFTp99: time.Minute},
|
||||
}, nil)
|
||||
}
|
||||
|
||||
func TestAnEngineThatGuardsItsMetricsIsReadWithItsOwnKey(t *testing.T) {
|
||||
// The requests go with the client's key and the metrics with the
|
||||
// engine's: llama.cpp's /metrics takes only its own.
|
||||
g := &guardedEngine{clientKey: "client", engineKey: "engine"}
|
||||
res, err := guardedRun(t, g, "engine")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
s := res.Steps[0]
|
||||
if s.MetricsUnread != "" || s.Waiting != 2 || s.Running != 1 {
|
||||
t.Errorf("metrics not read: waiting=%v running=%v unread=%q", s.Waiting, s.Running, s.MetricsUnread)
|
||||
}
|
||||
if res.SuggestedTargetWaiting != 2 {
|
||||
t.Errorf("target_waiting suggested %d", res.SuggestedTargetWaiting)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMetricsThatRefuseTheRunStopItBeforeAnyLoad(t *testing.T) {
|
||||
// Without the engine's key every sample is refused, and the run used to
|
||||
// report a queue depth of zero as though the engine had been idle.
|
||||
g := &guardedEngine{clientKey: "client", engineKey: "engine"}
|
||||
_, err := guardedRun(t, g, "")
|
||||
var refused *MetricsRefusedError
|
||||
if !errors.As(err, &refused) || refused.KeySent {
|
||||
t.Fatalf("expected the metrics refused for want of a key, got %v", err)
|
||||
}
|
||||
if n := g.chats.Load(); n != 0 {
|
||||
t.Errorf("%d requests were sent before the metrics were found unreadable", n)
|
||||
}
|
||||
|
||||
_, err = guardedRun(t, &guardedEngine{clientKey: "client", engineKey: "engine"}, "client")
|
||||
if !errors.As(err, &refused) || !refused.KeySent {
|
||||
t.Fatalf("expected the key sent reported refused, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAStepWhoseMetricsWereNotReadSaysSo(t *testing.T) {
|
||||
// Readable before the ramp and not during it — an engine restarted with
|
||||
// another key, say. The step's zeros are not an idle engine.
|
||||
g := &guardedEngine{clientKey: "client", engineKey: "engine", refuseAfter: 1}
|
||||
res, err := guardedRun(t, g, "engine")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if s := res.Steps[0]; s.MetricsUnread == "" {
|
||||
t.Errorf("a step with no metrics read reported none missing: %+v", s)
|
||||
}
|
||||
if res.SuggestedTargetWaiting != 0 || res.SuggestedKVCeiling != 0 {
|
||||
t.Errorf("thresholds suggested from metrics never read: %+v", res)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMetricsInAnotherEnginesNamesStopTheRun(t *testing.T) {
|
||||
g := &guardedEngine{clientKey: "client", engineKey: "engine"}
|
||||
srv := httptest.NewServer(g)
|
||||
defer srv.Close()
|
||||
_, err := Run(context.Background(), Config{
|
||||
Target: srv.URL, APIKey: "client", Model: "m",
|
||||
MetricsURL: srv.URL + "/metrics", Engine: "vllm", MetricsKey: "engine",
|
||||
Concurrency: []int{1}, Duration: time.Second,
|
||||
}, nil)
|
||||
if err == nil || g.chats.Load() != 0 {
|
||||
t.Fatalf("a run reading llama.cpp's metrics with vLLM's names started: %v", err)
|
||||
}
|
||||
}
|
||||
+64
-14
@@ -2,7 +2,9 @@ package cli
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/url"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
@@ -20,6 +22,7 @@ func newBenchCmd(client func() *ctlClient, asJSON *bool) *cobra.Command {
|
||||
metrics string
|
||||
engine string
|
||||
key string
|
||||
mkey string
|
||||
save bool
|
||||
)
|
||||
|
||||
@@ -41,10 +44,25 @@ func newBenchCmd(client func() *ctlClient, asJSON *bool) *cobra.Command {
|
||||
cfg.MetricsURL = metrics
|
||||
cfg.Engine = engine
|
||||
cfg.APIKey = key
|
||||
// A replica's engine demands its own key. Asked for rather than
|
||||
// typed: the admin API knows which replica answers at the target.
|
||||
if cfg.APIKey == "" {
|
||||
cfg.APIKey = replicaKey(cmd, client, target)
|
||||
cfg.MetricsKey = mkey
|
||||
// A replica's engine demands its own key, for its API and — on
|
||||
// llama.cpp and TensorRT-LLM — for its metrics. Asked for rather
|
||||
// than typed: the admin API knows which replica answers where, and
|
||||
// this runs wherever ctl does, which can reach it.
|
||||
needKeys := cfg.APIKey == "" || (metrics != "" && cfg.MetricsKey == "")
|
||||
if needKeys {
|
||||
reps, err := listReplicas(cmd, client)
|
||||
if cfg.APIKey == "" {
|
||||
cfg.APIKey = replicaKey(reps, target)
|
||||
}
|
||||
if metrics != "" && cfg.MetricsKey == "" {
|
||||
cfg.MetricsKey = replicaKey(reps, metrics)
|
||||
if err != nil {
|
||||
cmd.PrintErrf("The admin API could not be asked for the engine key of the replica at "+
|
||||
"--metrics (%v); an engine that guards its metrics will refuse them. "+
|
||||
"Pass --metrics-key to give it.\n", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
if len(steps) > 0 {
|
||||
cfg.Concurrency = steps
|
||||
@@ -72,17 +90,29 @@ func newBenchCmd(client func() *ctlClient, asJSON *bool) *cobra.Command {
|
||||
if s.Failed > 0 {
|
||||
verdict = fmt.Sprintf("%d failed", s.Failed)
|
||||
}
|
||||
// Unread is not zero: a dash, and the reason under the row.
|
||||
wait, kv := fmt.Sprintf("%.1f", s.Waiting), fmt.Sprintf("%.0f%%", s.KVUsage*100)
|
||||
if s.MetricsUnread != "" {
|
||||
wait, kv = "-", "-"
|
||||
}
|
||||
fmt.Fprintf(cmd.ErrOrStderr(), row,
|
||||
fmt.Sprint(s.Concurrency),
|
||||
fmt.Sprintf("%.1f", s.RequestsPerSec),
|
||||
fmt.Sprintf("%.0f", s.TokensPerSec),
|
||||
ms(s.TTFT.P50), ms(s.TTFT.P99), ms(s.TPOT.P99),
|
||||
fmt.Sprintf("%.1f", s.Waiting),
|
||||
fmt.Sprintf("%.0f%%", s.KVUsage*100),
|
||||
verdict)
|
||||
wait, kv, verdict)
|
||||
if s.MetricsUnread != "" {
|
||||
fmt.Fprintf(cmd.ErrOrStderr(), " metrics not read: %s\n", s.MetricsUnread)
|
||||
}
|
||||
}
|
||||
|
||||
res, err := bench.Run(cmd.Context(), cfg, progress)
|
||||
var refused *bench.MetricsRefusedError
|
||||
if errors.As(err, &refused) {
|
||||
return fmt.Errorf("%w. The engine's key is the admin API's, given for the replica "+
|
||||
"whose endpoint `llmgw ctl replicas` lists at --metrics' address; "+
|
||||
"or pass it with --metrics-key", err)
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -108,6 +138,8 @@ func newBenchCmd(client func() *ctlClient, asJSON *bool) *cobra.Command {
|
||||
f.StringVar(&metrics, "metrics", "", "the replica's /metrics, to observe queue depth and KV usage")
|
||||
f.StringVar(&engine, "engine", "", "which metric names --metrics uses: vllm | sglang | llamacpp | trtllm (whose are at /prometheus/metrics)")
|
||||
f.StringVar(&key, "key", "", "API key, when the gateway requires one")
|
||||
f.StringVar(&mkey, "metrics-key", "",
|
||||
"the engine key --metrics is read with (default: the replica's, from the admin API)")
|
||||
f.IntSliceVar(&steps, "concurrency", nil, "levels to try (default 1,2,4,…,128)")
|
||||
f.DurationVar(&cfg.Duration, "duration", cfg.Duration, "how long to hold each level")
|
||||
f.IntVar(&cfg.PromptTokens, "prompt-tokens", cfg.PromptTokens, "approximate prompt size")
|
||||
@@ -233,17 +265,35 @@ func ms(d time.Duration) string {
|
||||
}
|
||||
}
|
||||
|
||||
// replicaKey is the engine key of the replica at target, or "" when target is
|
||||
// not a replica (the gateway, say) or the admin API cannot be asked.
|
||||
func replicaKey(cmd *cobra.Command, client func() *ctlClient, target string) string {
|
||||
func listReplicas(cmd *cobra.Command, client func() *ctlClient) ([]adminapi.ReplicaView, error) {
|
||||
var reps []adminapi.ReplicaView
|
||||
if err := client().get(cmd.Context(), "/api/v1/replicas", &reps); err != nil {
|
||||
return ""
|
||||
}
|
||||
err := client().get(cmd.Context(), "/api/v1/replicas", &reps)
|
||||
return reps, err
|
||||
}
|
||||
|
||||
// replicaKey is the engine key of the replica serving at addr, or "" when
|
||||
// addr is not a replica's (the gateway's, say).
|
||||
//
|
||||
// Matched by scheme, host and port rather than by the whole URL, so the same
|
||||
// lookup serves the target, which is a replica's base URL, and --metrics,
|
||||
// which is a path under it: /metrics, or /prometheus/metrics on TensorRT-LLM.
|
||||
func replicaKey(reps []adminapi.ReplicaView, addr string) string {
|
||||
for _, rep := range reps {
|
||||
if sameEndpoint(rep.Endpoint, target) {
|
||||
if sameOrigin(rep.Endpoint, addr) {
|
||||
return rep.EngineKey
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func sameOrigin(a, b string) bool {
|
||||
ua, err := url.Parse(a)
|
||||
if err != nil || ua.Host == "" {
|
||||
return false
|
||||
}
|
||||
ub, err := url.Parse(b)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
return ua.Scheme == ub.Scheme && ua.Host == ub.Host
|
||||
}
|
||||
@@ -0,0 +1,96 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
|
||||
)
|
||||
|
||||
// benchCluster is a gateway taking the client's key, a llama.cpp replica
|
||||
// whose /metrics takes only its engine key, and an admin API that lists the
|
||||
// replica with that key, as the control plane does.
|
||||
type benchCluster struct {
|
||||
gateway, engine, admin *httptest.Server
|
||||
metricsAuth atomic.Value
|
||||
}
|
||||
|
||||
func newBenchCluster(t *testing.T) *benchCluster {
|
||||
t.Helper()
|
||||
c := &benchCluster{}
|
||||
c.metricsAuth.Store("")
|
||||
c.gateway = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Header.Get("Authorization") != "Bearer client" {
|
||||
http.Error(w, "no", http.StatusUnauthorized)
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "text/event-stream")
|
||||
fmt.Fprint(w, "data: {\"choices\":[{\"delta\":{\"content\":\"a\"}}]}\n\n")
|
||||
fmt.Fprint(w, "data: {\"choices\":[{\"delta\":{\"content\":\"b\"}}]}\n\n")
|
||||
fmt.Fprint(w, "data: [DONE]\n\n")
|
||||
}))
|
||||
c.engine = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
c.metricsAuth.Store(r.Header.Get("Authorization"))
|
||||
if r.URL.Path != "/metrics" || r.Header.Get("Authorization") != "Bearer engine-key" {
|
||||
http.Error(w, `{"error":"Invalid API Key"}`, http.StatusUnauthorized)
|
||||
return
|
||||
}
|
||||
fmt.Fprint(w, "llamacpp:requests_processing 1\nllamacpp:requests_deferred 3\n")
|
||||
}))
|
||||
c.admin = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
_ = json.NewEncoder(w).Encode([]adminapi.ReplicaView{
|
||||
{ID: "r1", Model: "m", Profile: "cpu", Endpoint: c.engine.URL, EngineKey: "engine-key"},
|
||||
})
|
||||
}))
|
||||
t.Cleanup(func() { c.gateway.Close(); c.engine.Close(); c.admin.Close() })
|
||||
return c
|
||||
}
|
||||
|
||||
func (c *benchCluster) bench(t *testing.T, args ...string) (string, error) {
|
||||
t.Helper()
|
||||
t.Setenv("LLMGW_ADMIN_TOKEN", "t")
|
||||
asJSON := false
|
||||
cmd := newBenchCmd(func() *ctlClient { return newCtlClient(c.admin.URL) }, &asJSON)
|
||||
var out bytes.Buffer
|
||||
cmd.SetOut(&out)
|
||||
cmd.SetErr(&out)
|
||||
cmd.SetArgs(append([]string{"m", "--concurrency", "1", "--duration", "1500ms", "--ttft-p99", "1m", "--tpot-p99", "1m"}, args...))
|
||||
err := cmd.ExecuteContext(context.Background())
|
||||
return out.String(), err
|
||||
}
|
||||
|
||||
func TestBenchReadsAGuardedEnginesMetricsWithTheKeyTheAdminAPIGives(t *testing.T) {
|
||||
// Through the gateway with a client's key, and the replica's metrics on
|
||||
// the side: the client's key is not the engine's, and the engine's is
|
||||
// found by the replica at --metrics' address, not at the target's.
|
||||
c := newBenchCluster(t)
|
||||
out, err := c.bench(t, "--target", c.gateway.URL, "--key", "client",
|
||||
"--metrics", c.engine.URL+"/metrics", "--engine", "llamacpp")
|
||||
if err != nil {
|
||||
t.Fatalf("%v\n%s", err, out)
|
||||
}
|
||||
if got := c.metricsAuth.Load(); got != "Bearer engine-key" {
|
||||
t.Errorf("metrics read with %q", got)
|
||||
}
|
||||
if !strings.Contains(out, "target_waiting: 3") {
|
||||
t.Errorf("the queue depth read was not suggested:\n%s", out)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBenchSaysWhyAnEnginesMetricsCannotBeRead(t *testing.T) {
|
||||
// A key the engine refuses fails the run before any load, saying where
|
||||
// the right one comes from, rather than reporting an idle engine.
|
||||
c := newBenchCluster(t)
|
||||
out, err := c.bench(t, "--target", c.gateway.URL, "--key", "client",
|
||||
"--metrics", c.engine.URL+"/metrics", "--engine", "llamacpp", "--metrics-key", "wrong")
|
||||
if err == nil || !strings.Contains(err.Error(), "refused the key") || !strings.Contains(err.Error(), "--metrics-key") {
|
||||
t.Fatalf("expected the refusal explained, got %v\n%s", err, out)
|
||||
}
|
||||
}
|
||||
@@ -79,6 +79,13 @@ func TestATruncatedHelpIsSuspected(t *testing.T) {
|
||||
// trtllm-serve's help through its module, it prints the help given; asked
|
||||
// anything else, it fails, as running the launcher here would.
|
||||
func fakeInterpreter(t *testing.T, help string) string {
|
||||
t.Helper()
|
||||
return fakePython(t, "-m tensorrt_llm.commands.serve serve --help", help)
|
||||
}
|
||||
|
||||
// fakePython is an interpreter that prints help when asked exactly ask, and
|
||||
// fails on anything else — the interpreter's own --help included.
|
||||
func fakePython(t *testing.T, ask, help string) string {
|
||||
t.Helper()
|
||||
dir := t.TempDir()
|
||||
if err := os.WriteFile(filepath.Join(dir, "help.txt"), []byte(help), 0o644); err != nil {
|
||||
@@ -86,7 +93,7 @@ func fakeInterpreter(t *testing.T, help string) string {
|
||||
}
|
||||
p := filepath.Join(dir, "python")
|
||||
body := "#!/bin/sh\n" +
|
||||
`if [ "$*" = "-m tensorrt_llm.commands.serve serve --help" ]; then cat "` + filepath.Join(dir, "help.txt") + `"; exit 0; fi` + "\n" +
|
||||
`if [ "$*" = "` + ask + `" ]; then cat "` + filepath.Join(dir, "help.txt") + `"; exit 0; fi` + "\n" +
|
||||
"echo \"unexpected: $*\" >&2; exit 2\n"
|
||||
if err := os.WriteFile(p, []byte(body), 0o755); err != nil { //nolint:gosec // an executable stand-in
|
||||
t.Fatal(err)
|
||||
@@ -137,6 +144,57 @@ func TestEngineCheckAsksTrtllmServeThroughItsModule(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// sglangHelp is the part of python -m sglang.launch_server --help (0.5.21)
|
||||
// that holds the flags llmgw builds, in argparse's layout.
|
||||
const sglangHelp = `usage: sglang serve [-h] --model-path MODEL_PATH [--host HOST] [--port PORT]
|
||||
|
||||
options:
|
||||
-h, --help show this help message and exit
|
||||
--model-path MODEL_PATH, --model MODEL_PATH
|
||||
Path to model weights.
|
||||
--host HOST The host of the HTTP server.
|
||||
--port PORT The port of the HTTP server.
|
||||
--served-model-name SERVED_MODEL_NAME
|
||||
Override the model name returned by the v1/models endpoint.
|
||||
--enable-metrics Enable log prometheus metrics.
|
||||
--dist-init-addr DIST_INIT_ADDR, --nccl-init-addr DIST_INIT_ADDR
|
||||
The host address for initializing distributed backend.
|
||||
--nnodes NNODES The number of nodes.
|
||||
--node-rank NODE_RANK
|
||||
The node rank.
|
||||
--tp-size TP_SIZE, --tensor-parallel-size TP_SIZE
|
||||
The tensor parallelism size.
|
||||
`
|
||||
|
||||
func TestEngineCheckAsksSGLangThroughItsModule(t *testing.T) {
|
||||
// The command is the interpreter running llmgw's launcher; asked for
|
||||
// --help it would describe Python, and every SGLang flag would look
|
||||
// missing — or, against a real python3, be checked against nothing.
|
||||
py := fakePython(t, "-m sglang.launch_server --help", sglangHelp)
|
||||
out, err := runEngineCheck(t, "sglang", "--bin", py)
|
||||
if err != nil {
|
||||
t.Fatalf("%v\n%s", err, out)
|
||||
}
|
||||
if !strings.Contains(out, "every flag llmgw builds is accepted") {
|
||||
t.Errorf("unexpected report:\n%s", out)
|
||||
}
|
||||
if strings.Contains(out, "unexpected:") || strings.Contains(out, "skipped") {
|
||||
t.Errorf("the help was not asked of SGLang:\n%s", out)
|
||||
}
|
||||
// Two nodes by default, so the rendezvous flags were among those checked.
|
||||
if !strings.Contains(out, "sglang: 9 flags checked") {
|
||||
t.Errorf("not every flag was checked:\n%s", out)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEngineCheckCatchesAnSGLangWithoutTheFlagForItsMetrics(t *testing.T) {
|
||||
old := strings.ReplaceAll(sglangHelp, " --enable-metrics Enable log prometheus metrics.\n", "")
|
||||
out, err := runEngineCheck(t, "sglang", "--bin", fakePython(t, "-m sglang.launch_server --help", old))
|
||||
if err == nil || !strings.Contains(out, "sglang: --enable-metrics") {
|
||||
t.Errorf("an SGLang without --enable-metrics passed: %v\n%s", err, out)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEngineCheckCatchesATensorRTLLMTooOldForTheServedName(t *testing.T) {
|
||||
// TensorRT-LLM 1.2 has no --served_model_name, and a replica of it would
|
||||
// never be seen serving the name the agent waits for.
|
||||
|
||||
@@ -48,6 +48,8 @@ func MappingFor(engine string) Mapping {
|
||||
return LlamaCppMapping()
|
||||
case "trtllm":
|
||||
return TRTLLMMapping()
|
||||
case "sglang":
|
||||
return SGLangMapping()
|
||||
default:
|
||||
return VLLMMapping()
|
||||
}
|
||||
@@ -106,6 +108,39 @@ func TRTLLMMapping() Mapping {
|
||||
}
|
||||
}
|
||||
|
||||
// SGLangMapping returns SGLang's metric names, as sglang.launch_server
|
||||
// publishes them under /metrics (checked against 0.5.21's collector).
|
||||
//
|
||||
// They are there only when the server runs with --enable-metrics, which
|
||||
// llmgw's launch command passes; without it /metrics is not served at all.
|
||||
// SGLang exempts /metrics from its API key, so the key the poller sends is
|
||||
// ignored rather than needed.
|
||||
//
|
||||
// "Queue" is SGLang's word for vLLM's waiting. The gauges are set by the
|
||||
// scheduler as it reports its batches, and at least every thirty seconds
|
||||
// while it is idle, so a server that has only just started may serve none of
|
||||
// them yet and is reported incomplete until it does.
|
||||
//
|
||||
// token_usage is the fraction of the KV pool in use, as vLLM's
|
||||
// kv_cache_usage_perc is (for a model with sliding-window layers, the fuller
|
||||
// of its two pools). cache_hit_rate is a different number — how much of the
|
||||
// prompts the prefix cache answered — and is not what saturation means.
|
||||
//
|
||||
// num_requests_total counts the requests that finished, the nearest thing it
|
||||
// has to vLLM's success counter; like the token counters it is split by
|
||||
// is_streaming and summed. There is no cache configuration to read block
|
||||
// sizes from.
|
||||
func SGLangMapping() Mapping {
|
||||
return Mapping{
|
||||
Running: "sglang:num_running_reqs",
|
||||
Waiting: "sglang:num_queue_reqs",
|
||||
KVCacheUsage: "sglang:token_usage",
|
||||
PromptTokens: "sglang:prompt_tokens_total",
|
||||
GenerationTokens: "sglang:generation_tokens_total",
|
||||
SuccessTotal: "sglang:num_requests_total",
|
||||
}
|
||||
}
|
||||
|
||||
// VLLMMapping returns vLLM's current metric names.
|
||||
func VLLMMapping() Mapping {
|
||||
return Mapping{
|
||||
|
||||
@@ -0,0 +1,116 @@
|
||||
package scrape
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/router/endpoint"
|
||||
)
|
||||
|
||||
// sglangMetrics is what sglang.launch_server --enable-metrics serves on
|
||||
// /metrics (0.5.21), trimmed to the families that matter, one it must not be
|
||||
// mistaken for, and a histogram the parser has to step over. The scheduler's
|
||||
// gauges carry its ranks; the tokenizer's counters are split by whether the
|
||||
// request streamed.
|
||||
const sglangMetrics = `# HELP sglang:num_running_reqs The number of running requests.
|
||||
# TYPE sglang:num_running_reqs gauge
|
||||
sglang:num_running_reqs{engine_type="unified",model_name="qwen3-8b",moe_ep_rank="0",pp_rank="0",tp_rank="0"} 6.0
|
||||
# HELP sglang:num_queue_reqs The number of requests in the waiting queue.
|
||||
# TYPE sglang:num_queue_reqs gauge
|
||||
sglang:num_queue_reqs{engine_type="unified",model_name="qwen3-8b",moe_ep_rank="0",pp_rank="0",tp_rank="0"} 3.0
|
||||
# HELP sglang:cache_hit_rate The prefix cache hit rate.
|
||||
# TYPE sglang:cache_hit_rate gauge
|
||||
sglang:cache_hit_rate{engine_type="unified",model_name="qwen3-8b",moe_ep_rank="0",pp_rank="0",tp_rank="0"} 0.9
|
||||
# HELP sglang:token_usage The token usage.
|
||||
# TYPE sglang:token_usage gauge
|
||||
sglang:token_usage{engine_type="unified",model_name="qwen3-8b",moe_ep_rank="0",pp_rank="0",tp_rank="0"} 0.42
|
||||
# HELP sglang:prompt_tokens_total Number of prefill tokens processed.
|
||||
# TYPE sglang:prompt_tokens_total counter
|
||||
sglang:prompt_tokens_total{engine_type="unified",is_streaming="true",model_name="qwen3-8b"} 12000.0
|
||||
sglang:prompt_tokens_total{engine_type="unified",is_streaming="false",model_name="qwen3-8b"} 345.0
|
||||
# HELP sglang:generation_tokens_total Number of generation tokens processed.
|
||||
# TYPE sglang:generation_tokens_total counter
|
||||
sglang:generation_tokens_total{engine_type="unified",is_streaming="true",model_name="qwen3-8b"} 8000.0
|
||||
sglang:generation_tokens_total{engine_type="unified",is_streaming="false",model_name="qwen3-8b"} 64.0
|
||||
# HELP sglang:num_requests_total Number of requests processed.
|
||||
# TYPE sglang:num_requests_total counter
|
||||
sglang:num_requests_total{engine_type="unified",is_streaming="true",model_name="qwen3-8b"} 30.0
|
||||
sglang:num_requests_total{engine_type="unified",is_streaming="false",model_name="qwen3-8b"} 4.0
|
||||
# HELP sglang:e2e_request_latency_seconds Histogram of End-to-end request latency in seconds
|
||||
# TYPE sglang:e2e_request_latency_seconds histogram
|
||||
sglang:e2e_request_latency_seconds_bucket{engine_type="unified",is_streaming="true",le="0.5",model_name="qwen3-8b"} 12.0
|
||||
sglang:e2e_request_latency_seconds_bucket{engine_type="unified",is_streaming="true",le="+Inf",model_name="qwen3-8b"} 30.0
|
||||
sglang:e2e_request_latency_seconds_count{engine_type="unified",is_streaming="true",model_name="qwen3-8b"} 30.0
|
||||
sglang:e2e_request_latency_seconds_sum{engine_type="unified",is_streaming="true",model_name="qwen3-8b"} 41.2
|
||||
`
|
||||
|
||||
func TestSGLangMetricsSatisfyTheContract(t *testing.T) {
|
||||
s, err := Parse(strings.NewReader(sglangMetrics), MappingFor("sglang"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if s.Running != 6 || s.Waiting != 3 {
|
||||
t.Errorf("running=%d waiting=%d", s.Running, s.Waiting)
|
||||
}
|
||||
// The pool's fill, not the prefix cache's hit rate beside it.
|
||||
if s.KVCacheUsage != 0.42 {
|
||||
t.Errorf("kv=%v", s.KVCacheUsage)
|
||||
}
|
||||
// Split by is_streaming, summed.
|
||||
if s.PromptTokens != 12345 || s.GenerationTokens != 8064 {
|
||||
t.Errorf("tokens=%v/%v", s.PromptTokens, s.GenerationTokens)
|
||||
}
|
||||
if s.SuccessTotal != 34 {
|
||||
t.Errorf("success=%v", s.SuccessTotal)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSGLangIsNotReadWithVLLMsNames(t *testing.T) {
|
||||
// Read with vLLM's vocabulary, an SGLang replica serves none of the
|
||||
// contract and the router schedules to it blind.
|
||||
var missing *MissingError
|
||||
if _, err := Parse(strings.NewReader(sglangMetrics), VLLMMapping()); !errors.As(err, &missing) {
|
||||
t.Fatalf("vLLM's names found something in SGLang's metrics: %v", err)
|
||||
}
|
||||
if MappingFor("sglang").Running == VLLMMapping().Running {
|
||||
t.Error("sglang falls back to vLLM's names")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAnSGLangEndpointIsMeasuredByThePoller(t *testing.T) {
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.URL.Path != "/metrics" {
|
||||
http.NotFound(w, r)
|
||||
return
|
||||
}
|
||||
_, _ = w.Write([]byte(sglangMetrics))
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
reg := endpoint.NewRegistry(endpoint.Config{})
|
||||
reg.Replace([]endpoint.Info{{ID: "s1", URL: srv.URL, Engine: "sglang", APIKey: "k1", Models: []string{"m"}}})
|
||||
p := NewPoller(reg, Config{Interval: 20 * time.Millisecond, Timeout: 2 * time.Second}, nil)
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
p.Sync(ctx)
|
||||
defer p.Stop()
|
||||
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
for _, snap := range reg.Snapshots("m") {
|
||||
if snap.ID == "s1" && snap.Stats.Running == 6 {
|
||||
if snap.Stats.Waiting != 3 || snap.Stats.KVCacheUsage != 0.42 {
|
||||
t.Errorf("stats read wrong: %+v", snap.Stats)
|
||||
}
|
||||
return
|
||||
}
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
t.Fatal("the endpoint was never measured")
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
## CHANGELOG
|
||||
|
||||
### SGLang measured, checked and benchmarked as itself
|
||||
|
||||
- **Fixed:** the router read SGLang replicas with vLLM's metric names, and
|
||||
SGLang was started without `--enable-metrics`, so it served no `/metrics`
|
||||
at all: every scrape failed as though the replica were unreachable, quietly,
|
||||
and requests were routed to it blind. SGLang is now started with
|
||||
`--enable-metrics` and read in its own names — `sglang:num_running_reqs`,
|
||||
`sglang:num_queue_reqs`, `sglang:token_usage` (the KV pool's fill), the
|
||||
prompt and generation token counters and `sglang:num_requests_total`, as
|
||||
0.5.21 publishes them.
|
||||
- **Fixed:** `llmgw ctl engine-check sglang` asked the interpreter for
|
||||
`--help` and checked SGLang's flags against Python's. It now asks
|
||||
`python -m sglang.launch_server --help`, in the environment SGLang was
|
||||
installed into (or `--bin`).
|
||||
- **Fixed:** SGLang was given `--tp`, which argparse accepts only as an
|
||||
abbreviation of `--tp-size`; it is given `--tp-size`.
|
||||
- **Fixed:** `llmgw ctl bench --metrics` read an engine's metrics with the
|
||||
key given for `--target` — a client's key when the target is the gateway,
|
||||
or none — so llama.cpp and TensorRT-LLM, which guard their metrics with the
|
||||
engine's key, refused every read, and the run reported a queue depth of
|
||||
zero and suggested no thresholds without saying why. The engine key is now
|
||||
asked of the admin API for the replica at `--metrics`' address (or given
|
||||
with the new `--metrics-key`); metrics that cannot be read — refused, or in
|
||||
another engine's names than `--engine` — fail the run before any load,
|
||||
saying why; and a step whose metrics went unread shows `-` and the reason
|
||||
rather than zeros.
|
||||
Reference in new issue
Block a user