From 8cc8d7139b04d68141e53de5dfec4a2ebdb82fbe Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Mon, 5 Oct 2026 00:20:25 +0200 Subject: [PATCH 1/2] SGLang read in its own metric names, started serving them, and asked for its own help The router read SGLang replicas with vLLM's names, and SGLang was started without --enable-metrics, so it served no /metrics: every scrape failed as if unreachable and the replica was routed to blind. It now gets --enable-metrics and a mapping of its own (sglang:num_running_reqs, num_queue_reqs, token_usage, the token counters and num_requests_total, as 0.5.21 publishes them). engine-check asks python -m sglang.launch_server --help instead of Python's own help, and SGLang is given --tp-size rather than the --tp argparse only takes as an abbreviation. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/install.md | 5 ++ internal/agent/engines/engines_test.go | 20 +++++ internal/agent/engines/sglang.go | 18 +++- internal/cli/enginecheck_test.go | 60 ++++++++++++- internal/router/scrape/parse.go | 35 ++++++++ internal/router/scrape/sglang_test.go | 116 +++++++++++++++++++++++++ 6 files changed, 251 insertions(+), 3 deletions(-) create mode 100644 internal/router/scrape/sglang_test.go diff --git a/docs/install.md b/docs/install.md index 114315a7..68be7040 100644 --- a/docs/install.md +++ b/docs/install.md @@ -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 /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`, diff --git a/internal/agent/engines/engines_test.go b/internal/agent/engines/engines_test.go index c795bcf7..7d0fb9dc 100644 --- a/internal/agent/engines/engines_test.go +++ b/internal/agent/engines/engines_test.go @@ -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"} { diff --git a/internal/agent/engines/sglang.go b/internal/agent/engines/sglang.go index ccaa5722..fa24a8c6 100644 --- a/internal/agent/engines/sglang.go +++ b/internal/agent/engines/sglang.go @@ -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) diff --git a/internal/cli/enginecheck_test.go b/internal/cli/enginecheck_test.go index 1b4d85f4..ac40b0a6 100644 --- a/internal/cli/enginecheck_test.go +++ b/internal/cli/enginecheck_test.go @@ -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. diff --git a/internal/router/scrape/parse.go b/internal/router/scrape/parse.go index ec35a65c..6b4d6161 100644 --- a/internal/router/scrape/parse.go +++ b/internal/router/scrape/parse.go @@ -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{ diff --git a/internal/router/scrape/sglang_test.go b/internal/router/scrape/sglang_test.go new file mode 100644 index 00000000..0c9ebe11 --- /dev/null +++ b/internal/router/scrape/sglang_test.go @@ -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") +} From 0416b827fd439ec59ad55071eb58813f25d0e11e Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Mon, 5 Oct 2026 00:20:25 +0200 Subject: [PATCH 2/2] ctl bench reads a guarded engine's metrics with the engine's key, or says why it cannot The sampler sent the key given for --target, which for the gateway is a client's, so llama.cpp and TensorRT-LLM refused every read and the run reported an idle engine. The engine key now comes from the admin API for the replica at --metrics' address, or --metrics-key; unreadable metrics stop the run before any load, and a step that read none shows why. Co-Authored-By: Claude Opus 5.5 (1M context) --- internal/bench/bench.go | 28 ++++++- internal/bench/sampler.go | 66 ++++++++++++++-- internal/bench/sampler_test.go | 133 +++++++++++++++++++++++++++++++++ internal/cli/bench.go | 78 +++++++++++++++---- internal/cli/bench_test.go | 96 ++++++++++++++++++++++++ notes/engine-fixes.md | 28 +++++++ 6 files changed, 404 insertions(+), 25 deletions(-) create mode 100644 internal/bench/sampler_test.go create mode 100644 internal/cli/bench_test.go create mode 100644 notes/engine-fixes.md diff --git a/internal/bench/bench.go b/internal/bench/bench.go index 2da271e5..f6f9d298 100644 --- a/internal/bench/bench.go +++ b/internal/bench/bench.go @@ -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 && diff --git a/internal/bench/sampler.go b/internal/bench/sampler.go index db2fcf31..9c926abb 100644 --- a/internal/bench/sampler.go +++ b/internal/bench/sampler.go @@ -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() +} diff --git a/internal/bench/sampler_test.go b/internal/bench/sampler_test.go new file mode 100644 index 00000000..ae80fdf2 --- /dev/null +++ b/internal/bench/sampler_test.go @@ -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) + } +} diff --git a/internal/cli/bench.go b/internal/cli/bench.go index b17f7f1e..60c737df 100644 --- a/internal/cli/bench.go +++ b/internal/cli/bench.go @@ -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 +} diff --git a/internal/cli/bench_test.go b/internal/cli/bench_test.go new file mode 100644 index 00000000..63f2241c --- /dev/null +++ b/internal/cli/bench_test.go @@ -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) + } +} diff --git a/notes/engine-fixes.md b/notes/engine-fixes.md new file mode 100644 index 00000000..fbd202d5 --- /dev/null +++ b/notes/engine-fixes.md @@ -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.