mirror of
https://github.com/lukaszraczylo/traefik-llmgateway.git
synced 2026-10-19 05:47:48 +00:00
feat(config): advertise a concurrency limit to clients
Self-hosted upstreams serve few requests at once. A client that opens many requests overloads them. The new maxConcurrency option sends X-Concurrency-Limit on every response. Clients that honour the header cap their own in-flight requests. The default of 0 sends no header, so behaviour does not change.
This commit is contained in:
@@ -383,6 +383,7 @@ config accepts them as YAML, which decodes to the same JSON shape.
|
||||
| `requireModelPrefix` | `bool` | `false` | When `true`, a model resolves only as `provider/model` or as a `modelAliases` id. A bare id returns 404 `unknown model` and `GET /v1/models` lists no bare ids. `false` keeps the sorted-first bare-id collision winner. |
|
||||
| `stripClientAnthropicBeta` | `bool` | `false` | When `true`, `POST /v1/messages` does not forward the client's `anthropic-beta` header to the provider, as native passthrough already does. `false` forwards it, so a client can enable beta features on the operator's own Anthropic account. |
|
||||
| `modelMeta` | `map[string]*ModelMetaConfig` | `{}` | Per-model context-window and cost overrides, keyed by exact `provider/model` or a bare model id — see [Model metadata](#model-metadata). Omitted or empty means no overrides; every model's metadata still resolves through discovery and the built-in table. |
|
||||
| `maxConcurrency` | `int` | `0` | When above `0`, every response carries `X-Concurrency-Limit: <n>`, so clients that honour it (omps) keep at most `n` requests in flight. Set it to what your upstreams serve at once. The gateway only advertises the value; it does not queue or reject on it. `0` sends no header. Negative values are rejected at startup. |
|
||||
| `passthroughUnknown` | `bool` | `false` | `false`: a request matching none of the plugin's routes gets a 404 JSON envelope. `true`: it falls through to the router's own backing service. **Does not cover the three media routes** — see the upgrade note below. |
|
||||
| `maxRequestBytes` | `int` | `0` (4 MiB) | Raises the unified chat/embeddings request-body cap. `0` or negative keeps 4 MiB; a positive value is clamped to 64 MiB. Pair with `maxInFlightBodyRequests` and the pod memory limit: an adversarial body decodes to ~12x live heap. |
|
||||
| `maxInFlightBodyRequests` | `int` | `0` (self-tuned) | Caps how many unified/media requests may concurrently be inside their own request-body read+decode step — see [Body-admission limit](#body-admission-limit). `0` self-tunes from the host's own `GOMAXPROCS`; a positive value pins an exact cap instead, clamped to a maximum of 10,000. |
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
package traefikllmgateway
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func newConcurrencyTestGateway(t *testing.T, maxConcurrency int) (http.Handler, error) {
|
||||
t.Helper()
|
||||
cfg := CreateConfig()
|
||||
cfg.MaxConcurrency = maxConcurrency
|
||||
cfg.Providers = map[string]*ProviderConfig{
|
||||
"openai": {Type: "openai", APIKey: "sk-test", Models: []string{"gpt-test"}},
|
||||
}
|
||||
cfg.Groups = map[string]*GroupConfig{"default": {}}
|
||||
cfg.Users = &UsersConfig{Inline: []*UserConfig{{Name: "tester", Group: "default", APIKey: "sk-user"}}}
|
||||
return New(context.Background(), http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}), cfg, "llmgw")
|
||||
}
|
||||
|
||||
func TestMaxConcurrency_AdvertisedOnResponses(t *testing.T) {
|
||||
t.Parallel()
|
||||
tests := []struct {
|
||||
name string
|
||||
header string
|
||||
max int
|
||||
}{
|
||||
{"configured limit is advertised", "3", 3},
|
||||
{"zero sends no header", "", 0},
|
||||
}
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
h, err := newConcurrencyTestGateway(t, tc.max)
|
||||
if err != nil {
|
||||
t.Fatalf("New: %v", err)
|
||||
}
|
||||
req := httptest.NewRequest(http.MethodGet, "/v1/models", nil)
|
||||
req.Header.Set("Authorization", "Bearer sk-user")
|
||||
rec := httptest.NewRecorder()
|
||||
h.ServeHTTP(rec, req)
|
||||
if got := rec.Header().Get("X-Concurrency-Limit"); got != tc.header {
|
||||
t.Fatalf("X-Concurrency-Limit = %q, want %q", got, tc.header)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestMaxConcurrency_NegativeRejected(t *testing.T) {
|
||||
t.Parallel()
|
||||
if _, err := newConcurrencyTestGateway(t, -1); err == nil {
|
||||
t.Fatal("New accepted a negative maxConcurrency")
|
||||
}
|
||||
}
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
"net/url"
|
||||
"os"
|
||||
"runtime"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -183,6 +184,13 @@ type Config struct {
|
||||
// resolved (404 unknown model) and is not listed by GET /v1/models. It
|
||||
// removes the sorted-first bare-id collision winner. Defaults to false.
|
||||
RequireModelPrefix bool `json:"requireModelPrefix,omitempty"`
|
||||
// MaxConcurrency, when > 0, is advertised on every response as
|
||||
// X-Concurrency-Limit so clients that honour it (omps) cap their own
|
||||
// in-flight requests to this gateway. It documents what the upstreams
|
||||
// can serve at once; the gateway does not queue or reject on it. 0
|
||||
// (the default) sends no header. Negative values are rejected at
|
||||
// construction.
|
||||
MaxConcurrency int `json:"maxConcurrency,omitempty"`
|
||||
}
|
||||
|
||||
// ProviderConfig describes one upstream LLM provider.
|
||||
@@ -927,6 +935,9 @@ func newGateway(ctx context.Context, next http.Handler, config *Config, name str
|
||||
if err := validateTargetURLs(config); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if config.MaxConcurrency < 0 {
|
||||
return nil, errors.New("llmgateway: maxConcurrency must not be negative")
|
||||
}
|
||||
// Finding F4, 2026-09 review: validate every Config.Pricing override
|
||||
// at construction — non-nil, finite, >= 0 — rather than letting a
|
||||
// negative/NaN/Inf entry reach costMicrosKnownFor (pricing.go), where it
|
||||
@@ -1347,9 +1358,15 @@ func attachUsersFile(auth *authStore, path string, log gatewayLogger) error {
|
||||
}
|
||||
|
||||
// ServeHTTP is the internal router. It grows in later tasks.
|
||||
// concurrencyLimitHeader advertises Config.MaxConcurrency to clients.
|
||||
const concurrencyLimitHeader = "X-Concurrency-Limit"
|
||||
|
||||
func (g *Gateway) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
sw := &statusTrackingWriter{ResponseWriter: w}
|
||||
defer recoverPanic(sw, g)
|
||||
if g.cfg.MaxConcurrency > 0 {
|
||||
sw.Header().Set(concurrencyLimitHeader, strconv.Itoa(g.cfg.MaxConcurrency))
|
||||
}
|
||||
// pruneRejectionsAfterReload only runs when maybeReload just performed
|
||||
// a REAL reload (users_file.go's own doc comment explains why it must
|
||||
// never run on every request) — a deleted file-sourced user must not
|
||||
|
||||
Reference in New Issue
Block a user