From 4a6723c6f22dabf07fe804f3ba07426f131b843e Mon Sep 17 00:00:00 2001 From: BMAD CI Fix Agent Date: Sun, 6 Sep 2026 16:47:13 -0500 Subject: [PATCH] feat(bench): add local-coder benchmark harness --- Makefile | 2 +- README.md | 12 ++ cmd/devrail-router/main.go | 30 ++++ docs/benchmarking.md | 76 ++++++++ internal/bench/bench.go | 276 ++++++++++++++++++++++++++++++ internal/bench/bench_test.go | 182 ++++++++++++++++++++ test/bench/local-coder.cases.json | 23 +++ 7 files changed, 600 insertions(+), 1 deletion(-) create mode 100644 docs/benchmarking.md create mode 100644 internal/bench/bench.go create mode 100644 internal/bench/bench_test.go create mode 100644 test/bench/local-coder.cases.json diff --git a/Makefile b/Makefile index 2385688..a85605f 100644 --- a/Makefile +++ b/Makefile @@ -139,7 +139,7 @@ package: build ## Build a Linux/macOS tarball package "$(DIST_DIR)/$(PACKAGE_NAME)/packaging/systemd" \ "$(DIST_DIR)/$(PACKAGE_NAME)/packaging/linux" cp configs/router.example.yaml "$(DIST_DIR)/$(PACKAGE_NAME)/configs/" - cp docs/architecture.md docs/packaging.md docs/roadmap.md "$(DIST_DIR)/$(PACKAGE_NAME)/docs/" + cp docs/architecture.md docs/benchmarking.md docs/packaging.md docs/roadmap.md "$(DIST_DIR)/$(PACKAGE_NAME)/docs/" cp packaging/systemd/devrail-router.service "$(DIST_DIR)/$(PACKAGE_NAME)/packaging/systemd/" cp packaging/linux/install.sh "$(DIST_DIR)/$(PACKAGE_NAME)/packaging/linux/" chmod 0755 "$(DIST_DIR)/$(PACKAGE_NAME)/devrail-router" \ diff --git a/README.md b/README.md index a04f826..7a53ee2 100644 --- a/README.md +++ b/README.md @@ -34,6 +34,7 @@ This repository is in early foundation work. The current service supports: first event latency, bytes, and token totals - consistent OpenAI-shaped errors for router-side failures - request IDs in router responses and logs +- a streamed benchmark harness for comparing model aliases with fixed prompts Routing policy, auth, native LM Studio lifecycle integration, richer telemetry, and Omarchy integration are planned next. See [docs/roadmap.md](docs/roadmap.md). @@ -100,6 +101,17 @@ The Compose stack starts DevRail Router plus a mock OpenAI-compatible backend and verifies health, model listing, alias rewriting, backend auth injection, and chat completion proxying. +Run repeatable streamed benchmark cases against a router alias: + +```sh +go run ./cmd/devrail-router bench \ + -base-url http://llm-srv-01.mfsoho.linkridge.net:18080/v1 \ + -model local-coder \ + -cases test/bench/local-coder.cases.json +``` + +See [docs/benchmarking.md](docs/benchmarking.md). + ## Configuration See `configs/router.example.yaml`. diff --git a/cmd/devrail-router/main.go b/cmd/devrail-router/main.go index 5c1ea53..e0f5c05 100644 --- a/cmd/devrail-router/main.go +++ b/cmd/devrail-router/main.go @@ -11,6 +11,7 @@ import ( "syscall" "time" + "github.com/devrail-dev/devrail-router/internal/bench" "github.com/devrail-dev/devrail-router/internal/config" "github.com/devrail-dev/devrail-router/internal/server" ) @@ -31,6 +32,8 @@ func run(args []string) int { return serve(args[1:]) case "check": return check(args[1:]) + case "bench": + return runBench(args[1:]) case "version": fmt.Println(version) return 0 @@ -97,6 +100,32 @@ func serve(args []string) int { return 0 } +func runBench(args []string) int { + fs := flag.NewFlagSet("bench", flag.ContinueOnError) + baseURL := fs.String("base-url", "http://127.0.0.1:8080/v1", "OpenAI-compatible base URL") + model := fs.String("model", "local-coder", "model alias to benchmark") + casesPath := fs.String("cases", "test/bench/local-coder.cases.json", "JSON benchmark cases file") + apiKey := fs.String("api-key", os.Getenv("OPENAI_API_KEY"), "API key for the target endpoint") + maxTokens := fs.Int("max-tokens", 512, "max completion tokens per case") + timeout := fs.Duration("timeout", 5*time.Minute, "timeout per benchmark case") + if err := fs.Parse(args); err != nil { + return 2 + } + + if err := bench.Run(context.Background(), bench.Options{ + BaseURL: *baseURL, + Model: *model, + APIKey: *apiKey, + CasesPath: *casesPath, + MaxTokens: *maxTokens, + Timeout: *timeout, + }); err != nil { + slog.Error("benchmark failed", "error", err) + return 1 + } + return 0 +} + func check(args []string) int { fs := flag.NewFlagSet("check", flag.ContinueOnError) configPath := fs.String("config", config.DefaultPath, "path to router config") @@ -120,6 +149,7 @@ func usage() { Usage: devrail-router serve [-config path] devrail-router check [-config path] + devrail-router bench [-base-url url] [-model alias] [-cases path] devrail-router version `, version) diff --git a/docs/benchmarking.md b/docs/benchmarking.md new file mode 100644 index 0000000..5ded15d --- /dev/null +++ b/docs/benchmarking.md @@ -0,0 +1,76 @@ +# Benchmarking + +DevRail Router includes a small streamed benchmark harness for comparing +OpenAI-compatible model aliases such as `local-coder`. + +The harness sends fixed chat-completion cases with `stream=true` and +`stream_options.include_usage=true`, then writes one JSON object per case to +stdout. Each result captures: + +- case ID +- model alias +- router request ID, when returned +- HTTP status +- time to first SSE event +- total request duration +- response bytes +- prompt, completion, and total tokens when the backend emits streamed usage +- a short first-content sample for sanity checking + +## Local-Coder Baseline + +Run the default coding-oriented cases against the llm-srv router: + +```sh +go run ./cmd/devrail-router bench \ + -base-url http://llm-srv-01.mfsoho.linkridge.net:18080/v1 \ + -model local-coder \ + -cases test/bench/local-coder.cases.json \ + -max-tokens 512 +``` + +Save a baseline: + +```sh +go run ./cmd/devrail-router bench \ + -base-url http://llm-srv-01.mfsoho.linkridge.net:18080/v1 \ + -model local-coder \ + -cases test/bench/local-coder.cases.json \ + -max-tokens 512 \ + > local-coder-baseline.jsonl +``` + +Run the same cases against another alias, such as an experimental +`local-coder-parallel`, by changing `-model` only. Keeping the case file and +token cap stable makes queue wait, first-token latency, duration, and token +throughput easier to compare in Grafana. + +## Custom Cases + +Case files are JSON arrays. A case can use a simple `prompt`: + +```json +[ + { + "id": "small-refactor", + "prompt": "Refactor this Go function and explain the tradeoff." + } +] +``` + +Or an explicit OpenAI-style message list: + +```json +[ + { + "id": "reviewer", + "messages": [ + {"role": "system", "content": "You are a concise Go reviewer."}, + {"role": "user", "content": "Find the highest-risk bug in this proxy."} + ] + } +] +``` + +Use stable, short IDs. They appear in JSONL output and make it easier to line +up command results with router request IDs, logs, and Prometheus samples. diff --git a/internal/bench/bench.go b/internal/bench/bench.go new file mode 100644 index 0000000..f31266a --- /dev/null +++ b/internal/bench/bench.go @@ -0,0 +1,276 @@ +package bench + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "os" + "strings" + "time" +) + +type Case struct { + ID string `json:"id"` + Prompt string `json:"prompt"` + Messages []Message `json:"messages,omitempty"` +} + +type Message struct { + Role string `json:"role"` + Content string `json:"content"` +} + +type Result struct { + CaseID string `json:"case_id"` + Model string `json:"model"` + RequestID string `json:"request_id,omitempty"` + Status int `json:"status"` + Streaming bool `json:"streaming"` + FirstEventMS int64 `json:"first_event_ms"` + DurationMS int64 `json:"duration_ms"` + ResponseBytes int64 `json:"response_bytes"` + PromptTokens int `json:"prompt_tokens,omitempty"` + CompletionTokens int `json:"completion_tokens,omitempty"` + TotalTokens int `json:"total_tokens,omitempty"` + Error string `json:"error,omitempty"` + FirstContentSample string `json:"first_content_sample,omitempty"` +} + +type Options struct { + BaseURL string + Model string + APIKey string + CasesPath string + Output io.Writer + HTTPClient *http.Client + MaxTokens int + Timeout time.Duration +} + +type streamUsage struct { + PromptTokens int `json:"prompt_tokens"` + CompletionTokens int `json:"completion_tokens"` + TotalTokens int `json:"total_tokens"` +} + +type streamChunk struct { + Choices []struct { + Delta struct { + Content string `json:"content"` + } `json:"delta"` + } `json:"choices"` + Usage *streamUsage `json:"usage"` +} + +func Run(ctx context.Context, opts Options) error { + if strings.TrimSpace(opts.BaseURL) == "" { + return errors.New("base URL is required") + } + if strings.TrimSpace(opts.Model) == "" { + return errors.New("model is required") + } + if opts.Output == nil { + opts.Output = os.Stdout + } + if opts.HTTPClient == nil { + opts.HTTPClient = http.DefaultClient + } + if opts.Timeout == 0 { + opts.Timeout = 5 * time.Minute + } + + cases, err := LoadCases(opts.CasesPath) + if err != nil { + return err + } + encoder := json.NewEncoder(opts.Output) + for _, benchCase := range cases { + result := RunCase(ctx, opts, benchCase) + if err := encoder.Encode(result); err != nil { + return fmt.Errorf("write result: %w", err) + } + } + return nil +} + +func LoadCases(path string) ([]Case, error) { + if strings.TrimSpace(path) == "" { + return nil, errors.New("cases path is required") + } + data, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("read cases: %w", err) + } + + var cases []Case + if err := json.Unmarshal(data, &cases); err != nil { + return nil, fmt.Errorf("decode cases: %w", err) + } + for i, benchCase := range cases { + if strings.TrimSpace(benchCase.ID) == "" { + return nil, fmt.Errorf("case %d missing id", i) + } + if strings.TrimSpace(benchCase.Prompt) == "" && len(benchCase.Messages) == 0 { + return nil, fmt.Errorf("case %q needs prompt or messages", benchCase.ID) + } + } + return cases, nil +} + +func RunCase(ctx context.Context, opts Options, benchCase Case) Result { + result := Result{CaseID: benchCase.ID, Model: opts.Model, Streaming: true, FirstEventMS: -1} + timeoutCtx, cancel := context.WithTimeout(ctx, opts.Timeout) + defer cancel() + + body, err := requestBody(opts.Model, opts.MaxTokens, benchCase) + if err != nil { + result.Error = err.Error() + return result + } + + endpoint, err := chatCompletionsURL(opts.BaseURL) + if err != nil { + result.Error = err.Error() + return result + } + + req, err := http.NewRequestWithContext(timeoutCtx, http.MethodPost, endpoint, bytes.NewReader(body)) + if err != nil { + result.Error = err.Error() + return result + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "text/event-stream") + if opts.APIKey != "" { + req.Header.Set("Authorization", "Bearer "+opts.APIKey) + } + + started := time.Now() + resp, err := opts.HTTPClient.Do(req) + if err != nil { + result.DurationMS = time.Since(started).Milliseconds() + result.Error = err.Error() + return result + } + defer func() { + _ = resp.Body.Close() + }() + + result.Status = resp.StatusCode + result.RequestID = resp.Header.Get("X-Devrail-Request-ID") + if !strings.Contains(resp.Header.Get("Content-Type"), "text/event-stream") { + body, readErr := io.ReadAll(resp.Body) + result.DurationMS = time.Since(started).Milliseconds() + result.ResponseBytes = int64(len(body)) + if readErr != nil { + result.Error = readErr.Error() + return result + } + result.Error = fmt.Sprintf("expected text/event-stream response, got status %d: %s", resp.StatusCode, truncate(strings.TrimSpace(string(body)), 240)) + return result + } + + usage, firstSample, bytesRead, firstEvent, err := ObserveSSE(resp.Body, started) + result.DurationMS = time.Since(started).Milliseconds() + result.ResponseBytes = bytesRead + result.FirstEventMS = firstEvent + result.FirstContentSample = firstSample + if usage != nil { + result.PromptTokens = usage.PromptTokens + result.CompletionTokens = usage.CompletionTokens + result.TotalTokens = usage.TotalTokens + } + if err != nil { + result.Error = err.Error() + } + return result +} + +func ObserveSSE(r io.Reader, started time.Time) (*streamUsage, string, int64, int64, error) { + scanner := bufio.NewScanner(r) + scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) + + var usage *streamUsage + var firstSample string + firstEventMS := int64(-1) + var bytesRead int64 + + for scanner.Scan() { + line := scanner.Text() + bytesRead += int64(len(line) + 1) + if !strings.HasPrefix(line, "data:") { + continue + } + payload := strings.TrimSpace(strings.TrimPrefix(line, "data:")) + if payload == "" || payload == "[DONE]" { + continue + } + if firstEventMS < 0 { + firstEventMS = time.Since(started).Milliseconds() + } + + var chunk streamChunk + if err := json.Unmarshal([]byte(payload), &chunk); err != nil { + return usage, firstSample, bytesRead, firstEventMS, fmt.Errorf("decode stream chunk: %w", err) + } + if chunk.Usage != nil { + usage = chunk.Usage + } + if firstSample == "" { + for _, choice := range chunk.Choices { + if choice.Delta.Content != "" { + firstSample = truncate(choice.Delta.Content, 120) + break + } + } + } + } + if err := scanner.Err(); err != nil { + return usage, firstSample, bytesRead, firstEventMS, err + } + return usage, firstSample, bytesRead, firstEventMS, nil +} + +func requestBody(model string, maxTokens int, benchCase Case) ([]byte, error) { + messages := benchCase.Messages + if len(messages) == 0 { + messages = []Message{{Role: "user", Content: benchCase.Prompt}} + } + body := map[string]any{ + "model": model, + "messages": messages, + "stream": true, + "stream_options": map[string]bool{ + "include_usage": true, + }, + } + if maxTokens > 0 { + body["max_tokens"] = maxTokens + } + return json.Marshal(body) +} + +func chatCompletionsURL(base string) (string, error) { + parsed, err := url.Parse(base) + if err != nil { + return "", fmt.Errorf("parse base URL: %w", err) + } + if parsed.Scheme == "" || parsed.Host == "" { + return "", fmt.Errorf("base URL must include scheme and host: %q", base) + } + parsed.Path = strings.TrimRight(parsed.Path, "/") + "/chat/completions" + return parsed.String(), nil +} + +func truncate(value string, limit int) string { + if len(value) <= limit { + return value + } + return value[:limit] +} diff --git a/internal/bench/bench_test.go b/internal/bench/bench_test.go new file mode 100644 index 0000000..0d9edf7 --- /dev/null +++ b/internal/bench/bench_test.go @@ -0,0 +1,182 @@ +package bench + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +func TestObserveSSECapturesFirstContentAndUsage(t *testing.T) { + t.Parallel() + + stream := strings.NewReader(strings.Join([]string{ + ": keepalive", + `data: {"choices":[{"delta":{"content":"hello"}}]}`, + `data: {"choices":[{"delta":{"content":" world"}}]}`, + `data: {"choices":[],"usage":{"prompt_tokens":12,"completion_tokens":3,"total_tokens":15}}`, + "data: [DONE]", + "", + }, "\n")) + + usage, firstSample, bytesRead, firstEventMS, err := ObserveSSE(stream, time.Now().Add(-150*time.Millisecond)) + if err != nil { + t.Fatalf("observe stream: %v", err) + } + if firstSample != "hello" { + t.Fatalf("unexpected first sample: %q", firstSample) + } + if firstEventMS < 100 { + t.Fatalf("first event too small: %d", firstEventMS) + } + if bytesRead == 0 { + t.Fatal("expected bytes to be counted") + } + if usage == nil { + t.Fatal("expected streamed usage") + } + if usage.PromptTokens != 12 || usage.CompletionTokens != 3 || usage.TotalTokens != 15 { + t.Fatalf("unexpected usage: %+v", usage) + } +} + +func TestLoadCasesValidatesInput(t *testing.T) { + t.Parallel() + + path := filepath.Join(t.TempDir(), "cases.json") + if err := os.WriteFile(path, []byte(`[{"id":"small","prompt":"write a test"}]`), 0o644); err != nil { + t.Fatalf("write cases: %v", err) + } + + cases, err := LoadCases(path) + if err != nil { + t.Fatalf("load cases: %v", err) + } + if len(cases) != 1 || cases[0].ID != "small" { + t.Fatalf("unexpected cases: %+v", cases) + } +} + +func TestRunCaseStreamsAgainstOpenAICompatibleBackend(t *testing.T) { + t.Parallel() + + var requestPayload struct { + Model string `json:"model"` + Messages []Message `json:"messages"` + Stream bool `json:"stream"` + StreamOptions struct { + IncludeUsage bool `json:"include_usage"` + } `json:"stream_options"` + MaxTokens int `json:"max_tokens"` + } + + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1/chat/completions" { + t.Errorf("unexpected path: %s", r.URL.Path) + } + if err := json.NewDecoder(r.Body).Decode(&requestPayload); err != nil { + t.Errorf("decode request: %v", err) + } + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("X-Devrail-Request-ID", "req-123") + _, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"ok\"}}]}\n\n")) + _, _ = w.Write([]byte("data: {\"choices\":[],\"usage\":{\"prompt_tokens\":4,\"completion_tokens\":2,\"total_tokens\":6}}\n\n")) + _, _ = w.Write([]byte("data: [DONE]\n\n")) + })) + t.Cleanup(backend.Close) + + result := RunCase(context.Background(), Options{ + BaseURL: backend.URL + "/v1", + Model: "local-coder", + HTTPClient: backend.Client(), + MaxTokens: 64, + Timeout: time.Second, + }, Case{ID: "smoke", Prompt: "reply ok"}) + + if result.Error != "" { + t.Fatalf("unexpected result error: %s", result.Error) + } + if result.Status != http.StatusOK { + t.Fatalf("unexpected status: %d", result.Status) + } + if result.RequestID != "req-123" { + t.Fatalf("unexpected request id: %q", result.RequestID) + } + if result.PromptTokens != 4 || result.CompletionTokens != 2 || result.TotalTokens != 6 { + t.Fatalf("unexpected usage: %+v", result) + } + if result.FirstContentSample != "ok" { + t.Fatalf("unexpected first sample: %q", result.FirstContentSample) + } + if requestPayload.Model != "local-coder" || !requestPayload.Stream { + t.Fatalf("unexpected request payload: %+v", requestPayload) + } + if !requestPayload.StreamOptions.IncludeUsage { + t.Fatal("expected streamed usage option") + } + if requestPayload.MaxTokens != 64 { + t.Fatalf("unexpected max tokens: %d", requestPayload.MaxTokens) + } +} + +func TestRunCaseReportsNonStreamResponse(t *testing.T) { + t.Parallel() + + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(`{"error":{"message":"bad model"}}`)) + })) + t.Cleanup(backend.Close) + + result := RunCase(context.Background(), Options{ + BaseURL: backend.URL + "/v1", + Model: "local-coder", + HTTPClient: backend.Client(), + Timeout: time.Second, + }, Case{ID: "bad", Prompt: "hello"}) + + if result.Status != http.StatusBadRequest { + t.Fatalf("unexpected status: %d", result.Status) + } + if result.ResponseBytes == 0 { + t.Fatal("expected response bytes") + } + if !strings.Contains(result.Error, "expected text/event-stream response") { + t.Fatalf("unexpected error: %q", result.Error) + } +} + +func TestRunWritesJSONL(t *testing.T) { + t.Parallel() + + path := filepath.Join(t.TempDir(), "cases.json") + if err := os.WriteFile(path, []byte(`[{"id":"bad-url","prompt":"hello"}]`), 0o644); err != nil { + t.Fatalf("write cases: %v", err) + } + + var out bytes.Buffer + err := Run(context.Background(), Options{ + BaseURL: "not-a-url", + Model: "local-coder", + CasesPath: path, + Output: &out, + }) + if err != nil { + t.Fatalf("run: %v", err) + } + + var result Result + if err := json.Unmarshal(bytes.TrimSpace(out.Bytes()), &result); err != nil { + t.Fatalf("decode jsonl: %v", err) + } + if result.CaseID != "bad-url" || result.Error == "" { + t.Fatalf("unexpected result: %+v", result) + } +} diff --git a/test/bench/local-coder.cases.json b/test/bench/local-coder.cases.json new file mode 100644 index 0000000..86b9a82 --- /dev/null +++ b/test/bench/local-coder.cases.json @@ -0,0 +1,23 @@ +[ + { + "id": "small-refactor", + "prompt": "You are reviewing a Go function that has duplicated nil checks. Describe a small refactor, then provide only the revised function body." + }, + { + "id": "test-plan", + "prompt": "Given a local-first LLM router with request queueing, streaming responses, and Prometheus metrics, write a focused test plan with five high-value checks." + }, + { + "id": "bug-hunt", + "messages": [ + { + "role": "system", + "content": "You are a senior Go reviewer. Be concise." + }, + { + "role": "user", + "content": "Find the most likely production risk in a reverse proxy that rewrites JSON request bodies, forwards streaming SSE responses, and records latency metrics. Explain the risk and one mitigation." + } + ] + } +]