Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions docs/concepts/architecture.md
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,14 @@ the header is forwarded on both. Streaming paths that duplicate request
construction are where credential omission bugs live; a single site plus a
test removes the class.

### Rate-limit state is observed and bucketed, never enforced

When a provider has `rate_limit_headers` configured, the proxy reads
the upstream's rate-limit headers at the same single capture site as
credentials. The values are accumulated per minute in memory. On minute
rollover the finished bucket is persisted. The proxy never throttles or
rejects a request based on these values. The data is telemetry only.

### Usage is merged max-wins across stream chunks

The proxy forces `stream_options.include_usage` on streamed requests (there
Expand Down
58 changes: 58 additions & 0 deletions docs/guides/providers.md
Original file line number Diff line number Diff line change
Expand Up @@ -152,6 +152,64 @@ copied identifiers rather than foreign keys into the catalog.

Disable instead of deleting if you might want the provider back.

## Track upstream rate limits

Some providers return rate-limit headers on every response.
The proxy captures these values and exports them.

### 1. Configure the header mapping

Set `rate_limit_headers` on the provider.
The standard OpenAI-style headers work for most providers:

```bash
curl -s -X PATCH $P/admin/v1/providers/tensorx -H "authorization: Bearer $ADMIN" \
-H 'content-type: application/json' -d '{
"rate_limit_headers": {
"limit_requests": "x-ratelimit-limit-requests",
"remaining_requests": "x-ratelimit-remaining-requests",
"limit_tokens": "x-ratelimit-limit-tokens",
"remaining_tokens": "x-ratelimit-remaining-tokens"
}
}'
```

Accepted metric keys:

- `limit_requests`
- `remaining_requests`
- `limit_tokens`
- `remaining_tokens`
- `reset_requests`
- `reset_tokens`

The UI has a "Use standard headers" button that fills these four.
An empty object disables tracking.

### 2. Read the current state

After the next proxied request, the reading appears in three places:

- Response headers on every proxied response: `x-llmproxy-ratelimit-requests-remaining`,
`x-llmproxy-ratelimit-requests-limit`, `x-llmproxy-ratelimit-tokens-remaining`,
`x-llmproxy-ratelimit-tokens-limit`.
- Prometheus gauges: `llmproxy_provider_rate_limit_remaining` and
`llmproxy_provider_rate_limit_limit`, labelled by `provider` and `unit`.
- The Providers sub-tab under Usage in the UI.

### 3. Query the API

```bash
# Current snapshot
curl -s $P/stats/rate-limits -H "authorization: Bearer $KEY"

# Stored minute-bucket history for one provider
curl -s "$P/stats/rate-limits/series?provider=tensorx" -H "authorization: Bearer $KEY"
```

Both endpoints require any authenticated user.
The snapshot includes a `stale` flag when the reading is older than five minutes.

## Where next

- [models.md](models.md) to bind the provider's models.
Expand Down
8 changes: 7 additions & 1 deletion docs/reference/api.md
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,11 @@ resolution. Calling an endpoint outside a model's capability set fails at the
proxy with a 400 naming the supported capabilities, never with a confusing
upstream 404.

Every response carries `x-llmproxy-provider` and `x-llmproxy-model`. Errors are
Every response carries `x-llmproxy-provider` and `x-llmproxy-model`. When the
provider has rate-limit tracking configured, responses also carry
`x-llmproxy-ratelimit-requests-remaining`, `x-llmproxy-ratelimit-requests-limit`,
`x-llmproxy-ratelimit-tokens-remaining` and `x-llmproxy-ratelimit-tokens-limit`
(only the observed values are set). Errors are
OpenAI-shaped with an added `llmproxy.source` field plus
`x-llmproxy-error-source: proxy|upstream`; upstream error bodies and status
codes pass through intact.
Expand Down Expand Up @@ -88,6 +92,8 @@ together; a pair nothing carries simply matches nothing. `outcome` takes
| `GET /stats/requests?limit&offset` | One page of the filtered request metadata log (never content), newest first; returns `{requests, limit, offset, total}` |
| `GET /stats/requests/facets?since&until` | Distinct principals, keys, providers, models, clients and tags in the window, for the explorer's filter options |
| `GET /stats/errors?bucket&since&until` | The errors dashboard in one call: a gap-filled series of counts per outcome, plus a breakdown per (provider, model, endpoint, client, tags, outcome, error_kind, status_code) with request count, average duration, last-seen and time-to-outcome bands (<1s, 1-5s, 5-15s, 15-30s, 30-60s, 60-120s, >=120s). Rows with outcome `ok` are included so error rates have their denominator. |
| `GET /stats/rate-limits` | Current rate-limit snapshot per configured provider: limits, remaining, worst remaining in the current minute, reset times, `observed_at` and a `stale` flag when the reading is older than five minutes. |
| `GET /stats/rate-limits/series?provider&since&until` | Stored minute-bucket rate-limit samples for one provider. Each row carries the last and worst remaining values for that minute. |

Failures carry an `error_kind` classification token: the transport class on
`unreachable` (`timeout`, `connection_error`, ...) and the upstream's
Expand Down
59 changes: 32 additions & 27 deletions internal/catalog/catalog.go
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,9 @@ type Route struct {
UpstreamName string
Capabilities map[string]bool
URLOverrides map[string]string
// RateLimitHeaders maps metric keys to upstream header names.
// Empty when the provider has no rate-limit tracking configured.
RateLimitHeaders map[string]string
}

func (r *Route) EndpointURL(endpoint string) string {
Expand Down Expand Up @@ -192,21 +195,22 @@ func (c *Catalog) load(ctx context.Context, alias string) (*Route, error) {
}
}
return &Route{
ProviderID: provider.ID,
ProviderName: provider.Name,
WireFormat: provider.WireFormat,
BaseURL: provider.BaseURL,
Credential: credential,
VerifyTLS: provider.VerifyTLS,
CAPEM: provider.CAPEM.String,
TimeoutConnect: time.Duration(provider.TimeoutConnect * float64(time.Second)),
TimeoutRead: time.Duration(provider.TimeoutRead * float64(time.Second)),
MaxConcurrency: int(provider.MaxConcurrency.Int64),
Alias: binding.Alias,
TargetAlias: binding.TargetAlias,
UpstreamName: binding.UpstreamName,
Capabilities: caps,
URLOverrides: overrides,
ProviderID: provider.ID,
ProviderName: provider.Name,
WireFormat: provider.WireFormat,
BaseURL: provider.BaseURL,
Credential: credential,
VerifyTLS: provider.VerifyTLS,
CAPEM: provider.CAPEM.String,
TimeoutConnect: time.Duration(provider.TimeoutConnect * float64(time.Second)),
TimeoutRead: time.Duration(provider.TimeoutRead * float64(time.Second)),
MaxConcurrency: int(provider.MaxConcurrency.Int64),
Alias: binding.Alias,
TargetAlias: binding.TargetAlias,
UpstreamName: binding.UpstreamName,
Capabilities: caps,
URLOverrides: overrides,
RateLimitHeaders: ParseRateLimitHeaders(provider.RateLimitHeaders),
}, nil
}

Expand All @@ -222,17 +226,18 @@ func RouteForProvider(p *store.Provider, secret []byte) (*Route, error) {
}
}
return &Route{
ProviderID: p.ID,
ProviderName: p.Name,
WireFormat: p.WireFormat,
BaseURL: p.BaseURL,
Credential: credential,
VerifyTLS: p.VerifyTLS,
CAPEM: p.CAPEM.String,
TimeoutConnect: time.Duration(p.TimeoutConnect * float64(time.Second)),
TimeoutRead: time.Duration(p.TimeoutRead * float64(time.Second)),
MaxConcurrency: int(p.MaxConcurrency.Int64),
Capabilities: map[string]bool{},
URLOverrides: map[string]string{},
ProviderID: p.ID,
ProviderName: p.Name,
WireFormat: p.WireFormat,
BaseURL: p.BaseURL,
Credential: credential,
VerifyTLS: p.VerifyTLS,
CAPEM: p.CAPEM.String,
TimeoutConnect: time.Duration(p.TimeoutConnect * float64(time.Second)),
TimeoutRead: time.Duration(p.TimeoutRead * float64(time.Second)),
MaxConcurrency: int(p.MaxConcurrency.Int64),
Capabilities: map[string]bool{},
URLOverrides: map[string]string{},
RateLimitHeaders: ParseRateLimitHeaders(p.RateLimitHeaders),
}, nil
}
62 changes: 62 additions & 0 deletions internal/catalog/ratelimit.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,62 @@
package catalog

import "strings"

// ValidRateLimitKeys are the metric keys a rate_limit_headers spec may map.
var ValidRateLimitKeys = []string{
"limit_requests", "remaining_requests",
"limit_tokens", "remaining_tokens",
"reset_requests", "reset_tokens",
}

func isValidRateLimitKey(key string) bool {
for _, k := range ValidRateLimitKeys {
if k == key {
return true
}
}
return false
}

// ParseRateLimitHeaders parses the stored comma-separated "key:header" spec
// into a map. Invalid pairs are silently dropped.
func ParseRateLimitHeaders(spec string) map[string]string {
if spec == "" {
return nil
}
out := make(map[string]string)
for _, pair := range strings.Split(spec, ",") {
key, header, ok := strings.Cut(pair, ":")
if !ok {
continue
}
key = strings.TrimSpace(key)
header = strings.TrimSpace(header)
if key == "" || header == "" {
continue
}
if !isValidRateLimitKey(key) {
continue
}
out[key] = header
}
if len(out) == 0 {
return nil
}
return out
}

// FormatRateLimitHeaders serialises a map back to the stored comma-separated
// "key:header" canonical form, sorted by key for stable output.
func FormatRateLimitHeaders(m map[string]string) string {
if len(m) == 0 {
return ""
}
var parts []string
for _, key := range ValidRateLimitKeys {
if header, ok := m[key]; ok && header != "" {
parts = append(parts, key+":"+header)
}
}
return strings.Join(parts, ",")
}
104 changes: 104 additions & 0 deletions internal/catalog/ratelimit_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,104 @@
package catalog

import (
"testing"
)

func TestParseRateLimitHeaders(t *testing.T) {
tests := []struct {
name string
spec string
want map[string]string
}{
{"empty", "", nil},
{"standard openai", "limit_requests:x-ratelimit-limit-requests,remaining_requests:x-ratelimit-remaining-requests,limit_tokens:x-ratelimit-limit-tokens,remaining_tokens:x-ratelimit-remaining-tokens",
map[string]string{
"limit_requests": "x-ratelimit-limit-requests",
"remaining_requests": "x-ratelimit-remaining-requests",
"limit_tokens": "x-ratelimit-limit-tokens",
"remaining_tokens": "x-ratelimit-remaining-tokens",
}},
{"with reset headers", "limit_requests:x-ratelimit-limit-requests,reset_requests:x-ratelimit-reset-requests,reset_tokens:x-ratelimit-reset-tokens",
map[string]string{
"limit_requests": "x-ratelimit-limit-requests",
"reset_requests": "x-ratelimit-reset-requests",
"reset_tokens": "x-ratelimit-reset-tokens",
}},
{"invalid key ignored", "limit_requests:x-foo,bogus_key:x-bar",
map[string]string{"limit_requests": "x-foo"}},
{"empty value ignored", "limit_requests:", nil},
{"no colon ignored", "limit_requests", nil},
{"whitespace trimmed", " limit_requests : x-foo , remaining_tokens : x-bar ",
map[string]string{"limit_requests": "x-foo", "remaining_tokens": "x-bar"}},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := ParseRateLimitHeaders(tt.spec)
if tt.want == nil {
if got != nil {
t.Fatalf("expected nil, got %v", got)
}
return
}
if len(got) != len(tt.want) {
t.Fatalf("expected %d entries, got %d: %v", len(tt.want), len(got), got)
}
for k, v := range tt.want {
if got[k] != v {
t.Errorf("key %q: expected %q, got %q", k, v, got[k])
}
}
})
}
}

func TestFormatRateLimitHeaders(t *testing.T) {
tests := []struct {
name string
m map[string]string
want string
}{
{"nil", nil, ""},
{"empty", map[string]string{}, ""},
{"single", map[string]string{"limit_requests": "x-foo"}, "limit_requests:x-foo"},
{"sorted by key order", map[string]string{
"remaining_tokens": "x-bar",
"limit_requests": "x-foo",
}, "limit_requests:x-foo,remaining_tokens:x-bar"},
{"all six", map[string]string{
"limit_requests": "a",
"remaining_requests": "b",
"limit_tokens": "c",
"remaining_tokens": "d",
"reset_requests": "e",
"reset_tokens": "f",
}, "limit_requests:a,remaining_requests:b,limit_tokens:c,remaining_tokens:d,reset_requests:e,reset_tokens:f"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := FormatRateLimitHeaders(tt.m)
if got != tt.want {
t.Fatalf("expected %q, got %q", tt.want, got)
}
})
}
}

func TestRoundtrip(t *testing.T) {
original := map[string]string{
"limit_requests": "x-ratelimit-limit-requests",
"remaining_requests": "x-ratelimit-remaining-requests",
"limit_tokens": "x-ratelimit-limit-tokens",
"remaining_tokens": "x-ratelimit-remaining-tokens",
}
spec := FormatRateLimitHeaders(original)
parsed := ParseRateLimitHeaders(spec)
if len(parsed) != len(original) {
t.Fatalf("roundtrip lost entries: %v -> %q -> %v", original, spec, parsed)
}
for k, v := range original {
if parsed[k] != v {
t.Errorf("key %q: expected %q, got %q", k, v, parsed[k])
}
}
}
20 changes: 17 additions & 3 deletions internal/metrics/metrics.go
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,11 @@ import (
type Metrics struct {
mu sync.Mutex
counters map[string]float64
gauges map[string]float64
}

func New() *Metrics {
return &Metrics{counters: make(map[string]float64)}
return &Metrics{counters: make(map[string]float64), gauges: make(map[string]float64)}
}

func key(name string, labels [][2]string) string {
Expand Down Expand Up @@ -63,16 +64,29 @@ func (m *Metrics) ObserveUnits(provider, alias, unit string, quantity float64, p
}
}

func (m *Metrics) Set(name string, labels [][2]string, value float64) {
k := key(name, labels)
m.mu.Lock()
m.gauges[k] = value
m.mu.Unlock()
}

func (m *Metrics) Render() string {
m.mu.Lock()
keys := make([]string, 0, len(m.counters))
keys := make([]string, 0, len(m.counters)+len(m.gauges))
for k := range m.counters {
keys = append(keys, k)
}
values := make(map[string]float64, len(m.counters))
for k := range m.gauges {
keys = append(keys, k)
}
values := make(map[string]float64, len(m.counters)+len(m.gauges))
for k, v := range m.counters {
values[k] = v
}
for k, v := range m.gauges {
values[k] = v
}
m.mu.Unlock()
sort.Strings(keys)
var b strings.Builder
Expand Down
Loading
Loading