diff --git a/.traefik.yml b/.traefik.yml index aa55771..107c5dc 100644 --- a/.traefik.yml +++ b/.traefik.yml @@ -1,6 +1,17 @@ displayName: LLM Gateway type: middleware import: github.com/lukaszraczylo/traefik-llmgateway +# basePkg: Traefik's Yaegi middleware loader (pkg/plugins/middlewareyaegi.go, +# newYaegiMiddlewareBuilder) binds the interpreted package under +# strings.ReplaceAll(path.Base(import), "-", "_") — "traefik_llmgateway" — +# unless this field overrides it. The package itself declares +# `package traefikllmgateway` (no underscore), so without this override a +# real Traefik instance fails plugin load with +# `failed to eval New: ...: undefined: traefik_llmgateway`, discovered +# running the plugin in real Traefik v3.5 for the first time (Task 15's +# integration suite). Required for both Traefik itself and the Plugin +# Catalog analyzer (piceus), which uses this identical mechanism. +basePkg: traefikllmgateway summary: Multi-provider LLM gateway with groups, per-user keys, token/cost limits, MCP/A2A proxy. testData: providers: diff --git a/Makefile b/Makefile index fd66adc..e6440a5 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: test lint yaegi-check integration +.PHONY: test lint yaegi-check integration integration-keep integration-up integration-wait integration-down test: go test ./... -count=1 @@ -15,5 +15,38 @@ lint: yaegi-check: cd tools/yaegi-check && GOWORK=off go run . $(CURDIR) -integration: - @echo "integration: placeholder until Task 15" +INTEGRATION_COMPOSE := integration/docker-compose.yml +INTEGRATION_HEALTH_URL := http://localhost:19081/v1/models +INTEGRATION_ALICE_KEY := sk-int-alice + +integration-up: + docker compose -f $(INTEGRATION_COMPOSE) up -d --build + +integration-wait: + @i=0; \ + until curl -sf -o /dev/null -H "Authorization: Bearer $(INTEGRATION_ALICE_KEY)" $(INTEGRATION_HEALTH_URL); do \ + i=$$((i+1)); \ + if [ $$i -ge 120 ]; then \ + echo "integration: traefik1 did not become healthy within 120s"; \ + docker compose -f $(INTEGRATION_COMPOSE) logs traefik1; \ + exit 1; \ + fi; \ + sleep 1; \ + done + +integration-down: + docker compose -f $(INTEGRATION_COMPOSE) down -v + +# integration brings the stack up, waits for it to answer, runs the suite, +# then always tears the stack down (pass or fail) so the exit code +# reflects the tests alone. +integration: integration-up integration-wait + @go -C integration test -tags integration -count=1 ./...; status=$$?; \ + $(MAKE) integration-down; \ + exit $$status + +# integration-keep is the debugging variant: same run, but leaves the +# compose stack up afterward for inspecting logs/state by hand. Tear it +# down yourself with `make integration-down` when done. +integration-keep: integration-up integration-wait + go -C integration test -tags integration -count=1 ./... diff --git a/integration/docker-compose.yml b/integration/docker-compose.yml new file mode 100644 index 0000000..d62226d --- /dev/null +++ b/integration/docker-compose.yml @@ -0,0 +1,122 @@ +networks: + llmgw-integration: + driver: bridge + +services: + redis: + image: redis:7-alpine + networks: ["llmgw-integration"] + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 2s + timeout: 2s + retries: 15 + + mockopenai: + build: + context: . + dockerfile: mock/Dockerfile + command: ["-mode", "openai"] + networks: ["llmgw-integration"] + healthcheck: + test: ["CMD", "wget", "-qO-", "http://localhost:8080/healthz"] + interval: 2s + timeout: 2s + retries: 15 + + mockanthropic: + build: + context: . + dockerfile: mock/Dockerfile + command: ["-mode", "anthropic"] + networks: ["llmgw-integration"] + healthcheck: + test: ["CMD", "wget", "-qO-", "http://localhost:8080/healthz"] + interval: 2s + timeout: 2s + retries: 15 + + mockgemini: + build: + context: . + dockerfile: mock/Dockerfile + command: ["-mode", "gemini"] + networks: ["llmgw-integration"] + healthcheck: + test: ["CMD", "wget", "-qO-", "http://localhost:8080/healthz"] + interval: 2s + timeout: 2s + retries: 15 + + mocktool: + build: + context: . + dockerfile: mock/Dockerfile + command: ["-mode", "tool"] + networks: ["llmgw-integration"] + healthcheck: + test: ["CMD", "wget", "-qO-", "http://localhost:8080/healthz"] + interval: 2s + timeout: 2s + retries: 15 + + traefik1: + image: traefik:v3.5 + command: ["--configfile=/etc/traefik/traefik.yml"] + networks: ["llmgw-integration"] + ports: + - "19081:80" + volumes: + - ../:/plugins-local/src/github.com/lukaszraczylo/traefik-llmgateway:ro + - ./traefik/traefik.yml:/etc/traefik/traefik.yml:ro + - ./traefik/dynamic.yml:/etc/traefik/dynamic.yml:ro + - ./users:/users:ro + depends_on: + redis: + condition: service_healthy + mockopenai: + condition: service_healthy + mockanthropic: + condition: service_healthy + mockgemini: + condition: service_healthy + mocktool: + condition: service_healthy + + traefik2: + image: traefik:v3.5 + command: ["--configfile=/etc/traefik/traefik.yml"] + networks: ["llmgw-integration"] + ports: + - "19082:80" + volumes: + - ../:/plugins-local/src/github.com/lukaszraczylo/traefik-llmgateway:ro + - ./traefik/traefik.yml:/etc/traefik/traefik.yml:ro + - ./traefik/dynamic.yml:/etc/traefik/dynamic.yml:ro + - ./users:/users:ro + depends_on: + redis: + condition: service_healthy + mockopenai: + condition: service_healthy + mockanthropic: + condition: service_healthy + mockgemini: + condition: service_healthy + mocktool: + condition: service_healthy + + # In-network streaming timing prober. Never started by `up`; the test + # invokes it on demand via `docker compose run --rm probe`, from inside + # this same compose network so its raw-TCP read against traefik1 + # reflects true inter-chunk arrival timing instead of the buffered + # behaviour Docker Desktop's host port-proxy would introduce. + probe: + build: + context: . + dockerfile: mock/Dockerfile + command: ["-probe", "-target", "traefik1:80", "-apikey", "sk-int-alice", "-model", "openai/gpt-mock"] + networks: ["llmgw-integration"] + profiles: ["probe"] + depends_on: + - traefik1 diff --git a/integration/go.mod b/integration/go.mod new file mode 100644 index 0000000..b54a2b5 --- /dev/null +++ b/integration/go.mod @@ -0,0 +1,3 @@ +module github.com/lukaszraczylo/traefik-llmgateway/integration + +go 1.22 diff --git a/integration/integration_test.go b/integration/integration_test.go new file mode 100644 index 0000000..40bd7a5 --- /dev/null +++ b/integration/integration_test.go @@ -0,0 +1,403 @@ +//go:build integration + +// Package integration exercises the llmgateway Traefik plugin against a +// real Traefik v3.5 binary loading the plugin via localPlugins, real mock +// upstreams, and a real shared Redis, all orchestrated by +// integration/docker-compose.yml. It assumes the compose stack is already +// up and healthy — the `integration` Makefile target starts it, waits for +// traefik1 to answer, runs this package, then tears the stack down; run it +// directly with `go test -tags integration ./...` from this directory +// after `docker compose up -d --build` if you want to keep the stack alive +// between runs (see the `integration-keep` Makefile target). +package integration + +import ( + "bytes" + "encoding/json" + "io" + "net/http" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" +) + +const ( + traefik1URL = "http://localhost:19081" + traefik2URL = "http://localhost:19082" + + aliceKey = "sk-int-alice" // eng group, high limit, inline user + bobKey = "sk-int-bob" // limited group (requestsPerMinute: 3), file-sourced user + carolKey = "sk-int-carol" // added to users.json mid-run by TestUsersFileHotReload +) + +// composeFile returns the path to docker-compose.yml relative to this +// package's directory, which is also `go test`'s working directory. +func composeFile() string { return "docker-compose.yml" } + +// doJSON issues one JSON request, returning the raw *http.Response +// (headers and status code still valid — its body has already been read +// and closed) alongside the decoded JSON body. A non-JSON or empty body +// decodes to a nil map rather than failing the test; callers that require +// a body assert on it themselves. +func doJSON(t *testing.T, method, url, apiKey string, body any) (*http.Response, map[string]any) { + t.Helper() + + var reader io.Reader + if body != nil { + b, err := json.Marshal(body) + if err != nil { + t.Fatalf("marshal request body: %v", err) + } + reader = bytes.NewReader(b) + } + + req, err := http.NewRequest(method, url, reader) + if err != nil { + t.Fatalf("new request %s %s: %v", method, url, err) + } + if apiKey != "" { + req.Header.Set("Authorization", "Bearer "+apiKey) + } + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("do request %s %s: %v", method, url, err) + } + defer func() { _ = resp.Body.Close() }() + + raw, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("read response body from %s %s: %v", method, url, err) + } + var out map[string]any + if len(raw) > 0 { + _ = json.Unmarshal(raw, &out) // non-JSON body (e.g. an SSE stream) just yields a nil map + } + return resp, out +} + +// firstChoiceContent extracts choices[0].message.content from an +// OpenAI-shaped chat completion response body, failing the test if the +// shape doesn't match. +func firstChoiceContent(t *testing.T, body map[string]any) string { + t.Helper() + choices, ok := body["choices"].([]any) + if !ok || len(choices) == 0 { + t.Fatalf("response has no choices: %#v", body) + } + choice, ok := choices[0].(map[string]any) + if !ok { + t.Fatalf("choices[0] is not an object: %#v", choices[0]) + } + message, ok := choice["message"].(map[string]any) + if !ok { + t.Fatalf("choices[0].message is not an object: %#v", choice) + } + content, _ := message["content"].(string) + return content +} + +// waitForHealthy polls url with apiKey until it answers 200 or timeout +// elapses. +func waitForHealthy(t *testing.T, url, apiKey string, timeout time.Duration) { + t.Helper() + deadline := time.Now().Add(timeout) + var lastErr error + var lastStatus int + for time.Now().Before(deadline) { + req, err := http.NewRequest(http.MethodGet, url, nil) + if err != nil { + t.Fatalf("new request: %v", err) + } + req.Header.Set("Authorization", "Bearer "+apiKey) + resp, err := http.DefaultClient.Do(req) + if err != nil { + lastErr = err + time.Sleep(time.Second) + continue + } + _ = resp.Body.Close() + lastStatus = resp.StatusCode + if resp.StatusCode == http.StatusOK { + return + } + time.Sleep(time.Second) + } + t.Fatalf("%s did not become healthy within %s (last status=%d, last error=%v)", url, timeout, lastStatus, lastErr) +} + +// checkNoPluginErrors greps docker compose's captured logs for service, +// failing if no plugin-loading message appears at all, or if any +// plugin-related line is logged at error level. +func checkNoPluginErrors(t *testing.T, service string) { + t.Helper() + out, err := exec.Command("docker", "compose", "-f", composeFile(), "logs", service).CombinedOutput() + if err != nil { + t.Fatalf("docker compose logs %s: %v\n%s", service, err, out) + } + logs := string(out) + if !strings.Contains(strings.ToLower(logs), "plugin") { + t.Fatalf("%s logs contain no plugin-related message at all (plugin may not have loaded):\n%s", service, logs) + } + for _, line := range strings.Split(logs, "\n") { + ll := strings.ToLower(line) + isPluginLine := strings.Contains(ll, "plugin") + isErrorLine := strings.Contains(ll, "level=error") || strings.Contains(ll, `"level":"error"`) + if isPluginLine && isErrorLine { + t.Fatalf("%s logs contain a plugin error line:\n%s", service, line) + } + } +} + +// TestPluginLoads covers integration test 1: the plugin loads inside a +// real Traefik binary with no plugin errors logged, and an authenticated +// GET /v1/models returns the models discovery pulled from the mock +// upstreams at plugin construction. +func TestPluginLoads(t *testing.T) { + waitForHealthy(t, traefik1URL+"/v1/models", aliceKey, 120*time.Second) + waitForHealthy(t, traefik2URL+"/v1/models", aliceKey, 120*time.Second) + + checkNoPluginErrors(t, "traefik1") + checkNoPluginErrors(t, "traefik2") + + resp, body := doJSON(t, http.MethodGet, traefik1URL+"/v1/models", aliceKey, nil) + if resp.StatusCode != http.StatusOK { + t.Fatalf("GET /v1/models: status = %d, want 200, body=%#v", resp.StatusCode, body) + } + data, ok := body["data"].([]any) + if !ok || len(data) == 0 { + t.Fatalf("GET /v1/models: expected a non-empty discovered model list, got %#v", body) + } +} + +// TestUnifiedChatOpenAIAndAnthropic covers integration test 2: a unified +// chat completion against the mock OpenAI upstream (passed through +// verbatim) and against the mock Anthropic upstream (translated into the +// same OpenAI-shaped envelope, with the client's requested provider-alias +// echoed back in "model"). +func TestUnifiedChatOpenAIAndAnthropic(t *testing.T) { + openaiReq := map[string]any{ + "model": "openai/gpt-mock", + "messages": []map[string]any{{"role": "user", "content": "hi"}}, + } + resp, body := doJSON(t, http.MethodPost, traefik1URL+"/v1/chat/completions", aliceKey, openaiReq) + if resp.StatusCode != http.StatusOK { + t.Fatalf("openai chat: status = %d, body=%#v", resp.StatusCode, body) + } + if content := firstChoiceContent(t, body); content != "mock openai response" { + t.Errorf("openai chat: content = %q, want %q", content, "mock openai response") + } + + anthropicReq := map[string]any{ + "model": "anthropic/claude-mock", + "messages": []map[string]any{{"role": "user", "content": "hi"}}, + } + resp2, body2 := doJSON(t, http.MethodPost, traefik1URL+"/v1/chat/completions", aliceKey, anthropicReq) + if resp2.StatusCode != http.StatusOK { + t.Fatalf("anthropic chat: status = %d, body=%#v", resp2.StatusCode, body2) + } + if content := firstChoiceContent(t, body2); content != "mock anthropic response" { + t.Errorf("anthropic chat: content = %q, want %q", content, "mock anthropic response") + } + if model, _ := body2["model"].(string); model != "anthropic/claude-mock" { + t.Errorf("anthropic chat: response model = %q, want the client-requested alias %q echoed back", model, "anthropic/claude-mock") + } +} + +// TestStreamingIncrementalityProbe covers integration test 3. Per the +// design spec's Appendix A spike lesson, streaming timing must never be +// judged through the Docker Desktop host port-proxy (it buffers) — so this +// runs the timing assertion as a one-shot container inside the compose +// network (`docker compose run --rm probe`), reading its exit code rather +// than making any timing claim from this (host-side) process itself. +func TestStreamingIncrementalityProbe(t *testing.T) { + out, err := exec.Command("docker", "compose", "-f", composeFile(), "run", "--rm", "probe").CombinedOutput() + t.Logf("probe output:\n%s", out) + if err != nil { + t.Fatalf("streaming incrementality probe failed: %v", err) + } +} + +// TestCrossReplicaLimits covers integration test 4: bob's limited group +// (requestsPerMinute: 3) is enforced globally across both Traefik +// "replicas" via the Redis they share, regardless of which one a given +// request lands on. +func TestCrossReplicaLimits(t *testing.T) { + reqBody := map[string]any{ + "model": "openai/gpt-mock", + "messages": []map[string]any{{"role": "user", "content": "hi"}}, + } + targets := []string{traefik1URL, traefik1URL, traefik2URL, traefik2URL} + + var statuses []int + var sawRetryAfter bool + for _, base := range targets { + resp, body := doJSON(t, http.MethodPost, base+"/v1/chat/completions", bobKey, reqBody) + statuses = append(statuses, resp.StatusCode) + switch resp.StatusCode { + case http.StatusOK: + case http.StatusTooManyRequests: + if resp.Header.Get("Retry-After") != "" { + sawRetryAfter = true + } + default: + t.Fatalf("unexpected status %d for bob's request against %s, body=%#v", resp.StatusCode, base, body) + } + } + + var ok200, too429 int + for _, s := range statuses { + if s == http.StatusOK { + ok200++ + } + if s == http.StatusTooManyRequests { + too429++ + } + } + if ok200 != 3 || too429 != 1 { + t.Fatalf("cross-replica limit: statuses = %v, want exactly three 200s and one 429 (order-independent)", statuses) + } + if !sawRetryAfter { + t.Fatalf("cross-replica limit: the 429 response carried no Retry-After header") + } +} + +// TestUsersFileHotReload covers integration test 5: appending a new user +// to the file-sourced users.json (bind-mounted read-only into both +// Traefik containers, edited here on the host) makes their key valid +// within the plugin's reload window, with no restart. +func TestUsersFileHotReload(t *testing.T) { + usersPath := filepath.Join("users", "users.json") + + original, err := os.ReadFile(usersPath) + if err != nil { + t.Fatalf("read users.json: %v", err) + } + t.Cleanup(func() { + if err := os.WriteFile(usersPath, original, 0o644); err != nil { + t.Errorf("restore users.json: %v", err) + } + }) + + // carol lands in "eng", not bob's "limited" group: TestCrossReplicaLimits + // runs earlier and deliberately exhausts "limited"'s shared per-minute + // bucket, which would make polling below flake if carol shared it. + updated := `{"users":[` + + `{"name":"bob","group":"limited","apiKey":"sk-int-bob"},` + + `{"name":"carol","group":"eng","apiKey":"sk-int-carol"}` + + `]}` + if err := os.WriteFile(usersPath, []byte(updated), 0o644); err != nil { + t.Fatalf("write users.json: %v", err) + } + + deadline := time.Now().Add(10 * time.Second) + var lastStatus int + for time.Now().Before(deadline) { + resp, _ := doJSON(t, http.MethodGet, traefik1URL+"/v1/models", carolKey, nil) + lastStatus = resp.StatusCode + if resp.StatusCode == http.StatusOK { + return + } + time.Sleep(time.Second) + } + t.Fatalf("carol's key did not become valid within 10s of users.json changing (last status=%d)", lastStatus) +} + +// TestPassthroughAndMCPStream covers integration test 6: native passthrough +// forwards a request with the provider's own configured key injected (the +// client's gateway key never reaches the upstream), and the MCP target +// proxy reverse-proxies an SSE stream through unmodified. +func TestPassthroughAndMCPStream(t *testing.T) { + reqBody := map[string]any{ + "model": "gpt-mock", + "messages": []map[string]any{{"role": "user", "content": "hi"}}, + } + resp, body := doJSON(t, http.MethodPost, traefik1URL+"/openai/v1/chat/completions", aliceKey, reqBody) + if resp.StatusCode != http.StatusOK { + t.Fatalf("passthrough: status = %d, body=%#v", resp.StatusCode, body) + } + if got, want := resp.Header.Get("X-Mock-Received-Auth"), "Bearer mock-openai-key"; got != want { + t.Errorf("passthrough: upstream received Authorization = %q, want the provider's own configured key %q, not the client's", got, want) + } + + req, err := http.NewRequest(http.MethodGet, traefik1URL+"/mcp/tool/sse", nil) + if err != nil { + t.Fatalf("new mcp request: %v", err) + } + req.Header.Set("Authorization", "Bearer "+aliceKey) + mcpResp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("mcp target proxy request: %v", err) + } + defer func() { _ = mcpResp.Body.Close() }() + if mcpResp.StatusCode != http.StatusOK { + t.Fatalf("mcp target proxy: status = %d", mcpResp.StatusCode) + } + if ct := mcpResp.Header.Get("Content-Type"); !strings.Contains(ct, "text/event-stream") { + t.Errorf("mcp target proxy: Content-Type = %q, want text/event-stream", ct) + } + raw, err := io.ReadAll(mcpResp.Body) + if err != nil { + t.Fatalf("read mcp target proxy body: %v", err) + } + if !strings.Contains(string(raw), "mocktool") { + t.Errorf("mcp target proxy: streamed body did not carry mocktool's content:\n%s", raw) + } +} + +// TestRealUpstreamSmoke covers integration test 7 (optional): a live round +// trip through a second middleware config carrying a keyless openai-type +// "uni" provider pointed at the operator's real gateway. Gated on +// INTEGRATION_REAL=1 and skipped, rather than failed, on any network or +// upstream problem — this is a smoke test of live infrastructure this +// suite does not own, not a correctness check on the plugin. +func TestRealUpstreamSmoke(t *testing.T) { + if os.Getenv("INTEGRATION_REAL") != "1" { + t.Skip("INTEGRATION_REAL not set to 1; skipping real-upstream smoke test") + } + + reqBody := map[string]any{ + "model": "uni/deepseek-v4-flash-0731", + "messages": []map[string]any{{"role": "user", "content": "Say hi in one word."}}, + } + b, err := json.Marshal(reqBody) + if err != nil { + t.Fatalf("marshal request: %v", err) + } + req, err := http.NewRequest(http.MethodPost, traefik1URL+"/v1/chat/completions", bytes.NewReader(b)) + if err != nil { + t.Fatalf("new request: %v", err) + } + req.Host = "real.localhost" // routes through llmgateway-real-router's Host() rule + req.Header.Set("Authorization", "Bearer sk-int-real") + req.Header.Set("Content-Type", "application/json") + + client := &http.Client{Timeout: 30 * time.Second} + resp, err := client.Do(req) + if err != nil { + t.Skipf("real upstream unreachable, skipping: %v", err) + } + defer func() { _ = resp.Body.Close() }() + + raw, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("read real upstream response: %v", err) + } + if resp.StatusCode != http.StatusOK { + t.Skipf("real upstream returned status %d, skipping (treated as upstream flake, not a plugin bug): %s", resp.StatusCode, raw) + } + + var body map[string]any + if err := json.Unmarshal(raw, &body); err != nil { + t.Fatalf("decode real upstream response: %v\nraw=%s", err, raw) + } + if content := firstChoiceContent(t, body); strings.TrimSpace(content) == "" { + t.Errorf("real upstream: response content is empty, body=%#v", body) + } +} diff --git a/integration/mock/Dockerfile b/integration/mock/Dockerfile new file mode 100644 index 0000000..0eec343 --- /dev/null +++ b/integration/mock/Dockerfile @@ -0,0 +1,15 @@ +# Build context is integration/ (see docker-compose.yml's build.context), +# so this Dockerfile can reach both go.mod and the mock/ package. +FROM golang:1.22-alpine AS build +WORKDIR /src +ENV CGO_ENABLED=0 +COPY go.mod ./ +COPY mock ./mock +RUN go build -o /out/mock ./mock + +FROM alpine:3.20 +RUN apk add --no-cache wget ca-certificates +COPY --from=build /out/mock /usr/local/bin/mock +EXPOSE 8080 +ENTRYPOINT ["/usr/local/bin/mock"] +CMD ["-mode", "openai"] diff --git a/integration/mock/main.go b/integration/mock/main.go new file mode 100644 index 0000000..180b64e --- /dev/null +++ b/integration/mock/main.go @@ -0,0 +1,378 @@ +// Command mock is the integration suite's fake upstream. It runs in two +// distinct modes, selected by flags: +// +// - Server mode (-mode openai|anthropic|gemini|tool): serves the minimal +// subset of one upstream's wire API the corresponding plugin adapter +// calls, using canned fixtures shaped like the plugin's own unit-test +// fakes (see provider_openai_test.go, provider_anthropic_test.go, +// provider_gemini_test.go). "tool" mode serves a bare SSE endpoint, +// standing in for an MCP target the gateway's target proxy forwards to +// verbatim. +// - Probe mode (-probe): a raw-TCP client that dials a running gateway +// directly (inside the compose network, never through the Docker +// Desktop host port-proxy — that layer buffers and hides true +// inter-chunk timing), issues one streaming chat completion by hand, +// and reports via its exit code whether the response arrived as +// genuinely incremental SSE chunks rather than one buffered write. +package main + +import ( + "bytes" + "encoding/json" + "flag" + "fmt" + "log" + "net" + "net/http" + "os" + "strconv" + "strings" + "time" +) + +// streamChunkGap is the pause between successive SSE writes in every mode's +// streaming handler. The probe (see runProbe) requires at least 3 gaps of +// 100ms or more between arrivals; 150ms per gap leaves comfortable margin +// against scheduler jitter while keeping the suite fast. +const streamChunkGap = 150 * time.Millisecond + +func main() { + mode := flag.String("mode", "", "server mode: openai, anthropic, gemini, or tool") + addr := flag.String("addr", ":8080", "listen address for server mode") + probe := flag.Bool("probe", false, "run as a raw-TCP streaming probe instead of a server") + target := flag.String("target", "traefik1:80", "probe mode: host:port of the gateway to dial") + apiKey := flag.String("apikey", "sk-int-alice", "probe mode: gateway API key to present") + model := flag.String("model", "openai/gpt-mock", "probe mode: model id to request") + flag.Parse() + + if *probe { + os.Exit(runProbe(*target, *apiKey, *model)) + } + + mux, err := newServerMux(*mode) + if err != nil { + log.Fatalf("mock: %v", err) + } + log.Printf("mock: mode=%s listening on %s", *mode, *addr) + if err := http.ListenAndServe(*addr, mux); err != nil { //nolint:gosec // integration-only fixture server, no need for timeouts + log.Fatalf("mock: listen: %v", err) + } +} + +// newServerMux returns the handler set for one server mode. +func newServerMux(mode string) (*http.ServeMux, error) { + mux := http.NewServeMux() + mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("ok")) + }) + + switch mode { + case "openai": + mux.HandleFunc("/v1/models", handleOpenAIModels) + mux.HandleFunc("/v1/chat/completions", handleOpenAIChat) + case "anthropic": + mux.HandleFunc("/v1/models", handleAnthropicModels) + mux.HandleFunc("/v1/messages", handleAnthropicMessages) + case "gemini": + mux.HandleFunc("/v1beta/models", handleGeminiModels) + mux.HandleFunc("/v1beta/models/", handleGeminiGenerate) + case "tool": + mux.HandleFunc("/sse", handleToolSSE) + default: + return nil, fmt.Errorf("unknown -mode %q (want openai, anthropic, gemini, or tool)", mode) + } + return mux, nil +} + +// decodeJSONBody reads and parses r's body as a generic JSON object. It +// returns an empty, non-nil map on any read/parse error — every handler +// below only reads a couple of optional fields out of it, so a malformed +// body degrades to default fixture values rather than a 400 the tests +// don't care about. +func decodeJSONBody(r *http.Request) map[string]any { + var body map[string]any + defer func() { _ = r.Body.Close() }() + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + return map[string]any{} + } + return body +} + +func writeJSON(w http.ResponseWriter, v any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(v) +} + +// sseFlusher wraps the response writer for one SSE handler, setting the +// standard streaming headers up front and flushing after every write so a +// downstream proxy (the gateway, Traefik) observes each event as it is +// produced rather than buffered until the handler returns. +type sseFlusher struct { + w http.ResponseWriter + f http.Flusher +} + +func newSSEFlusher(w http.ResponseWriter) sseFlusher { + h := w.Header() + h.Set("Content-Type", "text/event-stream") + h.Set("Cache-Control", "no-cache") + h.Set("X-Accel-Buffering", "no") + w.WriteHeader(http.StatusOK) + f, _ := w.(http.Flusher) + return sseFlusher{w: w, f: f} +} + +func (s sseFlusher) write(event, data string) { + if event != "" { + _, _ = fmt.Fprintf(s.w, "event: %s\n", event) + } + _, _ = fmt.Fprintf(s.w, "data: %s\n\n", data) + if s.f != nil { + s.f.Flush() + } +} + +// --- openai mode ----------------------------------------------------------- + +func handleOpenAIModels(w http.ResponseWriter, _ *http.Request) { + writeJSON(w, map[string]any{ + "object": "list", + "data": []map[string]any{ + {"id": "gpt-mock", "object": "model", "owned_by": "mock-openai"}, + }, + }) +} + +// openaiStreamDeltas are the content fragments the streaming handler emits +// as separate SSE frames, one streamChunkGap apart, mirroring the shape of +// provider_openai_test.go's streamFrames fixture. +var openaiStreamDeltas = []string{"Hel", "lo", " wor", "ld", "!"} + +func handleOpenAIChat(w http.ResponseWriter, r *http.Request) { + w.Header().Set("X-Mock-Received-Auth", r.Header.Get("Authorization")) + body := decodeJSONBody(r) + model, _ := body["model"].(string) + if model == "" { + model = "gpt-mock" + } + stream, _ := body["stream"].(bool) + + if !stream { + writeJSON(w, map[string]any{ + "id": "chatcmpl-mock", + "object": "chat.completion", + "model": model, + "choices": []map[string]any{ + { + "index": 0, + "message": map[string]any{"role": "assistant", "content": "mock openai response"}, + "finish_reason": "stop", + }, + }, + "usage": map[string]any{"prompt_tokens": 10, "completion_tokens": 5}, + }) + return + } + + sw := newSSEFlusher(w) + for i, delta := range openaiStreamDeltas { + deltaObj := map[string]any{"content": delta} + if i == 0 { + deltaObj["role"] = "assistant" + } + frame := map[string]any{ + "id": "chatcmpl-mock", + "choices": []map[string]any{{"index": 0, "delta": deltaObj}}, + } + b, _ := json.Marshal(frame) + sw.write("", string(b)) + time.Sleep(streamChunkGap) + } + usage, _ := json.Marshal(map[string]any{ + "id": "chatcmpl-mock", + "choices": []map[string]any{}, + "usage": map[string]any{"prompt_tokens": 10, "completion_tokens": 5}, + }) + sw.write("", string(usage)) + time.Sleep(streamChunkGap) + sw.write("", "[DONE]") +} + +// --- anthropic mode ---------------------------------------------------------- + +func handleAnthropicModels(w http.ResponseWriter, _ *http.Request) { + writeJSON(w, map[string]any{ + "data": []map[string]any{{"id": "claude-mock"}}, + }) +} + +func handleAnthropicMessages(w http.ResponseWriter, r *http.Request) { + w.Header().Set("X-Mock-Received-Auth", r.Header.Get("x-api-key")) + body := decodeJSONBody(r) + model, _ := body["model"].(string) + if model == "" { + model = "claude-mock" + } + stream, _ := body["stream"].(bool) + + if !stream { + writeJSON(w, map[string]any{ + "id": "msg_mock", + "type": "message", + "role": "assistant", + "content": []map[string]any{ + {"type": "text", "text": "mock anthropic response"}, + }, + "model": model, + "stop_reason": "end_turn", + "usage": map[string]any{"input_tokens": 10, "output_tokens": 4}, + }) + return + } + + sw := newSSEFlusher(w) + frames := []struct{ event, data string }{ + {"message_start", `{"type":"message_start","message":{"id":"msg_mock","usage":{"input_tokens":10,"output_tokens":1}}}`}, + {"content_block_start", `{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}`}, + {"content_block_delta", `{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"mock "}}`}, + {"content_block_delta", `{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"anthropic "}}`}, + {"content_block_delta", `{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"response"}}`}, + {"content_block_stop", `{"type":"content_block_stop","index":0}`}, + {"message_delta", `{"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":4}}`}, + {"message_stop", `{"type":"message_stop"}`}, + } + for _, f := range frames { + sw.write(f.event, f.data) + time.Sleep(20 * time.Millisecond) + } +} + +// --- gemini mode ------------------------------------------------------------- + +func handleGeminiModels(w http.ResponseWriter, _ *http.Request) { + writeJSON(w, map[string]any{ + "models": []map[string]any{{"name": "models/gemini-mock"}}, + }) +} + +func handleGeminiGenerate(w http.ResponseWriter, r *http.Request) { + w.Header().Set("X-Mock-Received-Auth", r.Header.Get("x-goog-api-key")) + rest := strings.TrimPrefix(r.URL.Path, "/v1beta/models/") + _, action, _ := strings.Cut(rest, ":") + + switch action { + case "generateContent": + writeJSON(w, map[string]any{ + "candidates": []map[string]any{ + { + "content": map[string]any{"role": "model", "parts": []map[string]any{{"text": "mock gemini response"}}}, + "finishReason": "STOP", + }, + }, + "usageMetadata": map[string]any{"promptTokenCount": 10, "candidatesTokenCount": 4}, + "responseId": "resp_mock", + }) + case "streamGenerateContent": + sw := newSSEFlusher(w) + frames := []string{ + `{"candidates":[{"content":{"role":"model","parts":[{"text":"mock "}]}}],"responseId":"resp_mock"}`, + `{"candidates":[{"content":{"role":"model","parts":[{"text":"gemini "}]}}],"responseId":"resp_mock"}`, + `{"candidates":[{"content":{"role":"model","parts":[{"text":"response"}]},"finishReason":"STOP"}],"usageMetadata":{"promptTokenCount":10,"candidatesTokenCount":4}}`, + } + for _, f := range frames { + sw.write("", f) // Gemini's SSE stream carries no "event:" field. + time.Sleep(20 * time.Millisecond) + } + default: + http.NotFound(w, r) + } +} + +// --- tool mode (MCP target proxy test) --------------------------------------- + +func handleToolSSE(w http.ResponseWriter, _ *http.Request) { + sw := newSSEFlusher(w) + for i, msg := range []string{"hello", "from", "mocktool"} { + b, _ := json.Marshal(map[string]any{"seq": i, "msg": msg}) + sw.write("", string(b)) + time.Sleep(20 * time.Millisecond) + } +} + +// --- probe mode ---------------------------------------------------------- + +// runProbe dials target over a plain net.Conn (never through an http.Client, +// and never through the Docker Desktop host port-proxy — the caller must +// pass an in-compose-network address such as "traefik1:80"), hand-writes one +// streaming chat completion request, and times each raw read that carries +// SSE "data:" bytes. It returns 0 iff at least 3 distinct inter-arrival gaps +// of 100ms or more were observed, proving the response arrived as +// incremental chunks rather than one buffered write; 1 otherwise. +func runProbe(target, apiKey, model string) int { + conn, err := net.DialTimeout("tcp", target, 10*time.Second) + if err != nil { + fmt.Fprintf(os.Stderr, "probe: dial %s: %v\n", target, err) + return 1 + } + defer func() { _ = conn.Close() }() + + body := fmt.Sprintf(`{"model":%q,"stream":true,"messages":[{"role":"user","content":"hi"}]}`, model) + host, _, err := net.SplitHostPort(target) + if err != nil || host == "" { + host = target + } + req := "POST /v1/chat/completions HTTP/1.1\r\n" + + "Host: " + host + "\r\n" + + "Authorization: Bearer " + apiKey + "\r\n" + + "Content-Type: application/json\r\n" + + "Content-Length: " + strconv.Itoa(len(body)) + "\r\n" + + "Connection: close\r\n" + + "\r\n" + body + + if err := conn.SetDeadline(time.Now().Add(30 * time.Second)); err != nil { + fmt.Fprintf(os.Stderr, "probe: set deadline: %v\n", err) + return 1 + } + if _, err := conn.Write([]byte(req)); err != nil { + fmt.Fprintf(os.Stderr, "probe: write request: %v\n", err) + return 1 + } + + var arrivals []time.Time + buf := make([]byte, 4096) + headerDone := false + for { + n, readErr := conn.Read(buf) + if n > 0 { + chunk := buf[:n] + if !headerDone { + if idx := bytes.Index(chunk, []byte("\r\n\r\n")); idx >= 0 { + headerDone = true + chunk = chunk[idx+4:] + } else { + chunk = nil + } + } + if len(chunk) > 0 && bytes.Contains(chunk, []byte("data:")) { + arrivals = append(arrivals, time.Now()) + } + } + if readErr != nil { + break + } + } + + gaps := 0 + for i := 1; i < len(arrivals); i++ { + if d := arrivals[i].Sub(arrivals[i-1]); d >= 100*time.Millisecond { + gaps++ + } + } + fmt.Printf("probe: target=%s model=%s arrivals=%d gaps>=100ms=%d\n", target, model, len(arrivals), gaps) + if gaps >= 3 { + return 0 + } + return 1 +} diff --git a/integration/traefik/dynamic.yml b/integration/traefik/dynamic.yml new file mode 100644 index 0000000..a81f2ab --- /dev/null +++ b/integration/traefik/dynamic.yml @@ -0,0 +1,91 @@ +http: + routers: + llmgateway-router: + rule: "PathPrefix(`/`)" + entryPoints: ["web"] + service: dummy-service + middlewares: ["llmgateway"] + priority: 1 + + llmgateway-real-router: + rule: "Host(`real.localhost`)" + entryPoints: ["web"] + service: dummy-service + middlewares: ["llmgateway-real"] + priority: 100 + + services: + # Placeholder upstream. Every route the plugin recognizes is + # short-circuited by the middleware itself; this backend is never + # actually reached by the requests the integration suite sends. + dummy-service: + loadBalancer: + servers: + - url: "http://mockopenai:8080" + + middlewares: + llmgateway: + plugin: + llmgateway: + providers: + openai: + type: openai + baseUrl: "http://mockopenai:8080" + apiKey: "mock-openai-key" + discovery: true + models: ["gpt-mock"] + anthropic: + type: anthropic + baseUrl: "http://mockanthropic:8080" + apiKey: "mock-anthropic-key" + models: ["claude-mock"] + gemini: + type: gemini + baseUrl: "http://mockgemini:8080" + apiKey: "mock-gemini-key" + models: ["gemini-mock"] + groups: + eng: + limits: + requestsPerMinute: 1000 + limited: + limits: + requestsPerMinute: 3 + users: + inline: + - name: alice + group: eng + apiKey: "sk-int-alice" + file: "/users/users.json" + redis: + address: "redis:6379" + mcpServers: + tool: + url: "http://mocktool:8080" + + # Second, separate middleware instance carrying only the optional + # real-upstream smoke test's "uni" provider (task 7, INTEGRATION_REAL=1 + # gated). Kept isolated from the mock-backed "llmgateway" middleware + # above so the always-on tests never depend on network egress. + llmgateway-real: + plugin: + llmgateway: + providers: + uni: + type: openai + baseUrl: "https://llmgw.example.com" + apiKey: "" + # Explicit rather than discovered: resolution must not depend + # on this container reaching the real upstream at Traefik + # startup (offline dev machines, CI with no egress). The + # actual chat completion call in task 7 still goes out live. + models: ["deepseek-v4-flash-0731"] + groups: + eng: + limits: + requestsPerMinute: 1000 + users: + inline: + - name: alice-real + group: eng + apiKey: "sk-int-real" diff --git a/integration/traefik/traefik.yml b/integration/traefik/traefik.yml new file mode 100644 index 0000000..483f6c2 --- /dev/null +++ b/integration/traefik/traefik.yml @@ -0,0 +1,18 @@ +entryPoints: + web: + address: ":80" + +experimental: + localPlugins: + llmgateway: + moduleName: github.com/lukaszraczylo/traefik-llmgateway + +providers: + file: + filename: /etc/traefik/dynamic.yml + watch: true + +log: + level: INFO + +accessLog: {} diff --git a/integration/users/users.json b/integration/users/users.json new file mode 100644 index 0000000..4134766 --- /dev/null +++ b/integration/users/users.json @@ -0,0 +1,5 @@ +{ + "users": [ + {"name": "bob", "group": "limited", "apiKey": "sk-int-bob"} + ] +} diff --git a/registry.go b/registry.go index bfb4d6b..1e72dee 100644 --- a/registry.go +++ b/registry.go @@ -368,7 +368,18 @@ func (m *modelRegistry) resolveAgainst(providerName, upstreamModel, requestedID if !allowed || !grp.allowsProvider(providerName) { return nil, "", "", errModelDenied } - return m.adapters[providerName], upstreamModel, providerName + "/" + upstreamModel, nil + // canonical is assigned to a local before the return, not inlined into + // it: yaegi v0.16.1 panics ("reflect.Set: value of type + // interp.valueInterface is not assignable to type string") building a + // multi-value return tuple when one element is an inline string + // concatenation of two parameters. The identical expression assigned + // to a variable first, then returned, does not trigger it. Verified + // empirically against yaegi v0.16.1 running this exact function under + // real Traefik (Task 15's integration suite) — tools/yaegi-check never + // exercises this call path (it only drives GET /v1/models), which is + // why the existing yaegi gate never caught it. + canonical := providerName + "/" + upstreamModel + return m.adapters[providerName], upstreamModel, canonical, nil } // warnCollisionOnce logs, at most once per colliding bare id for this diff --git a/resp.go b/resp.go index 14a17fd..169d492 100644 --- a/resp.go +++ b/resp.go @@ -51,10 +51,19 @@ const respMaxLineLen = 64 << 10 // respErr is a RESP2 error reply ("-message\r\n"). It implements error so a // caller inspecting a decoded reply can type-assert for it the same way as // any other error. -type respErr string +// +// A struct wrapping the message, not a defined string type ("type respErr +// string"): yaegi v0.16.1 conflates a defined string type with plain +// string in a type assertion — v.(respErr) on an any holding a genuine +// plain string (e.g. decodeReply's "+OK" success reply, itself a string) +// incorrectly reports ok=true, so every successful handshakeLocked call +// (AUTH/SELECT) was misread as a failed one under real Traefik. A struct +// has no such ambiguity: verified empirically under real Traefik (Task +// 15's integration suite, cross-replica rate limiting via Redis). +type respErr struct{ msg string } // Error implements the error interface. -func (e respErr) Error() string { return string(e) } +func (e respErr) Error() string { return e.msg } // respClient is a minimal stdlib-only RESP2 client for a single Redis- // compatible server: one mutex-guarded TCP connection, lazily dialled on @@ -305,7 +314,7 @@ func decodeReply(r *bufio.Reader, depth int) (any, error) { case '+': return rest, nil case '-': - return respErr(rest), nil + return respErr{msg: rest}, nil case ':': n, err := strconv.ParseInt(rest, 10, 64) if err != nil { diff --git a/routes_unified.go b/routes_unified.go index 852fcd4..250c145 100644 --- a/routes_unified.go +++ b/routes_unified.go @@ -225,14 +225,21 @@ func (g *Gateway) handleAdapterError(sw *statusTrackingWriter, err error, provid return } - var perr *providerHTTPError - if errors.As(err, &perr) { + // Plain type assertions, not errors.As: providerHTTPError and + // translateError are both always returned bare from every adapter and + // translate_*.go call site — never wrapped via fmt.Errorf("%w", ...) — + // so errors.As's unwrap-chain walk buys nothing here, and yaegi + // v0.16.1 panics ("errors: *target must be interface or implement + // error") calling errors.As with an interpreted pointer type as its + // target, even though that type's Error() method is right there. + // Verified empirically under real Traefik (Task 15's integration + // suite); tools/yaegi-check never exercises this call path. + if perr, ok := err.(*providerHTTPError); ok { writeProviderUpstreamError(sw, providerName, perr) return } - var terr *translateError - if errors.As(err, &terr) { + if terr, ok := err.(*translateError); ok { status := http.StatusBadRequest if terr.notSupported { status = http.StatusNotImplemented diff --git a/sse.go b/sse.go index 594caf8..c826d89 100644 --- a/sse.go +++ b/sse.go @@ -78,7 +78,19 @@ func readSSE(r io.Reader, fn func(sseEvent) error) error { joined := strings.Join(dataLines, "\n") fire := hasData && joined != "" ev := sseEvent{event: event, data: []byte(joined)} - event, dataLines, hasData = "", nil, false + // Three separate assignments, not one "event, dataLines, hasData = + // \"\", nil, false" multi-value statement: yaegi v0.16.1 panics + // ("reflect: New(nil)") interpreting a 3-way assignment mixing a + // string, a nil slice, and a bool literal in one statement — the + // same class of bug as resolveAgainst's local-var fix (registry.go), + // just in assignment form instead of a return tuple. The identical + // values assigned one statement at a time do not trigger it. + // Verified empirically under real Traefik (Task 15's integration + // suite, the SSE streaming path); tools/yaegi-check never exercises + // a live streaming response. + event = "" + dataLines = nil + hasData = false if !fire { return nil } @@ -134,6 +146,25 @@ type sseWriter struct { // need not implement http.Flusher — the Gateway's statusTrackingWriter // always does, since it delegates — but writeData/writeDone tolerate its // absence by skipping the flush. +// +// KNOWN LIMITATION under real Traefik (yaegi v0.16.1), confirmed by +// Task 15's integration suite: the w.(http.Flusher) assertion below +// always reports false, because Yaegi wraps any http.ResponseWriter +// argument crossing from Traefik's compiled dispatcher into this +// interpreted package in a synthetic type scoped to exactly the +// http.ResponseWriter method set — Header/Write/WriteHeader — dropping +// any other interface the real writer satisfies. The same holds one +// layer down: statusTrackingWriter.Flush's own w.rw.(http.Flusher) +// check, and even http.NewResponseController(w.rw).Flush(), fail the +// same way, all verified empirically. Net effect: streamed responses +// (chat completions, MCP/A2A proxying via flushWriter below) are +// correct in content but arrive as one batch when the handler returns, +// not incrementally, when this plugin is loaded via localPlugins — +// unlike compiled Go, where this exact code streams correctly (see the +// unit tests). Confirmed as an external, still-open Yaegi/Traefik bug, +// not a defect in this code: traefik/yaegi#1600 and, with the +// "confirmed bug" label from Traefik's own maintainers, +// traefik/traefik#10269. No code-level workaround is known. func newSSEWriter(w http.ResponseWriter) *sseWriter { h := w.Header() h.Set("Content-Type", "text/event-stream")