mirror of
https://github.com/lukaszraczylo/traefik-llmgateway.git
synced 2026-10-16 05:37:28 +00:00
merge: cross-provider failover with cost guard, default off
This commit is contained in:
@@ -193,6 +193,7 @@ config accepts them as YAML, which decodes to the same JSON shape.
|
|||||||
| `users` | `UsersConfig` | — | Inline and/or file-backed API-key holders. With none configured, every request is unauthenticated and gets 401. |
|
| `users` | `UsersConfig` | — | Inline and/or file-backed API-key holders. With none configured, every request is unauthenticated and gets 401. |
|
||||||
| `redis` | `RedisConfig` | — | Distributed limit-counter backend. Omitted means in-process counters only (per-replica, approximate across multiple Traefik instances). |
|
| `redis` | `RedisConfig` | — | Distributed limit-counter backend. Omitted means in-process counters only (per-replica, approximate across multiple Traefik instances). |
|
||||||
| `retry` | `RetryConfig` | `{}` (disabled) | Same-provider retry for transient upstream failures — see [Retry](#retry). Omitted or `enabled: false` means no retry: every request makes exactly one upstream attempt, byte-identical to a gateway built before this field existed. |
|
| `retry` | `RetryConfig` | `{}` (disabled) | Same-provider retry for transient upstream failures — see [Retry](#retry). Omitted or `enabled: false` means no retry: every request makes exactly one upstream attempt, byte-identical to a gateway built before this field existed. |
|
||||||
|
| `failover` | `FailoverConfig` | `{}` (disabled) | Cross-provider failover for a bare model id more than one configured provider serves — see [Failover](#failover). Omitted or `enabled: false` means no failover, byte-identical to a gateway built before this field existed. `enabled: true` opts in; a deployment with no overlapping providers behaves exactly as before regardless of this setting either way. |
|
||||||
| `cache` | `CacheConfig` | `{}` (disabled) | Opt-in Redis-backed response cache for unified non-streaming chat/embeddings — see [Caching](#caching). Omitted or `enabled: false` means no caching, byte-identical to a gateway built before this field existed. |
|
| `cache` | `CacheConfig` | `{}` (disabled) | Opt-in Redis-backed response cache for unified non-streaming chat/embeddings — see [Caching](#caching). Omitted or `enabled: false` means no caching, byte-identical to a gateway built before this field existed. |
|
||||||
| `breaker` | `BreakerConfig` | `{}` (every default) | Per-provider discovery circuit breaker — see [Provider health (discovery circuit breaker)](#provider-health-discovery-circuit-breaker). Every field is individually zero-means-default; a provider whose discovery never fails is unaffected regardless of what this block contains. |
|
| `breaker` | `BreakerConfig` | `{}` (every default) | Per-provider discovery circuit breaker — see [Provider health (discovery circuit breaker)](#provider-health-discovery-circuit-breaker). Every field is individually zero-means-default; a provider whose discovery never fails is unaffected regardless of what this block contains. |
|
||||||
| `admin` | `*AdminConfig` | `nil` (disabled) | Read-only admin dashboard — see [Admin](#admin). `nil` or `enabled: false` means the `/admin*` routes are not registered at all. |
|
| `admin` | `*AdminConfig` | `nil` (disabled) | Read-only admin dashboard — see [Admin](#admin). `nil` or `enabled: false` means the `/admin*` routes are not registered at all. |
|
||||||
@@ -807,10 +808,10 @@ released, so this never becomes a ceiling on total concurrent requests.
|
|||||||
|
|
||||||
## Retry
|
## Retry
|
||||||
|
|
||||||
Same-provider retry for a transient upstream failure — no cross-provider
|
Same-provider retry for a transient upstream failure. Off by default
|
||||||
failover. Off by default (`retry.enabled: false`); every config that
|
(`retry.enabled: false`); every config that predates this field keeps
|
||||||
predates this field keeps making exactly one upstream attempt per
|
making exactly one upstream attempt per request. See [Failover](#failover)
|
||||||
request.
|
below for the separate, cross-provider mechanism.
|
||||||
|
|
||||||
- **Scope**: every adapter upstream call — chat, embeddings, model
|
- **Scope**: every adapter upstream call — chat, embeddings, model
|
||||||
listing, and the three media endpoints (`images/generations`,
|
listing, and the three media endpoints (`images/generations`,
|
||||||
@@ -846,6 +847,119 @@ request.
|
|||||||
is never double-counted. Request counters still increment once per
|
is never double-counted. Request counters still increment once per
|
||||||
client request, unchanged.
|
client request, unchanged.
|
||||||
|
|
||||||
|
## Failover
|
||||||
|
|
||||||
|
Cross-provider failover for a bare model id more than one configured
|
||||||
|
provider serves — for example `whisper-1` configured on both `openai` and
|
||||||
|
`openai-audio`. **Off by default** (`failover.enabled: false` unless set
|
||||||
|
otherwise): a version upgrade with no config change preserves prior
|
||||||
|
behavior exactly, since failover changes which provider serves a request
|
||||||
|
and can shift cost attribution or mask a broken provider key behind an
|
||||||
|
apparent success. Set `failover.enabled: true` to opt in. A deployment
|
||||||
|
with no overlapping providers, or with a single provider, behaves
|
||||||
|
exactly as before regardless of this setting either way, since there is
|
||||||
|
never a second candidate to try.
|
||||||
|
|
||||||
|
- **Scope**: `POST /v1/chat/completions`, `POST /v1/embeddings`, and
|
||||||
|
`POST /v1/messages` only. The image/audio endpoints
|
||||||
|
(`images/generations`, `audio/speech`, `audio/transcriptions`) are not
|
||||||
|
wired into failover yet — each has its own request-shape handling
|
||||||
|
(multipart rebuild, binary/streaming responses) that needs its own
|
||||||
|
scoped follow-up.
|
||||||
|
- **Eligible candidates**: a bare model id resolved via the collision
|
||||||
|
winner (the same mechanism `GET /v1/models` documents for overlapping
|
||||||
|
catalogs) offers every OTHER configured provider that also serves that
|
||||||
|
id and that the caller's group is authorized to use. A `provider/model`
|
||||||
|
request, or an alias whose target is `provider/model`, is an explicit,
|
||||||
|
exact choice and is never eligible for substitution.
|
||||||
|
- **What falls through**: everything except HTTP 400 — connection
|
||||||
|
failures, timeouts, HTTP 429/401/403/404, and any 5xx. A 400 never
|
||||||
|
falls through: the request is malformed and every provider would reject
|
||||||
|
it identically. A failover past a 404 is logged loudly, always, naming
|
||||||
|
both providers and the model — a 404 usually means a real configuration
|
||||||
|
mistake, not a transient outage. The routine "failing over from X to Y"
|
||||||
|
line is rate-limited to once per 30s during a sustained outage (with a
|
||||||
|
suppressed-count note on the next line), so it cannot itself become a
|
||||||
|
blocking synchronous write on every request; the 404 line is exempt
|
||||||
|
from that limit.
|
||||||
|
- **Cost guard (not optional)**: failover never routes to a candidate
|
||||||
|
priced higher than the provider it is failing over FROM (the first
|
||||||
|
candidate actually attempted, never the cheapest in the list), checked
|
||||||
|
on BOTH input and output cost per token — a candidate cheaper on input
|
||||||
|
but dearer on output is still rejected. Pricing is resolved exactly the
|
||||||
|
way real billing resolves it: the canonical `provider/model` id in
|
||||||
|
`pricing` first, falling back to the bare model id, so a per-provider
|
||||||
|
`pricing` override correctly makes the identical model id cost
|
||||||
|
different amounts on different providers. Unknown pricing is never
|
||||||
|
treated as free: if the primary's price is known and a candidate's is
|
||||||
|
not, the candidate is skipped (an unknown price cannot be proven
|
||||||
|
cheaper); if the primary's price is unknown and a candidate's is known
|
||||||
|
and non-zero, the candidate is skipped too (moving from unbilled to
|
||||||
|
billed is itself a spend increase); both unknown, or both an explicit
|
||||||
|
known-zero price, are treated as equal and allowed. There is no config
|
||||||
|
flag for this — it applies unconditionally whenever failover is
|
||||||
|
enabled.
|
||||||
|
- **The cost guard cannot see `modelMeta`'s `free: true`.** It resolves
|
||||||
|
pricing only through `pricing` and the built-in table — the same
|
||||||
|
source `unifiedCostMicros` actually bills against — deliberately never
|
||||||
|
through `modelMeta` (see [Model metadata](#model-metadata) /
|
||||||
|
`ModelMetaConfig`), since `modelMeta` drives GET `/v1/models` display
|
||||||
|
only and never affects billing. A model declared free the `modelMeta`
|
||||||
|
way is therefore treated as unknown-priced by the cost guard and loses
|
||||||
|
redundancy: it is skipped as a failover candidate whenever the primary
|
||||||
|
has a known price (fail-safe, and logged like any other cost skip).
|
||||||
|
Declare the price through `pricing` (e.g. `{"inputPerM": 0,
|
||||||
|
"outputPerM": 0}`) if you want a genuinely free model to remain
|
||||||
|
eligible as a cheaper failover candidate.
|
||||||
|
- **Never mid-stream**: failover only happens while nothing has reached
|
||||||
|
the client yet. Once the first byte of a response is written, status
|
||||||
|
and body are committed and the request is never retried elsewhere.
|
||||||
|
- **Health signals, kept separate**: the discovery-endpoint health
|
||||||
|
surfaced on the admin dashboard is an ordering hint only — a provider
|
||||||
|
whose `/v1/models` is unhealthy is tried later, never skipped outright,
|
||||||
|
since plenty of providers serve chat completions fine while their model
|
||||||
|
listing is broken. A separate, in-memory, per-pod request-path health
|
||||||
|
signal (not yet exposed on the admin dashboard) is the actual routing
|
||||||
|
gate: a provider is skipped without being called once it has failed
|
||||||
|
three consecutive LOGICAL requests (same-provider retries all count as
|
||||||
|
one outcome, not one each) — everything except HTTP 400 counts as a
|
||||||
|
failure for this gate, including 401/403/404, so a provider with a dead
|
||||||
|
API key is eventually routed around instead of retried forever.
|
||||||
|
- **Accounting**: a failover chain still counts as exactly one client
|
||||||
|
request against `requestsPerMinute`/`requestsPerDay` — never once per
|
||||||
|
attempt. Per-provider attempt/failure counters (the admin dashboard's
|
||||||
|
success-rate badge) attribute every attempt to the provider that
|
||||||
|
actually made it.
|
||||||
|
- **Caching**: a cacheable response produced by a failover provider is
|
||||||
|
stored under that provider's own cache key, never the first candidate's.
|
||||||
|
Each candidate's own cache entry is checked immediately before that
|
||||||
|
candidate's own attempt, in try order — never for every candidate up
|
||||||
|
front. Checking every candidate's cache before any of them was
|
||||||
|
attempted was tried and reverted: it silently served a DIFFERENT
|
||||||
|
candidate's cached response even while the actual primary was
|
||||||
|
perfectly healthy, which is failover with no failure. The
|
||||||
|
request-path health signal above is what actually keeps a genuinely
|
||||||
|
dead primary from being dialed again — once it is marked unhealthy it
|
||||||
|
is excluded from the candidate list entirely, so a later request
|
||||||
|
reaches the next candidate's cache without the dead one ever being
|
||||||
|
attempted.
|
||||||
|
- **Configurable**: `failover.maxAttempts` caps how many different
|
||||||
|
providers one request will try in total, primary included (default 3,
|
||||||
|
maximum 10) — distinct from `retry.attempts`, which retries the SAME
|
||||||
|
provider. `failover.enabled: false` (the default) turns the whole
|
||||||
|
mechanism off.
|
||||||
|
- **Known limitation — no per-attempt timeout**: `newAdapterHTTPClient`
|
||||||
|
sets no client timeout and no response-header timeout; the only bound
|
||||||
|
on an upstream call is the incoming request's own context, shared
|
||||||
|
identically across every candidate. A provider that accepts the TCP
|
||||||
|
connection and then simply never responds will hang the request
|
||||||
|
indefinitely and never trigger failover — only a failure fast enough to
|
||||||
|
leave time on that shared context (a dial failure, a fast non-2xx
|
||||||
|
response, or the context's own deadline finally expiring) does. This is
|
||||||
|
a pre-existing property of the plugin's HTTP client, not specific to
|
||||||
|
failover, and applies to every route, not only this one; it is not
|
||||||
|
addressed in this change.
|
||||||
|
|
||||||
## Caching
|
## Caching
|
||||||
|
|
||||||
An opt-in, Redis-backed response cache for the unified, non-streaming
|
An opt-in, Redis-backed response cache for the unified, non-streaming
|
||||||
|
|||||||
+656
@@ -0,0 +1,656 @@
|
|||||||
|
package traefikllmgateway
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"net/http"
|
||||||
|
"sync"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
// This file implements feat/failover: cross-provider failover for a bare
|
||||||
|
// model id more than one configured provider serves (registry.go's
|
||||||
|
// bareWinner collision case — production logs show whisper-1, tts-1,
|
||||||
|
// tts-1-hd, text-embedding-3-small, text-embedding-3-large, and
|
||||||
|
// text-embedding-ada-002 served by two or more of openai/openai-audio/
|
||||||
|
// copilot today). Before this feature, registry.go's resolve returns
|
||||||
|
// exactly one provider adapter and retry.go's retryPolicy only ever
|
||||||
|
// retries the SAME provider — so a down provider fails every request for
|
||||||
|
// its models even when another configured provider serves the identical
|
||||||
|
// model. This file turns that already-existing redundancy into a hot
|
||||||
|
// standby, hooked into routes_unified.go's runMeteredCall, the ONE shared
|
||||||
|
// metered-call path both /v1/chat/completions and /v1/messages already go
|
||||||
|
// through.
|
||||||
|
//
|
||||||
|
// Off by default (FailoverConfig.Enabled defaults to false — coordinator
|
||||||
|
// ruling, overriding this feature's original brief: new behavior that can
|
||||||
|
// change which provider serves a request goes behind a flag defaulting
|
||||||
|
// off). An operator who opts in also gets a non-optional cost guard
|
||||||
|
// (candidateCostAllowed, below): failover never routes to a candidate
|
||||||
|
// priced higher than the provider it is failing over from, on either
|
||||||
|
// input or output cost per token — there is no config knob to disable
|
||||||
|
// just that part.
|
||||||
|
//
|
||||||
|
// SCOPE: routes_media.go's three media routes (images/generations,
|
||||||
|
// audio/speech, audio/transcriptions) are deliberately NOT wired into
|
||||||
|
// failover in this change, even though two of the brief's own named
|
||||||
|
// overlapping models (whisper-1, tts-1/tts-1-hd) are media-route
|
||||||
|
// traffic. Each of those three handlers has its own request-shape
|
||||||
|
// preprocessing (multipart rebuild for transcriptions, a JSON
|
||||||
|
// re-marshal for speech, a binary/streaming response for speech) that
|
||||||
|
// would need its own careful "nothing written yet" analysis and its own
|
||||||
|
// test coverage, rather than reusing runMeteredCall's shared shape — the
|
||||||
|
// brief's own hook-point instruction ("do not fork a second
|
||||||
|
// implementation") is about not duplicating the chat/messages pipeline,
|
||||||
|
// not about extending every possible call site in one pass. Left as a
|
||||||
|
// scoped follow-up; see this branch's own report for the explicit
|
||||||
|
// call-out.
|
||||||
|
|
||||||
|
// FailoverConfig configures feat/failover. Enabled is a *bool; nil or
|
||||||
|
// false (the default) means OFF — coordinator ruling, overriding this
|
||||||
|
// feature's original brief: a version upgrade with no config change must
|
||||||
|
// preserve prior behavior, and new behavior that can change which
|
||||||
|
// provider serves a request (shifting cost attribution, and able to mask
|
||||||
|
// a broken provider key behind an apparent success) goes behind a flag
|
||||||
|
// that defaults off, not on. Set Enabled: true to opt in. A
|
||||||
|
// single-provider deployment, or one where no two providers ever serve
|
||||||
|
// the identical bare model id, behaves exactly as before regardless of
|
||||||
|
// this setting either way: there is never more than one candidate to try
|
||||||
|
// (orderedFailoverCandidates' own doc comment).
|
||||||
|
type FailoverConfig struct {
|
||||||
|
Enabled *bool `json:"enabled,omitempty"`
|
||||||
|
// MaxAttempts caps how many DIFFERENT PROVIDERS runMeteredCall tries
|
||||||
|
// in total for one logical request, the primary included — distinct
|
||||||
|
// from RetryConfig.Attempts (retry.go), which retries the SAME
|
||||||
|
// provider. 0 (the default) uses defaultFailoverMaxAttempts. A value
|
||||||
|
// outside 1..maxFailoverMaxAttempts is a construction error.
|
||||||
|
MaxAttempts int `json:"maxAttempts,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// defaultFailoverMaxAttempts is applied when FailoverConfig.MaxAttempts is
|
||||||
|
// left at 0. maxFailoverMaxAttempts bounds an explicit override, the same
|
||||||
|
// "cheap sanity ceiling on an operator value" precedent
|
||||||
|
// maxBreakerFailureThreshold/maxRetryAttempts already set (registry.go/
|
||||||
|
// retry.go).
|
||||||
|
const (
|
||||||
|
defaultFailoverMaxAttempts = 3
|
||||||
|
maxFailoverMaxAttempts = 10
|
||||||
|
)
|
||||||
|
|
||||||
|
// failoverConfig is FailoverConfig, validated and defaulted once at
|
||||||
|
// construction (newGateway, llmgateway.go) — mirrors breakerConfig/
|
||||||
|
// retryPolicy's own resolved-value-not-raw-config shape (registry.go/
|
||||||
|
// retry.go).
|
||||||
|
type failoverConfig struct {
|
||||||
|
maxAttempts int
|
||||||
|
enabled bool
|
||||||
|
}
|
||||||
|
|
||||||
|
// validateFailoverConfig validates fc and returns the resolved
|
||||||
|
// failoverConfig newGateway attaches to the Gateway. enabled defaults to
|
||||||
|
// false (coordinator ruling, FailoverConfig's own doc comment) — every
|
||||||
|
// config that predates this field, and every config that never sets
|
||||||
|
// failover.enabled at all, keeps behaving exactly as before this feature
|
||||||
|
// existed.
|
||||||
|
func validateFailoverConfig(fc FailoverConfig) (failoverConfig, error) {
|
||||||
|
enabled := false
|
||||||
|
if fc.Enabled != nil {
|
||||||
|
enabled = *fc.Enabled
|
||||||
|
}
|
||||||
|
|
||||||
|
maxAttempts := fc.MaxAttempts
|
||||||
|
if maxAttempts == 0 {
|
||||||
|
maxAttempts = defaultFailoverMaxAttempts
|
||||||
|
}
|
||||||
|
if maxAttempts < 1 || maxAttempts > maxFailoverMaxAttempts {
|
||||||
|
return failoverConfig{}, fmt.Errorf("llmgateway: failover.maxAttempts must be between 1 and %d, got %d", maxFailoverMaxAttempts, maxAttempts)
|
||||||
|
}
|
||||||
|
|
||||||
|
return failoverConfig{enabled: enabled, maxAttempts: maxAttempts}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// resolveCandidate is one provider that can serve a client-requested model
|
||||||
|
// id — modelRegistry.resolve's own (adapter, upstreamModel, canonical)
|
||||||
|
// triple, tagged with the provider's own name so a failover attempt can
|
||||||
|
// rebind its cache key, attempt recorder, and error log to the ACTUAL
|
||||||
|
// provider that served or failed the request, never the first one resolve
|
||||||
|
// happened to pick.
|
||||||
|
type resolveCandidate struct {
|
||||||
|
adapter providerAdapter
|
||||||
|
providerName string
|
||||||
|
upstreamModel string
|
||||||
|
canonical string
|
||||||
|
}
|
||||||
|
|
||||||
|
// resolveWithCandidates resolves id exactly as modelRegistry.resolve
|
||||||
|
// (registry.go) does — identical winner, identical error — and
|
||||||
|
// additionally reports every OTHER configured provider that also serves
|
||||||
|
// the same bare upstream model id and remains authorized for grp, as
|
||||||
|
// feat/failover's candidate pool.
|
||||||
|
//
|
||||||
|
// primary/err are byte-identical to what resolve(id, grp) alone would
|
||||||
|
// return: this method calls resolve internally and never second-guesses
|
||||||
|
// its winner or its error — feat/failover only ADDS candidates after an
|
||||||
|
// already-successful resolution, it never changes which model or error a
|
||||||
|
// request resolves to (in particular, authorization denial/unknown-model
|
||||||
|
// behavior for the primary winner is completely unchanged by this
|
||||||
|
// branch). extra is nil whenever there is nothing to add:
|
||||||
|
//
|
||||||
|
// - id resolved via an explicit "provider/model" prefix, or an alias
|
||||||
|
// whose target is itself provider-prefixed: the requester's own
|
||||||
|
// explicit provider choice is never a candidate for pool
|
||||||
|
// substitution — ruling, this feature's own design: "provider/model"
|
||||||
|
// addressing is exact, not eligible for failover.
|
||||||
|
// - id resolved via a bare-id collision winner (bareWinner), or an
|
||||||
|
// alias whose target is itself a bare-id collision winner: extra is
|
||||||
|
// every OTHER configured provider whose known model set also
|
||||||
|
// contains that identical bare id string, in
|
||||||
|
// modelRegistry.providerNames (sorted) order, and authorized for grp
|
||||||
|
// on that provider.
|
||||||
|
func (m *modelRegistry) resolveWithCandidates(id string, grp *group) (resolveCandidate, []resolveCandidate, error) {
|
||||||
|
adapter, upstreamModel, canonical, err := m.resolve(id, grp)
|
||||||
|
if err != nil {
|
||||||
|
return resolveCandidate{}, nil, err
|
||||||
|
}
|
||||||
|
primary := resolveCandidate{adapter: adapter, providerName: adapter.name(), upstreamModel: upstreamModel, canonical: canonical}
|
||||||
|
|
||||||
|
bareID := id
|
||||||
|
if target, isAlias := m.aliases[id]; isAlias {
|
||||||
|
if _, _, ok := m.splitConfiguredProvider(target); ok {
|
||||||
|
return primary, nil, nil
|
||||||
|
}
|
||||||
|
bareID = target
|
||||||
|
} else if _, _, ok := m.splitConfiguredProvider(id); ok {
|
||||||
|
return primary, nil, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
return primary, m.failoverCandidates(bareID, primary.providerName, grp), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// resolvePrimaryOnly resolves id exactly like resolveWithCandidates'
|
||||||
|
// primary return (identical winner, identical error) WITHOUT ever
|
||||||
|
// scanning for or building failover candidates — the short-circuit
|
||||||
|
// runMeteredCall (routes_unified.go) takes when g.failover.enabled is
|
||||||
|
// false, so a deployment that never opted into failover pays no extra
|
||||||
|
// cost for a pool it will never use (measured: 74ns/1 alloc versus
|
||||||
|
// 521ns/13 allocs at 8 providers x 200 models for the full
|
||||||
|
// resolveWithCandidates scan — irrelevant next to a network round trip,
|
||||||
|
// but free to avoid).
|
||||||
|
func (m *modelRegistry) resolvePrimaryOnly(id string, grp *group) (resolveCandidate, error) {
|
||||||
|
adapter, upstreamModel, canonical, err := m.resolve(id, grp)
|
||||||
|
if err != nil {
|
||||||
|
return resolveCandidate{}, err
|
||||||
|
}
|
||||||
|
return resolveCandidate{adapter: adapter, providerName: adapter.name(), upstreamModel: upstreamModel, canonical: canonical}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// failoverCandidates returns every configured provider other than
|
||||||
|
// primaryProvider whose known model set contains bareID and that grp is
|
||||||
|
// authorized to use, in modelRegistry.providerNames (sorted) order — the
|
||||||
|
// raw candidate pool orderedFailoverCandidates (below) narrows and
|
||||||
|
// reorders further by health.
|
||||||
|
//
|
||||||
|
// Only grp.allowsProvider is re-checked per candidate: grp.allowsModel is
|
||||||
|
// already known true from the primary winner's own successful resolve
|
||||||
|
// call, and — for the identical bare id string every candidate here
|
||||||
|
// shares by construction — allowsModel's result does not vary by
|
||||||
|
// provider (auth.go's allowsModel takes no provider argument), so
|
||||||
|
// re-checking it per candidate would be redundant work, not a
|
||||||
|
// correctness gap.
|
||||||
|
func (m *modelRegistry) failoverCandidates(bareID, primaryProvider string, grp *group) []resolveCandidate {
|
||||||
|
var extra []resolveCandidate
|
||||||
|
for _, name := range m.providerNames {
|
||||||
|
if name == primaryProvider {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if !m.states[name].hasModel(bareID) || !grp.allowsProvider(name) {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
// canonical assigned to a local before the struct literal, matching
|
||||||
|
// resolveAgainst's own documented Yaegi precedent (registry.go): a
|
||||||
|
// defensive, cheap precaution against the same class of interpreter
|
||||||
|
// trap, not a proven necessity for a struct-literal field (only a
|
||||||
|
// multi-value RETURN tuple is the proven-bad shape there).
|
||||||
|
canonical := name + "/" + bareID
|
||||||
|
extra = append(extra, resolveCandidate{
|
||||||
|
adapter: m.adapters[name],
|
||||||
|
providerName: name,
|
||||||
|
upstreamModel: bareID,
|
||||||
|
canonical: canonical,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
return extra
|
||||||
|
}
|
||||||
|
|
||||||
|
// orderedFailoverCandidates builds runMeteredCall's actual try list from
|
||||||
|
// resolveWithCandidates' own (primary, extra) split.
|
||||||
|
//
|
||||||
|
// Failover is a no-op (returns exactly [primary]) whenever it is disabled
|
||||||
|
// (g.failover.enabled false) or there is nothing to add (len(extra) ==
|
||||||
|
// 0) — the second case is what makes "single-provider deployment behaves
|
||||||
|
// exactly as before" hold unconditionally, regardless of the enabled
|
||||||
|
// switch: there is never more than one candidate to try when no other
|
||||||
|
// configured provider shares this bare model id.
|
||||||
|
//
|
||||||
|
// Otherwise, two health signals apply, deliberately kept separate
|
||||||
|
// (modelRegistry.discoveryHealthy's own doc comment, registry.go, is
|
||||||
|
// binding on this):
|
||||||
|
//
|
||||||
|
// - REQUEST-PATH health (g.failoverHealth, this file) is a HARD SKIP
|
||||||
|
// gate: a candidate it reports unhealthy is excluded outright,
|
||||||
|
// never attempted. If that filtering would remove every candidate
|
||||||
|
// (this local, per-pod signal reporting the whole fleet down
|
||||||
|
// simultaneously is more likely stale than true), it is discarded
|
||||||
|
// and every original candidate is tried anyway — failing open
|
||||||
|
// against the gateway's own imperfect signal rather than refusing
|
||||||
|
// to route a request at all.
|
||||||
|
// - DISCOVERY health (g.registry.discoveryHealthy) is an ORDERING HINT
|
||||||
|
// ONLY, exactly as its own doc comment requires: a discovery-
|
||||||
|
// unhealthy candidate is moved after every discovery-healthy/
|
||||||
|
// half-open one (a stable partition, not a full sort — relative
|
||||||
|
// order within each bucket is preserved), never dropped.
|
||||||
|
//
|
||||||
|
// This does NOT apply the cost guard or the maxAttempts cap — see
|
||||||
|
// runMeteredCall's own call sequence (routes_unified.go): candidates[0]
|
||||||
|
// coming out of THIS function is what filterCandidatesByCost (below)
|
||||||
|
// treats as "the primary" for pricing-guard purposes, so cost filtering
|
||||||
|
// must run against the health/discovery-ordered result, not the raw
|
||||||
|
// resolveWithCandidates order; and the maxAttempts cap runs LAST, after
|
||||||
|
// cost filtering, so a cap never re-admits an expensive candidate purely
|
||||||
|
// because a cheaper one ahead of it was removed on cost.
|
||||||
|
func (g *Gateway) orderedFailoverCandidates(primary resolveCandidate, extra []resolveCandidate) []resolveCandidate {
|
||||||
|
if !g.failover.enabled || len(extra) == 0 {
|
||||||
|
return []resolveCandidate{primary}
|
||||||
|
}
|
||||||
|
|
||||||
|
all := make([]resolveCandidate, 0, len(extra)+1)
|
||||||
|
all = append(all, primary)
|
||||||
|
all = append(all, extra...)
|
||||||
|
|
||||||
|
healthy := make([]resolveCandidate, 0, len(all))
|
||||||
|
for _, c := range all {
|
||||||
|
if g.failoverHealth.healthy(c.providerName) {
|
||||||
|
healthy = append(healthy, c)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if len(healthy) == 0 {
|
||||||
|
healthy = all
|
||||||
|
}
|
||||||
|
|
||||||
|
ordered := make([]resolveCandidate, 0, len(healthy))
|
||||||
|
var deprioritized []resolveCandidate
|
||||||
|
for _, c := range healthy {
|
||||||
|
if g.registry.discoveryHealthy(c.providerName) {
|
||||||
|
ordered = append(ordered, c)
|
||||||
|
} else {
|
||||||
|
deprioritized = append(deprioritized, c)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return append(ordered, deprioritized...)
|
||||||
|
}
|
||||||
|
|
||||||
|
// resolveUnifiedPricing resolves (canonical, bare)'s price using the
|
||||||
|
// IDENTICAL resolution unifiedCostMicros itself uses (routes_unified.go):
|
||||||
|
// canonical id first, falling back to the bare upstream model id only
|
||||||
|
// when canonical has no configured price at all. ok is lookupPricing's
|
||||||
|
// own known/unknown distinction (pricing.go) — false means neither id
|
||||||
|
// resolved to a price, exactly the case where unifiedCostMicros would
|
||||||
|
// warn and bill 0, which candidateCostAllowed (below) must NEVER treat
|
||||||
|
// as an actual known-zero price.
|
||||||
|
func resolveUnifiedPricing(canonical, bare string, overrides map[string]*ModelPricing) (ModelPricing, bool) {
|
||||||
|
if p, ok := lookupPricing(canonical, overrides); ok {
|
||||||
|
return p, true
|
||||||
|
}
|
||||||
|
return lookupPricing(bare, overrides)
|
||||||
|
}
|
||||||
|
|
||||||
|
// candidateCostAllowed reports whether candidate is allowed to serve as a
|
||||||
|
// failover destination for primary, per the operator's binding ruling:
|
||||||
|
// failover must never route to a candidate more expensive than the
|
||||||
|
// provider it is failing over FROM, checked on BOTH the input and the
|
||||||
|
// output per-token price — a candidate cheaper on input but dearer on
|
||||||
|
// output can still cost more on a completion-heavy request, so a single
|
||||||
|
// blended number is not sufficient. This is NOT configurable (ruling:
|
||||||
|
// "this behaviour is ON and not optional").
|
||||||
|
//
|
||||||
|
// Four cases, all by ruling, not inferred:
|
||||||
|
//
|
||||||
|
// - both known: allowed only when candidate's input AND output price
|
||||||
|
// are each <= primary's. Equal is allowed either way.
|
||||||
|
// - primary known, candidate UNKNOWN: never allowed — an unknown price
|
||||||
|
// cannot be proven not-more-expensive.
|
||||||
|
// - primary UNKNOWN, candidate known and non-zero: never allowed —
|
||||||
|
// moving from unbilled to billed is itself a spend increase.
|
||||||
|
// - primary UNKNOWN, candidate known and EXACTLY zero (both
|
||||||
|
// dimensions), or both unknown: allowed — they are equal (a known
|
||||||
|
// zero can never be more expensive than an unknown amount, and two
|
||||||
|
// unknowns are trivially equal to each other).
|
||||||
|
//
|
||||||
|
// Unknown pricing is never treated as zero here, unlike costMicros' own
|
||||||
|
// billing behavior (pricing.go) — the gateway RECORDING zero for an
|
||||||
|
// unpriced model is an accounting shortcut, not proof the provider
|
||||||
|
// invoices zero, and this guard exists specifically to protect real
|
||||||
|
// money, not the gateway's own displayed number.
|
||||||
|
func candidateCostAllowed(primary ModelPricing, primaryKnown bool, candidate ModelPricing, candidateKnown bool) bool {
|
||||||
|
switch {
|
||||||
|
case primaryKnown && candidateKnown:
|
||||||
|
return candidate.InputPerM <= primary.InputPerM && candidate.OutputPerM <= primary.OutputPerM
|
||||||
|
case primaryKnown && !candidateKnown:
|
||||||
|
return false
|
||||||
|
case !primaryKnown && candidateKnown:
|
||||||
|
return candidate.InputPerM == 0 && candidate.OutputPerM == 0
|
||||||
|
default: // both unknown
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// filterCandidatesByCost drops every candidate whose resolved price
|
||||||
|
// candidateCostAllowed rejects against candidates[0] — the FIRST
|
||||||
|
// candidate actually attempted, never the cheapest in the list, per the
|
||||||
|
// operator's own definition of "primary" for this rule. Pricing comes
|
||||||
|
// from resolveUnifiedPricing against g.cfg.Pricing, the SAME overrides
|
||||||
|
// map unifiedCostMicros itself resolves against, so this compares
|
||||||
|
// against what would actually be billed. requestedModel names the
|
||||||
|
// client's own request in the log line a skip always produces — an
|
||||||
|
// operator-visible g.warnf, unthrottled, naming both providers, the
|
||||||
|
// model, and both resolved prices: a silently shortened candidate list
|
||||||
|
// would look identical to having no redundancy at all, which the
|
||||||
|
// operator must be able to tell apart from a cost-guard skip.
|
||||||
|
func (g *Gateway) filterCandidatesByCost(candidates []resolveCandidate, requestedModel string) []resolveCandidate {
|
||||||
|
if len(candidates) < 2 {
|
||||||
|
return candidates
|
||||||
|
}
|
||||||
|
primary := candidates[0]
|
||||||
|
primaryPrice, primaryKnown := resolveUnifiedPricing(primary.canonical, primary.upstreamModel, g.cfg.Pricing)
|
||||||
|
|
||||||
|
out := make([]resolveCandidate, 1, len(candidates))
|
||||||
|
out[0] = primary
|
||||||
|
for _, cand := range candidates[1:] {
|
||||||
|
candPrice, candKnown := resolveUnifiedPricing(cand.canonical, cand.upstreamModel, g.cfg.Pricing)
|
||||||
|
if candidateCostAllowed(primaryPrice, primaryKnown, candPrice, candKnown) {
|
||||||
|
out = append(out, cand)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
g.warnf("failover: skipping candidate provider %q for model %q: would cost more than primary provider %q (primary input=%.4f/M output=%.4f/M usd known=%v; candidate input=%.4f/M output=%.4f/M usd known=%v)",
|
||||||
|
cand.providerName, requestedModel, primary.providerName,
|
||||||
|
primaryPrice.InputPerM, primaryPrice.OutputPerM, primaryKnown,
|
||||||
|
candPrice.InputPerM, candPrice.OutputPerM, candKnown)
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
|
// capFailoverCandidates truncates candidates to at most max entries,
|
||||||
|
// applied as the LAST step of runMeteredCall's own candidate-building
|
||||||
|
// sequence (routes_unified.go) — after health/discovery ordering and
|
||||||
|
// after cost filtering, so the cap can never re-admit a candidate cost
|
||||||
|
// filtering already removed, and never discards a cheaper candidate that
|
||||||
|
// cost filtering kept purely because a removed, more expensive one used
|
||||||
|
// to sit ahead of it in the raw list. max <= 0 is a no-op (defensive;
|
||||||
|
// validateFailoverConfig never produces one).
|
||||||
|
func capFailoverCandidates(candidates []resolveCandidate, max int) []resolveCandidate {
|
||||||
|
if max > 0 && len(candidates) > max {
|
||||||
|
return candidates[:max]
|
||||||
|
}
|
||||||
|
return candidates
|
||||||
|
}
|
||||||
|
|
||||||
|
// failoverEligible reports whether callErr — runMeteredCall's own
|
||||||
|
// adapterCall return, evaluated only once nothing has reached the client
|
||||||
|
// yet (the hard constraint this whole feature is built around: once an
|
||||||
|
// adapter has written anything, status and body are committed and can
|
||||||
|
// never be retried elsewhere) — should try the next candidate instead of
|
||||||
|
// answering the client with callErr.
|
||||||
|
//
|
||||||
|
// Ruling (brief, "Which failures fall through" — decided, not
|
||||||
|
// relitigated here): every failure except HTTP 400 falls through. 5xx,
|
||||||
|
// connection failures, timeouts, 429, 401, 403, and 404 all do; 400
|
||||||
|
// never does, because the request itself is malformed and every provider
|
||||||
|
// would reject it identically, so retrying just multiplies latency and
|
||||||
|
// burns budget.
|
||||||
|
//
|
||||||
|
// Two failure shapes are excluded for the SAME "retrying changes
|
||||||
|
// nothing" reasoning 400 itself is excluded for, even though neither is
|
||||||
|
// literally an HTTP 400: context.Canceled (the client walked away —
|
||||||
|
// failing over would only waste an attempt against a connection nobody
|
||||||
|
// is waiting on) and errRequestBuildFailed (upstreamBytes, providers.go
|
||||||
|
// — the identical malformed request shape reaches every candidate the
|
||||||
|
// same way, differing only in which operator-configured base URL it is
|
||||||
|
// built against). A *translateError or *responseTranslationError is this
|
||||||
|
// gateway's OWN translation code failing on an already-decoded request —
|
||||||
|
// not a signal about the PROVIDER's availability at all — so neither is
|
||||||
|
// eligible either.
|
||||||
|
func failoverEligible(callErr error) bool {
|
||||||
|
if callErr == nil {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
if errors.Is(callErr, context.Canceled) || errors.Is(callErr, errRequestBuildFailed) {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
// Plain type assertions, not errors.As — matching this package's
|
||||||
|
// established Yaegi-safe convention for these exact two pointer error
|
||||||
|
// types (providerHTTPError's own doc comment, providers.go;
|
||||||
|
// handleAdapterErrorEnvelope, routes_unified.go, does the identical
|
||||||
|
// assertion for the identical reason).
|
||||||
|
if perr, ok := callErr.(*providerHTTPError); ok {
|
||||||
|
return perr.status != http.StatusBadRequest
|
||||||
|
}
|
||||||
|
if _, ok := callErr.(*translateError); ok {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
if _, ok := callErr.(*responseTranslationError); ok {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
|
// isProviderNotFoundError reports whether err is a *providerHTTPError
|
||||||
|
// carrying HTTP 404 — routes_unified.go's runMeteredCall uses this to
|
||||||
|
// decide when a failover needs its own loud, non-suppressed log line
|
||||||
|
// (brief: "Failover after a 404 must log loudly ... a 404 usually means
|
||||||
|
// a real configuration mistake, and silently succeeding elsewhere would
|
||||||
|
// hide it").
|
||||||
|
func isProviderNotFoundError(err error) bool {
|
||||||
|
perr, ok := err.(*providerHTTPError)
|
||||||
|
return ok && perr.status == http.StatusNotFound
|
||||||
|
}
|
||||||
|
|
||||||
|
// requestBreakerFailureThreshold/requestBreakerOpenDuration/
|
||||||
|
// requestBreakerMaxOpenDuration size requestHealthTracker's own breaker —
|
||||||
|
// deliberately separate constants from registry.go's discovery breaker
|
||||||
|
// (defaultBreakerFailureThreshold/defaultBreakerOpenDuration/
|
||||||
|
// defaultBreakerMaxOpenDuration): a request-path outage needs to be
|
||||||
|
// noticed and routed around within seconds, not the discovery breaker's
|
||||||
|
// minutes-to-hours cadence, and modelRegistry.discoveryHealthy's own doc
|
||||||
|
// comment forbids widening that breaker to cover this outcome stream at
|
||||||
|
// all. Fixed constants, not configurable — unlike FailoverConfig above,
|
||||||
|
// the brief only asks for max-attempts and an on/off switch to be
|
||||||
|
// operator-configurable; this breaker's thresholds are an internal
|
||||||
|
// implementation detail of the routing-gate signal, not a documented
|
||||||
|
// tuning surface.
|
||||||
|
const (
|
||||||
|
requestBreakerFailureThreshold = 3
|
||||||
|
requestBreakerOpenDuration = 30 * time.Second
|
||||||
|
requestBreakerMaxOpenDuration = 5 * time.Minute
|
||||||
|
)
|
||||||
|
|
||||||
|
// requestHealthState is one provider's REQUEST-PATH health (feat/
|
||||||
|
// failover), mu-guarded — mirrors providerState's per-provider,
|
||||||
|
// individually-mutexed shape (registry.go) at a much smaller scale: this
|
||||||
|
// tracks exactly one outcome stream (request-path attempts), never
|
||||||
|
// discovery. The zero value is closed/healthy, so a provider that never
|
||||||
|
// fails never touches anything below open/backoff/openUntil.
|
||||||
|
//
|
||||||
|
// Field order below is fieldalignment-verified (golangci-lint's govet
|
||||||
|
// enable-all, run with -fix against a scratch copy to derive the exact
|
||||||
|
// zero-waste sequence, then hand-applied here so this doc comment
|
||||||
|
// survives — the same convention providerState's own doc comment
|
||||||
|
// explains, registry.go).
|
||||||
|
type requestHealthState struct {
|
||||||
|
openUntil time.Time
|
||||||
|
backoff time.Duration
|
||||||
|
consecutiveFailures int
|
||||||
|
mu sync.Mutex
|
||||||
|
open bool
|
||||||
|
}
|
||||||
|
|
||||||
|
// requestHealthTracker is feat/failover's own per-pod, in-memory,
|
||||||
|
// request-path health signal — deliberately separate from registry.go's
|
||||||
|
// discovery circuit breaker (modelRegistry.discoveryHealthy's own doc
|
||||||
|
// comment forbids widening that one to cover request-path traffic). Fed
|
||||||
|
// by failoverEligible's own classification (record's own doc comment
|
||||||
|
// explains why, not limiter.recordProviderAttempt's narrower isTransient/
|
||||||
|
// isDeadlineExceeded one), once per candidate per logical request, via
|
||||||
|
// record's own callers in routes_unified.go, but keeps its own
|
||||||
|
// independent state per provider name.
|
||||||
|
//
|
||||||
|
// Every method is nil-receiver-safe (both report "healthy"/no-op on a
|
||||||
|
// nil *requestHealthTracker), matching this package's own nil-means-
|
||||||
|
// disabled convention for an optional subsystem (responseCache's own doc
|
||||||
|
// comment, llmgateway.go's Gateway.cache field) — a Gateway assembled
|
||||||
|
// directly in a test without going through newGateway, or any future
|
||||||
|
// caller that does the same, degrades to "failover never skips a
|
||||||
|
// candidate for request health" rather than a nil-pointer panic.
|
||||||
|
type requestHealthTracker struct {
|
||||||
|
states map[string]*requestHealthState
|
||||||
|
nowFn func() time.Time
|
||||||
|
mu sync.Mutex
|
||||||
|
}
|
||||||
|
|
||||||
|
// newRequestHealthTracker returns an empty tracker — every provider name
|
||||||
|
// reads healthy until its first recorded failure.
|
||||||
|
func newRequestHealthTracker() *requestHealthTracker {
|
||||||
|
return &requestHealthTracker{states: make(map[string]*requestHealthState), nowFn: time.Now}
|
||||||
|
}
|
||||||
|
|
||||||
|
// stateFor returns provider's requestHealthState, creating it on first
|
||||||
|
// use.
|
||||||
|
func (t *requestHealthTracker) stateFor(provider string) *requestHealthState {
|
||||||
|
t.mu.Lock()
|
||||||
|
defer t.mu.Unlock()
|
||||||
|
st, ok := t.states[provider]
|
||||||
|
if !ok {
|
||||||
|
st = &requestHealthState{}
|
||||||
|
t.states[provider] = st
|
||||||
|
}
|
||||||
|
return st
|
||||||
|
}
|
||||||
|
|
||||||
|
// record accounts ONE LOGICAL REQUEST's outcome for provider — called
|
||||||
|
// exactly once per candidate per client request, after call() has
|
||||||
|
// already exhausted whatever same-provider retries retryPolicy.do
|
||||||
|
// performed internally (routes_unified.go's runMeteredCall), never once
|
||||||
|
// per raw upstream attempt. success=false must be
|
||||||
|
// !failoverEligible(callErr) (routes_unified.go/failover.go's own
|
||||||
|
// classification of the FINAL adapter-level error) — adversarial-review
|
||||||
|
// ruling (F3/F6): an EARLIER version fed this from isTransient(resp,
|
||||||
|
// err) || isDeadlineExceeded(err), limits.go/retry.go's own
|
||||||
|
// retry-shaped classification, once per RAW attempt inside
|
||||||
|
// retryPolicy.do. That was wrong on two counts. isTransient treats a
|
||||||
|
// non-429 4xx as "the provider answered correctly to a request it did
|
||||||
|
// not like" — correct for deciding whether retrying the SAME provider
|
||||||
|
// is worth it, wrong for deciding whether the POOL should route around
|
||||||
|
// it: a provider whose API key is dead answers 401 to every request and
|
||||||
|
// must eventually be skipped, not retried forever, which is exactly the
|
||||||
|
// "masking a broken provider key" risk this feature exists to reduce.
|
||||||
|
// And recording per raw attempt let ONE client request with
|
||||||
|
// retry.enabled (attempts: 3) report FOUR failures against a threshold
|
||||||
|
// documented as three REQUESTS. failoverEligible already classifies
|
||||||
|
// "everything except HTTP 400" as a failure, matching what this gate
|
||||||
|
// needs with no separate function required.
|
||||||
|
//
|
||||||
|
// requestBreakerFailureThreshold consecutive failures opens the breaker
|
||||||
|
// for requestBreakerOpenDuration. While open, healthy (below) reports
|
||||||
|
// unhealthy until that window passes; once it has, the NEXT record call
|
||||||
|
// for this provider — success or failure — is what healthy already
|
||||||
|
// exposed as a post-cooldown probe attempt: success closes the breaker
|
||||||
|
// outright, failure re-opens it with a doubled backoff (capped at
|
||||||
|
// requestBreakerMaxOpenDuration). This is the same probe-then-escalate-
|
||||||
|
// or-close shape providerState's discovery breaker uses (registry.go's
|
||||||
|
// recordHealthLocked/openBreakerLocked), reimplemented independently
|
||||||
|
// rather than shared — the two track different outcome streams, and
|
||||||
|
// discoveryHealthy's own doc comment forbids widening it to cover this
|
||||||
|
// one.
|
||||||
|
//
|
||||||
|
// A failure arriving while ALREADY open and still within its cooldown
|
||||||
|
// window is a deliberate no-op (adversarial-review fix, F5): an earlier
|
||||||
|
// version fell through to the closed-state consecutiveFailures++ branch
|
||||||
|
// for this case too, which reset backoff to the flat base duration and
|
||||||
|
// slid openUntil forward on every such failure instead of leaving the
|
||||||
|
// existing window alone — doubling never actually happened outside the
|
||||||
|
// narrow post-cooldown probe path the doc comment claimed was the only
|
||||||
|
// place it did.
|
||||||
|
//
|
||||||
|
// Deliberately no single-flight guard on the probe window (unlike
|
||||||
|
// registry.go's tryBeginRefresh inFlight flag for its own, much lower-
|
||||||
|
// volume, background-goroutine probe): this signal sits on the live
|
||||||
|
// request path, where many concurrent requests can observe the same
|
||||||
|
// post-cooldown window at once. Serializing them would need its own
|
||||||
|
// lock-and-CAS dance for a benefit — one broken provider getting at most
|
||||||
|
// one real probe instead of a short burst of them right as its backoff
|
||||||
|
// expires — this per-pod, best-effort signal does not need to pay for.
|
||||||
|
func (t *requestHealthTracker) record(provider string, success bool) {
|
||||||
|
if t == nil || provider == "" {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
st := t.stateFor(provider)
|
||||||
|
st.mu.Lock()
|
||||||
|
defer st.mu.Unlock()
|
||||||
|
now := t.nowFn()
|
||||||
|
|
||||||
|
if success {
|
||||||
|
st.open = false
|
||||||
|
st.consecutiveFailures = 0
|
||||||
|
st.backoff = 0
|
||||||
|
st.openUntil = time.Time{}
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// Already open, cooldown not yet elapsed: this failure changes
|
||||||
|
// nothing (F5 fix, above).
|
||||||
|
if st.open && now.Before(st.openUntil) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
if st.open {
|
||||||
|
// Cooldown elapsed: this failure IS the probe outcome. Escalate.
|
||||||
|
if st.backoff <= 0 {
|
||||||
|
st.backoff = requestBreakerOpenDuration
|
||||||
|
} else {
|
||||||
|
st.backoff *= 2
|
||||||
|
}
|
||||||
|
if st.backoff > requestBreakerMaxOpenDuration {
|
||||||
|
st.backoff = requestBreakerMaxOpenDuration
|
||||||
|
}
|
||||||
|
st.openUntil = now.Add(st.backoff)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
st.consecutiveFailures++
|
||||||
|
if st.consecutiveFailures >= requestBreakerFailureThreshold {
|
||||||
|
st.open = true
|
||||||
|
st.backoff = requestBreakerOpenDuration
|
||||||
|
st.openUntil = now.Add(st.backoff)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// healthy reports whether provider's request-path breaker currently
|
||||||
|
// permits routing to it — feat/failover's hard SKIP gate
|
||||||
|
// (orderedFailoverCandidates, above), distinct from
|
||||||
|
// modelRegistry.discoveryHealthy's deprioritize-only contract. An
|
||||||
|
// unrecorded provider (never failed, or never attempted at all)
|
||||||
|
// reads healthy — the same "never seen a failure" default the discovery
|
||||||
|
// breaker's own zero value already applies (providerState's doc
|
||||||
|
// comment, registry.go).
|
||||||
|
func (t *requestHealthTracker) healthy(provider string) bool {
|
||||||
|
if t == nil || provider == "" {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
st := t.stateFor(provider)
|
||||||
|
st.mu.Lock()
|
||||||
|
defer st.mu.Unlock()
|
||||||
|
if !st.open {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
return !t.nowFn().Before(st.openUntil)
|
||||||
|
}
|
||||||
+1963
File diff suppressed because it is too large
Load Diff
+49
-8
@@ -70,7 +70,13 @@ type Config struct {
|
|||||||
// at all) resolves to every documented default and behaves exactly
|
// at all) resolves to every documented default and behaves exactly
|
||||||
// like a deployment with no circuit breaker at all for a provider
|
// like a deployment with no circuit breaker at all for a provider
|
||||||
// that never fails — see validateBreakerConfig's own doc comment.
|
// that never fails — see validateBreakerConfig's own doc comment.
|
||||||
Breaker BreakerConfig `json:"breaker"`
|
// Failover configures feat/failover — cross-provider failover for a
|
||||||
|
// bare model id more than one configured provider serves. See
|
||||||
|
// FailoverConfig's own doc comment (failover.go) for the full
|
||||||
|
// contract; a struct value, not a pointer, for the same
|
||||||
|
// zero-means-default reason as Breaker/Retry/Cache below.
|
||||||
|
Failover FailoverConfig `json:"failover"`
|
||||||
|
Breaker BreakerConfig `json:"breaker"`
|
||||||
// Retry is a struct value, not a pointer, because its own Enabled
|
// Retry is a struct value, not a pointer, because its own Enabled
|
||||||
// field is the on/off signal (unlike Redis/Users, where the block's
|
// field is the on/off signal (unlike Redis/Users, where the block's
|
||||||
// mere presence is the signal) — so its tag omits "omitempty":
|
// mere presence is the signal) — so its tag omits "omitempty":
|
||||||
@@ -493,18 +499,24 @@ func defaultBodyAdmissionCap() int {
|
|||||||
const maxExplicitBodyAdmissionCap = 10_000
|
const maxExplicitBodyAdmissionCap = 10_000
|
||||||
|
|
||||||
// Gateway is the Traefik middleware handler.
|
// Gateway is the Traefik middleware handler.
|
||||||
|
//
|
||||||
|
// Field order below is fieldalignment-verified (golangci-lint's govet
|
||||||
|
// enable-all, run with -fix against a scratch copy to derive the exact
|
||||||
|
// zero-waste sequence, then hand-applied here so every field keeps its
|
||||||
|
// original doc comment) — see providerState's own doc comment
|
||||||
|
// (registry.go) for the general convention.
|
||||||
type Gateway struct {
|
type Gateway struct {
|
||||||
next http.Handler
|
next http.Handler
|
||||||
cfg *Config
|
|
||||||
auth *authStore
|
|
||||||
limiter *limiter
|
|
||||||
registry *modelRegistry
|
|
||||||
adapters map[string]providerAdapter
|
|
||||||
// targetClient is the shared, connection-pooled *http.Client the
|
// targetClient is the shared, connection-pooled *http.Client the
|
||||||
// MCP/A2A target proxy (mcp_a2a.go) issues every upstream request
|
// MCP/A2A target proxy (mcp_a2a.go) issues every upstream request
|
||||||
// through — built once via newAdapterHTTPClient, the same constructor
|
// through — built once via newAdapterHTTPClient, the same constructor
|
||||||
// each provider adapter uses for its own client.
|
// each provider adapter uses for its own client.
|
||||||
targetClient *http.Client
|
targetClient *http.Client
|
||||||
|
auth *authStore
|
||||||
|
limiter *limiter
|
||||||
|
registry *modelRegistry
|
||||||
|
adapters map[string]providerAdapter
|
||||||
|
cfg *Config
|
||||||
// cache is nil whenever response caching is not configured or not
|
// cache is nil whenever response caching is not configured or not
|
||||||
// usable (cfg.Cache.Enabled is false, or true with no config.Redis —
|
// usable (cfg.Cache.Enabled is false, or true with no config.Redis —
|
||||||
// see buildResponseCache, cache.go). Every call site checks for nil
|
// see buildResponseCache, cache.go). Every call site checks for nil
|
||||||
@@ -525,7 +537,25 @@ type Gateway struct {
|
|||||||
// resized afterward, matching a Go channel's own fixed-capacity
|
// resized afterward, matching a Go channel's own fixed-capacity
|
||||||
// contract.
|
// contract.
|
||||||
bodyAdmission chan struct{}
|
bodyAdmission chan struct{}
|
||||||
name string
|
// failoverHealth is feat/failover's own per-pod, in-memory
|
||||||
|
// request-path health signal (failover.go), built once, here, by
|
||||||
|
// newGateway. Every method on it is nil-receiver-safe, so a Gateway
|
||||||
|
// assembled directly (bypassing newGateway, as a few older tests do)
|
||||||
|
// degrades to "failover never skips a candidate for request health"
|
||||||
|
// rather than a nil-pointer panic.
|
||||||
|
failoverHealth *requestHealthTracker
|
||||||
|
name string
|
||||||
|
// failoverLogGate rate-limits runMeteredCall's generic "failing over"
|
||||||
|
// log line (routes_unified.go) to once per storeErrorLogEvery
|
||||||
|
// (adversarial-review fix, F10) — reuses auth.go's own logGate type,
|
||||||
|
// the identical technique shouldLogAuthFailure/logStoreError already
|
||||||
|
// apply elsewhere in this package. Zero-value-usable: the first call
|
||||||
|
// on a fresh Gateway always logs. Deliberately does NOT gate the 404
|
||||||
|
// loud-log line (operator ruling: that one must always log).
|
||||||
|
failoverLogGate logGate
|
||||||
|
// failover is Config.Failover, validated and resolved once by
|
||||||
|
// newGateway (validateFailoverConfig, failover.go).
|
||||||
|
failover failoverConfig
|
||||||
}
|
}
|
||||||
|
|
||||||
// telemetryStartupOnce keeps the anonymous "plugin loaded" ping to one per
|
// telemetryStartupOnce keeps the anonymous "plugin loaded" ping to one per
|
||||||
@@ -634,6 +664,17 @@ func newGateway(ctx context.Context, next http.Handler, config *Config, name str
|
|||||||
}
|
}
|
||||||
g := &Gateway{next: next, name: name, cfg: config, auth: auth}
|
g := &Gateway{next: next, name: name, cfg: config, auth: auth}
|
||||||
|
|
||||||
|
// feat/failover: validated once, here — see FailoverConfig's own doc
|
||||||
|
// comment (failover.go) for the default (disabled, coordinator
|
||||||
|
// ruling) and what "single-provider deployment behaves exactly as
|
||||||
|
// before" means in practice.
|
||||||
|
failoverCfg, err := validateFailoverConfig(config.Failover)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
g.failover = failoverCfg
|
||||||
|
g.failoverHealth = newRequestHealthTracker()
|
||||||
|
|
||||||
// bodyAdmissionCap: an explicit MaxInFlightBodyRequests always wins
|
// bodyAdmissionCap: an explicit MaxInFlightBodyRequests always wins
|
||||||
// over the self-tuned default (house rule: explicit override always
|
// over the self-tuned default (house rule: explicit override always
|
||||||
// wins) — see Config.MaxInFlightBodyRequests' own doc comment. 0 (the
|
// wins) — see Config.MaxInFlightBodyRequests' own doc comment. 0 (the
|
||||||
|
|||||||
+329
-102
@@ -5,10 +5,12 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"errors"
|
"errors"
|
||||||
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"math"
|
"math"
|
||||||
"net/http"
|
"net/http"
|
||||||
"strconv"
|
"strconv"
|
||||||
|
"time"
|
||||||
)
|
)
|
||||||
|
|
||||||
// maxRequestBytes caps a client request body decoded into a
|
// maxRequestBytes caps a client request body decoded into a
|
||||||
@@ -256,123 +258,348 @@ func (g *Gateway) runUnified(w http.ResponseWriter, r *http.Request, u *user, gr
|
|||||||
// envelope shape they answer in. logPrefix names the route in every log
|
// envelope shape they answer in. logPrefix names the route in every log
|
||||||
// line this function writes, so a log reader can always tell which route
|
// line this function writes, so a log reader can always tell which route
|
||||||
// produced a given line without the classification logic existing twice.
|
// produced a given line without the classification logic existing twice.
|
||||||
|
//
|
||||||
|
// feat/failover: this function now tries a LIST of candidate providers
|
||||||
|
// (g.registry.resolveWithCandidates/g.orderedFailoverCandidates,
|
||||||
|
// failover.go), not just the one resolve alone would pick, and is the
|
||||||
|
// ONE hook point both /v1/chat/completions (runUnified, above) and
|
||||||
|
// /v1/messages (handleMessages, routes_messages.go) share — per the
|
||||||
|
// brief's own instruction, failover is wired here once rather than
|
||||||
|
// forked into a second implementation. THE HARD CONSTRAINT this loop is
|
||||||
|
// built around: call(...) writes directly to respWriter, which tees
|
||||||
|
// straight through to sw; the very first byte written commits the
|
||||||
|
// response, so a candidate is only ever retried while sw.wroteHeader is
|
||||||
|
// still false. Every step that was a one-shot computation before this
|
||||||
|
// feature (the cache key, the attempt recorder's bound provider name,
|
||||||
|
// the usage/canonical id billed) now happens fresh INSIDE the loop, once
|
||||||
|
// per candidate — this is what "rebind the attempt recorder per
|
||||||
|
// attempt" and "cache under the winning provider's key" actually mean
|
||||||
|
// in code, not just in the brief's prose.
|
||||||
func (g *Gateway) runMeteredCall(sw *statusTrackingWriter, r *http.Request, scopes []limitScope, body []byte, req map[string]any, requestedModel string, grp *group, endpoint, logPrefix string, envelope envelopeWriter, writeUpstream providerUpstreamErrorWriter, call adapterCall) {
|
func (g *Gateway) runMeteredCall(sw *statusTrackingWriter, r *http.Request, scopes []limitScope, body []byte, req map[string]any, requestedModel string, grp *group, endpoint, logPrefix string, envelope envelopeWriter, writeUpstream providerUpstreamErrorWriter, call adapterCall) {
|
||||||
adapter, upstreamModel, canonical, err := g.registry.resolve(requestedModel, grp)
|
// F9 (adversarial-review, optional perf fix): a disabled deployment
|
||||||
|
// never needs the candidate pool at all — resolvePrimaryOnly skips
|
||||||
|
// failoverCandidates' own provider scan entirely, so a config that
|
||||||
|
// never opted into failover pays nothing extra for a pool it will
|
||||||
|
// never use.
|
||||||
|
var primary resolveCandidate
|
||||||
|
var extra []resolveCandidate
|
||||||
|
var err error
|
||||||
|
if g.failover.enabled {
|
||||||
|
primary, extra, err = g.registry.resolveWithCandidates(requestedModel, grp)
|
||||||
|
} else {
|
||||||
|
primary, err = g.registry.resolvePrimaryOnly(requestedModel, grp)
|
||||||
|
}
|
||||||
if err != nil {
|
if err != nil {
|
||||||
writeModelResolveErrorEnvelope(sw, err, envelope)
|
writeModelResolveErrorEnvelope(sw, err, envelope)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
// Sequence matters (each step's own doc comment explains why):
|
||||||
|
// health/discovery ordering first, so candidates[0] below is the
|
||||||
|
// ACTUAL first-attempted candidate the cost guard compares everyone
|
||||||
|
// else against; cost filtering second; the maxAttempts cap last, so
|
||||||
|
// it never re-admits a candidate the cost guard already removed.
|
||||||
|
candidates := g.orderedFailoverCandidates(primary, extra)
|
||||||
|
candidates = g.filterCandidatesByCost(candidates, requestedModel)
|
||||||
|
candidates = capFailoverCandidates(candidates, g.failover.maxAttempts)
|
||||||
|
|
||||||
req["model"] = upstreamModel
|
// cacheable/streaming gate every cache step below on spec §2's scope,
|
||||||
|
// exactly as before this feature: a non-streaming request, with a
|
||||||
// cacheable gates every cache step below on spec §2's scope: a
|
// working cache (g.cache is nil whenever caching is off or Redis was
|
||||||
// non-streaming request, with a working cache (g.cache is nil
|
// absent — buildResponseCache, cache.go), for a group that has not
|
||||||
// whenever caching is off or Redis was absent — buildResponseCache,
|
// opted out (groupCacheEnabled). Computed once, outside the loop: it
|
||||||
// cache.go), for a group that has not opted out (groupCacheEnabled).
|
// depends only on req["stream"]/grp, neither of which a failover
|
||||||
// cacheKey is computed here — after "model" is rewritten to
|
// attempt changes.
|
||||||
// upstreamModel but deliberately BEFORE gatewayAliasKey is injected
|
|
||||||
// below — so the hashed request body reflects exactly what goes
|
|
||||||
// upstream. requestedModel (the client's own, un-rewritten alias
|
|
||||||
// string) and endpoint are passed as separate cacheKey arguments,
|
|
||||||
// not left in req: a translating adapter bakes the echoed alias
|
|
||||||
// straight into the cached response body's own "model" field, so two
|
|
||||||
// clients requesting the same upstream model under different alias
|
|
||||||
// forms (e.g. "claude-x" vs "anthropic/claude-x") must never collide
|
|
||||||
// into one cache entry (cacheKey's own doc comment, cache.go).
|
|
||||||
streaming, _ := req["stream"].(bool)
|
streaming, _ := req["stream"].(bool)
|
||||||
cacheable := !streaming && g.cache != nil && groupCacheEnabled(grp)
|
cacheable := !streaming && g.cache != nil && groupCacheEnabled(grp)
|
||||||
var cacheKeyStr string
|
|
||||||
if cacheable {
|
// pristineReq is the client's own decoded request, snapshotted ONCE,
|
||||||
cacheKeyStr = cacheKey(adapter.name(), upstreamModel, requestedModel, endpoint, req)
|
// BEFORE any candidate's adapter ever touches it, with "model"
|
||||||
if cached, hit := g.cache.lookup(cacheKeyStr); hit {
|
// already rewritten to the upstream id every candidate in this
|
||||||
// A cache hit accounts the request only — checkAndCount
|
// failover chain shares by construction (failoverCandidates,
|
||||||
// above already counted it — never token/cost, and never
|
// failover.go, only ever adds providers serving the IDENTICAL bare
|
||||||
// runs the zero-usage estimate branch below: the client
|
// id resolve's primary already resolved to). Every attempt below
|
||||||
// never reached the upstream provider its body size would
|
// builds its OWN fresh top-level copy from this snapshot rather than
|
||||||
// be estimating against.
|
// reusing or mutating one shared map (adversarial-review fix, F1):
|
||||||
sw.Header().Set("X-Llmgw-Cache", "hit")
|
// openai-type adapters mutate req in place — chatCompletion forces
|
||||||
if cached.ContentType != "" {
|
// stream_options.include_usage=true onto it (provider_openai.go) —
|
||||||
sw.Header().Set("Content-Type", cached.ContentType)
|
// and reusing the same map across attempts let a LATER candidate
|
||||||
|
// read back an EARLIER candidate's own forced injection and
|
||||||
|
// misreport clientAskedUsage=true to a client that never asked for
|
||||||
|
// it, leaking the gateway's own suppressed usage-only SSE chunk.
|
||||||
|
// Proven against a single-provider control; this is what fixes it
|
||||||
|
// for openai, openai-audio, and copilot alike.
|
||||||
|
pristineReq := make(map[string]any, len(req)+1)
|
||||||
|
for k, v := range req {
|
||||||
|
pristineReq[k] = v
|
||||||
|
}
|
||||||
|
if len(candidates) > 0 {
|
||||||
|
pristineReq["model"] = candidates[0].upstreamModel
|
||||||
|
}
|
||||||
|
|
||||||
|
var lastErr error
|
||||||
|
var lastProviderName string
|
||||||
|
for i, cand := range candidates {
|
||||||
|
if i > 0 {
|
||||||
|
// F4 fix (adversarial-review): a shared request context an
|
||||||
|
// EARLIER candidate already exhausted must never be
|
||||||
|
// misattributed as THIS candidate's own failure. There is no
|
||||||
|
// per-candidate timeout budget in this architecture (every
|
||||||
|
// candidate shares r's SAME context — see README's own
|
||||||
|
// documented limitation), so once an earlier candidate burns
|
||||||
|
// the deadline, attempting the next one would fail near-
|
||||||
|
// instantly for a reason that has nothing to do with its own
|
||||||
|
// health. Stop here with the LAST REAL error instead of
|
||||||
|
// chasing further doomed candidates and wrongly opening a
|
||||||
|
// perfectly healthy provider's request-health breaker.
|
||||||
|
if r.Context().Err() != nil {
|
||||||
|
g.handleAdapterErrorEnvelope(sw, lastErr, lastProviderName, logPrefix, envelope, writeUpstream)
|
||||||
|
return
|
||||||
}
|
}
|
||||||
sw.WriteHeader(cached.Status)
|
|
||||||
_, _ = sw.Write(cached.Body)
|
// checkAndCount above (admitRequest, this file) already
|
||||||
|
// counted this request exactly once — never double-count: a
|
||||||
|
// failover attempt is a NEW upstream attempt, not a new
|
||||||
|
// logical request, so nothing here calls checkAndCount
|
||||||
|
// again. Per-provider ATTEMPT counters, below, still record
|
||||||
|
// every attempt independently via the freshly-bound
|
||||||
|
// recorder.
|
||||||
|
//
|
||||||
|
// Rate-limited (F10 fix, adversarial-review): during a
|
||||||
|
// sustained outage this line would otherwise be one
|
||||||
|
// blocking synchronous stderr write per request on shared
|
||||||
|
// Traefik ingress — the exact failure shouldLogAuthFailure/
|
||||||
|
// logStoreError (auth.go/limits.go) already exist to
|
||||||
|
// prevent elsewhere in this package. The 404 line below is
|
||||||
|
// deliberately NOT gated by this (operator ruling: it must
|
||||||
|
// always log).
|
||||||
|
if log, suppressed := g.failoverLogGate.shouldLog(time.Now()); log {
|
||||||
|
suffix := ""
|
||||||
|
if suppressed > 0 {
|
||||||
|
suffix = fmt.Sprintf(" (%d more failovers suppressed since last log)", suppressed)
|
||||||
|
}
|
||||||
|
g.warnf("%s: failing over from provider %q to %q for model %q%s", logPrefix, candidates[i-1].providerName, cand.providerName, requestedModel, suffix)
|
||||||
|
}
|
||||||
|
|
||||||
|
// F2 fix, correctly scoped (adversarial-review round 2 — the
|
||||||
|
// round-1 version cleared unconditionally, including i == 0,
|
||||||
|
// which deleted response headers a DIFFERENT, earlier Traefik
|
||||||
|
// middleware had already staged before this plugin ever ran
|
||||||
|
// (CORS, HSTS, request-id, ...) on the default, non-failover
|
||||||
|
// path — proven by differential harness, and it contradicted
|
||||||
|
// this package's own "byte-identical to a gateway built
|
||||||
|
// before this field existed" claim. Only i > 0 needs this at
|
||||||
|
// all: newSSEWriter (sse.go) sets Content-Type: text/event-
|
||||||
|
// stream, Cache-Control: no-cache, and X-Accel-Buffering: no
|
||||||
|
// the moment a streaming attempt starts, before any byte is
|
||||||
|
// written, and if THAT PRIOR candidate then failed before
|
||||||
|
// writing (still safe to fail over, THE HARD CONSTRAINT
|
||||||
|
// below), those staged headers must not survive into a
|
||||||
|
// plain-JSON response from THIS candidate. Safe to clear
|
||||||
|
// unconditionally once i > 0: this loop only ever reaches a
|
||||||
|
// new iteration while sw.wroteHeader is false, so nothing
|
||||||
|
// committed is ever at risk of being wiped.
|
||||||
|
hdr := sw.Header()
|
||||||
|
for k := range hdr {
|
||||||
|
hdr.Del(k)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// attemptReq is a FRESH top-level copy of pristineReq for every
|
||||||
|
// candidate (F1 fix, this function's own doc comment above).
|
||||||
|
// gatewayAliasKey is (re-)injected here: every providerAdapter
|
||||||
|
// implementation unconditionally deletes it before it ever
|
||||||
|
// touches the network (its own doc comment, providers.go), so
|
||||||
|
// each fresh copy needs it set again to echo the client's own
|
||||||
|
// requested alias correctly (anthropic/gemini-type adapters read
|
||||||
|
// it to build their own response envelope).
|
||||||
|
attemptReq := make(map[string]any, len(pristineReq)+1)
|
||||||
|
for k, v := range pristineReq {
|
||||||
|
attemptReq[k] = v
|
||||||
|
}
|
||||||
|
attemptReq[gatewayAliasKey] = requestedModel
|
||||||
|
|
||||||
|
// Cache lookup for THIS candidate, immediately before its own
|
||||||
|
// attempt (adversarial-review round 2 ruling — supersedes the
|
||||||
|
// round-1 "check every candidate's cache up front" pre-pass,
|
||||||
|
// which silently substituted ANOTHER candidate's cached response
|
||||||
|
// while the actual primary was perfectly healthy: proven —
|
||||||
|
// alpha healthy and serving a distinct body, never called, the
|
||||||
|
// client instead got beta's cached body with no failure and no
|
||||||
|
// "failing over" log line, i.e. failover with no failure. Health
|
||||||
|
// (F3/F6, this file/failover.go) is the correct mechanism for
|
||||||
|
// "do not call a dead primary": once a genuinely broken
|
||||||
|
// candidate is marked request-unhealthy it is excluded from the
|
||||||
|
// candidate list entirely (orderedFailoverCandidates,
|
||||||
|
// failover.go), so the NEXT candidate's cache is still reached
|
||||||
|
// on a later request without ever dialing the dead one — the
|
||||||
|
// pre-pass solved that problem by accident, with the side
|
||||||
|
// effect of mis-serving healthy traffic from an unrelated
|
||||||
|
// candidate's cache. cacheKeyStr is computed from pristineReq,
|
||||||
|
// BEFORE gatewayAliasKey was injected above would have mattered
|
||||||
|
// — gatewayAliasKey is a purely internal echo-back mechanism
|
||||||
|
// cacheKey does not itself strip (its own doc comment,
|
||||||
|
// cache.go), so hashing attemptReq instead would add noise no
|
||||||
|
// two requests would otherwise disagree on.
|
||||||
|
var cacheKeyStr string
|
||||||
|
if cacheable {
|
||||||
|
cacheKeyStr = cacheKey(cand.providerName, cand.upstreamModel, requestedModel, endpoint, pristineReq)
|
||||||
|
if cached, hit := g.cache.lookup(cacheKeyStr); hit {
|
||||||
|
sw.Header().Set("X-Llmgw-Cache", "hit")
|
||||||
|
if cached.ContentType != "" {
|
||||||
|
sw.Header().Set("Content-Type", cached.ContentType)
|
||||||
|
}
|
||||||
|
sw.WriteHeader(cached.Status)
|
||||||
|
_, _ = sw.Write(cached.Body)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
// Set before call() below writes anything — headers must
|
||||||
|
// precede the body a miss is about to produce (spec §2). A
|
||||||
|
// later iteration's own miss overwrites this harmlessly: it
|
||||||
|
// is never observable until SOME candidate actually writes,
|
||||||
|
// by which point the loop has already committed to that
|
||||||
|
// candidate's outcome (THE HARD CONSTRAINT, this function's
|
||||||
|
// own doc comment).
|
||||||
|
sw.Header().Set("X-Llmgw-Cache", "miss")
|
||||||
|
}
|
||||||
|
|
||||||
|
// respWriter is sw, wrapped in a capture tee only when this
|
||||||
|
// request is cacheable — cacheCaptureWriter buffers everything
|
||||||
|
// written so a 200 non-stream response can be stored after
|
||||||
|
// call() returns, without any adapter knowing caching exists
|
||||||
|
// (adapterCall's contract is unchanged either way). A fresh
|
||||||
|
// capture per candidate: reusing one across attempts would let a
|
||||||
|
// candidate that never wrote anything (the common failover case)
|
||||||
|
// leave a stale buffer behind — capture.status stays 0 either
|
||||||
|
// way, so this is defensive clarity, not a proven bug fix.
|
||||||
|
var respWriter http.ResponseWriter = sw
|
||||||
|
var capture *cacheCaptureWriter
|
||||||
|
if cacheable {
|
||||||
|
capture = newCacheCaptureWriter(sw, g.cache.maxBodyBytes)
|
||||||
|
respWriter = capture
|
||||||
|
}
|
||||||
|
|
||||||
|
// Feature A (v0.22) attempt recorder, REBOUND every iteration to
|
||||||
|
// THIS candidate's own provider name (feat/failover's own "never
|
||||||
|
// misattribute a retried attempt to the first provider"
|
||||||
|
// requirement) — every upstream attempt call() makes, via
|
||||||
|
// upstreamJSON's retryPolicy.do (providers.go), reports through
|
||||||
|
// here. limiter.recordProviderAttempt keeps its own existing
|
||||||
|
// per-raw-attempt classification unchanged (limits.go); feat/
|
||||||
|
// failover's OWN request-path health gate is recorded separately
|
||||||
|
// below, once per candidate rather than once per raw attempt —
|
||||||
|
// see requestHealthTracker.record's own doc comment (failover.go)
|
||||||
|
// for why (F3/F6 fix, adversarial-review).
|
||||||
|
providerName := cand.providerName
|
||||||
|
upstreamModel := cand.upstreamModel
|
||||||
|
ctx := withAttemptRecorder(r.Context(), func(resp *http.Response, attemptErr error) {
|
||||||
|
g.limiter.recordProviderAttempt(providerName, upstreamModel, resp, attemptErr)
|
||||||
|
})
|
||||||
|
result, callErr := call(cand.adapter, ctx, respWriter, attemptReq)
|
||||||
|
|
||||||
|
// Request-path health, recorded ONCE per candidate per logical
|
||||||
|
// request (F3/F6 fix, adversarial-review) — see
|
||||||
|
// requestHealthTracker.record's own doc comment (failover.go)
|
||||||
|
// for the full ruling: failoverEligible's classification
|
||||||
|
// (everything except HTTP 400 is a failure) is reused directly,
|
||||||
|
// not limiter's narrower isTransient/isDeadlineExceeded one, so
|
||||||
|
// a provider answering 401/403/404 to every request is
|
||||||
|
// eventually routed around instead of retried forever. Guarded
|
||||||
|
// by g.failover.enabled (adversarial-review round 2, finding 3):
|
||||||
|
// this state is never READ on the disabled path
|
||||||
|
// (orderedFailoverCandidates' own early return never calls
|
||||||
|
// healthy()), so recording it there was dead work plus a new
|
||||||
|
// global-mutex contention point (requestHealthTracker.stateFor's
|
||||||
|
// own t.mu) on EVERY metered request, disabled or not.
|
||||||
|
if g.failover.enabled {
|
||||||
|
g.failoverHealth.record(providerName, !failoverEligible(callErr))
|
||||||
|
}
|
||||||
|
|
||||||
|
// Usage is accounted before the error branch below runs, not
|
||||||
|
// after: every adapter that can fail mid-stream (forwardStream
|
||||||
|
// in each of the three provider files) still returns whatever
|
||||||
|
// usage it had already captured alongside the error — a usage
|
||||||
|
// chunk that arrived just before a dropped connection must
|
||||||
|
// still be billed, or a client that aborts right after that
|
||||||
|
// frame arrives could repeat the trick to dodge every
|
||||||
|
// token/cost budget. A non-streaming failure (providerHTTPError,
|
||||||
|
// translateError, or a connection failure before any write)
|
||||||
|
// always carries zero usage by contract, so accounting it here
|
||||||
|
// is a no-op — account skips every write once both total tokens
|
||||||
|
// and cost are zero. routes_messages.go's
|
||||||
|
// callTranslatedMessages/callAnthropicMessagesPassthrough follow
|
||||||
|
// the identical contract for their own error paths (item 1 fix,
|
||||||
|
// 2026-08-22 review). This still runs exactly once per
|
||||||
|
// CANDIDATE, including a candidate that goes on to fail over —
|
||||||
|
// its own (zero, by contract) usage is billed as a no-op, never
|
||||||
|
// skipped, so a later successful candidate's real usage is never
|
||||||
|
// silently doubled with a phantom first entry either.
|
||||||
|
if callErr == nil && result.total() == 0 {
|
||||||
|
if streaming {
|
||||||
|
g.logf("%s: zero usage reported for a streaming response from model %q; accounting the request only", logPrefix, cand.canonical)
|
||||||
|
} else {
|
||||||
|
result.prompt = int64(math.Ceil(float64(len(body)) / 4))
|
||||||
|
result.estimated = true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
g.limiter.account(scopes, result, unifiedCostMicros(cand.canonical, cand.upstreamModel, result, g.cfg.Pricing))
|
||||||
|
if result.estimated {
|
||||||
|
g.logf("%s: usage for model %q logged as estimated (%d prompt tokens derived from request body size, not the provider's reported usage)", logPrefix, cand.canonical, result.prompt)
|
||||||
|
}
|
||||||
|
|
||||||
|
// A miss stores the response after everything above has already
|
||||||
|
// run — accounting must never be skipped or delayed waiting on a
|
||||||
|
// cache write. Only a genuine upstream 200 is stored (spec §2's
|
||||||
|
// "on 200 non-stream, tee and SET"): a non-2xx status never
|
||||||
|
// reaches here as capture.status (forwardJSON/translate error
|
||||||
|
// paths return *providerHTTPError/*translateError instead of
|
||||||
|
// writing through respWriter, so callErr is non-nil and
|
||||||
|
// capture.status stays 0), and callErr == nil is checked
|
||||||
|
// directly regardless. capture.oversize excludes a response
|
||||||
|
// cacheCaptureWriter stopped buffering past maxBodyBytes:
|
||||||
|
// store's own maxBodyBytes check would reject it too, but only
|
||||||
|
// after being handed a silently truncated body — skip the call
|
||||||
|
// outright instead of ever constructing a corrupt cache entry.
|
||||||
|
if cacheable && callErr == nil && capture.status == http.StatusOK && !capture.oversize {
|
||||||
|
g.cache.store(cacheKeyStr, capture.status, capture.contentType, capture.buf.Bytes(), effectiveTTL(g.cache, grp))
|
||||||
|
}
|
||||||
|
|
||||||
|
if callErr == nil {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
// Set before call() below writes anything — headers must precede
|
|
||||||
// the body a miss is about to produce (spec §2).
|
|
||||||
sw.Header().Set("X-Llmgw-Cache", "miss")
|
|
||||||
}
|
|
||||||
|
|
||||||
req[gatewayAliasKey] = requestedModel
|
// THE HARD CONSTRAINT: once sw has written anything, status and
|
||||||
|
// body are committed and can never be retried elsewhere — a
|
||||||
// respWriter is sw, wrapped in a capture tee only when this request
|
// mid-stream failure (headers, or a partial SSE body, already
|
||||||
// is cacheable — cacheCaptureWriter buffers everything written so a
|
// reached the client) always falls straight through to the
|
||||||
// 200 non-stream response can be stored after call() returns, without
|
// existing, unchanged error classification, exactly as it did
|
||||||
// any adapter knowing caching exists (adapterCall's contract is
|
// before this feature existed, regardless of how many candidates
|
||||||
// unchanged either way).
|
// remain.
|
||||||
var respWriter http.ResponseWriter = sw
|
if sw.wroteHeader {
|
||||||
var capture *cacheCaptureWriter
|
g.handleAdapterErrorEnvelope(sw, callErr, providerName, logPrefix, envelope, writeUpstream)
|
||||||
if cacheable {
|
return
|
||||||
capture = newCacheCaptureWriter(sw, g.cache.maxBodyBytes)
|
|
||||||
respWriter = capture
|
|
||||||
}
|
|
||||||
|
|
||||||
// Feature A (v0.22): every upstream attempt call() makes — via
|
|
||||||
// upstreamJSON's retryPolicy.do, providers.go — reports through to
|
|
||||||
// this recorder, closing over the adapter/model this request already
|
|
||||||
// resolved to. See attemptRecorder's own doc comment (providers.go).
|
|
||||||
ctx := withAttemptRecorder(r.Context(), func(resp *http.Response, attemptErr error) {
|
|
||||||
g.limiter.recordProviderAttempt(adapter.name(), upstreamModel, resp, attemptErr)
|
|
||||||
})
|
|
||||||
result, callErr := call(adapter, ctx, respWriter, req)
|
|
||||||
|
|
||||||
// Usage is accounted before the error branch below runs, not after:
|
|
||||||
// every adapter that can fail mid-stream (forwardStream in each of the
|
|
||||||
// three provider files) still returns whatever usage it had already
|
|
||||||
// captured alongside the error — a usage chunk that arrived just before
|
|
||||||
// a dropped connection must still be billed, or a client that aborts
|
|
||||||
// right after that frame arrives could repeat the trick to dodge every
|
|
||||||
// token/cost budget. A non-streaming failure (providerHTTPError,
|
|
||||||
// translateError, or a connection failure before any write) always
|
|
||||||
// carries zero usage by contract, so accounting it here is a no-op —
|
|
||||||
// account skips every write once both total tokens and cost are zero.
|
|
||||||
// routes_messages.go's callTranslatedMessages/
|
|
||||||
// callAnthropicMessagesPassthrough follow the identical contract for
|
|
||||||
// their own error paths (item 1 fix, 2026-08-22 review: a translation
|
|
||||||
// failure AFTER a successful, billed upstream call must still return
|
|
||||||
// that call's real usage, not usage{}).
|
|
||||||
if callErr == nil && result.total() == 0 {
|
|
||||||
if streaming {
|
|
||||||
g.logf("%s: zero usage reported for a streaming response from model %q; accounting the request only", logPrefix, canonical)
|
|
||||||
} else {
|
|
||||||
result.prompt = int64(math.Ceil(float64(len(body)) / 4))
|
|
||||||
result.estimated = true
|
|
||||||
}
|
}
|
||||||
}
|
|
||||||
g.limiter.account(scopes, result, unifiedCostMicros(canonical, upstreamModel, result, g.cfg.Pricing))
|
|
||||||
if result.estimated {
|
|
||||||
g.logf("%s: usage for model %q logged as estimated (%d prompt tokens derived from request body size, not the provider's reported usage)", logPrefix, canonical, result.prompt)
|
|
||||||
}
|
|
||||||
|
|
||||||
// A miss stores the response after everything above has already run —
|
lastErr = callErr
|
||||||
// accounting must never be skipped or delayed waiting on a cache
|
lastProviderName = providerName
|
||||||
// write. Only a genuine upstream 200 is stored (spec §2's "on 200
|
|
||||||
// non-stream, tee and SET"): a non-2xx status never reaches here as
|
|
||||||
// capture.status (forwardJSON/translate error paths return
|
|
||||||
// *providerHTTPError/*translateError instead of writing through
|
|
||||||
// respWriter, so callErr is non-nil and capture.status stays 0), and
|
|
||||||
// callErr == nil is checked directly regardless. capture.oversize
|
|
||||||
// excludes a response cacheCaptureWriter stopped buffering past
|
|
||||||
// maxBodyBytes: store's own maxBodyBytes check would reject it too,
|
|
||||||
// but only after being handed a silently truncated body — skip the
|
|
||||||
// call outright instead of ever constructing a corrupt cache entry.
|
|
||||||
if cacheable && callErr == nil && capture.status == http.StatusOK && !capture.oversize {
|
|
||||||
g.cache.store(cacheKeyStr, capture.status, capture.contentType, capture.buf.Bytes(), effectiveTTL(g.cache, grp))
|
|
||||||
}
|
|
||||||
|
|
||||||
if callErr != nil {
|
hasNext := i < len(candidates)-1 && failoverEligible(callErr)
|
||||||
g.handleAdapterErrorEnvelope(sw, callErr, adapter.name(), logPrefix, envelope, writeUpstream)
|
if hasNext && isProviderNotFoundError(callErr) {
|
||||||
return
|
// Loud, non-suppressed log line (brief: "Failover after a 404
|
||||||
|
// must log loudly ... a 404 usually means a real
|
||||||
|
// configuration mistake, and silently succeeding elsewhere
|
||||||
|
// would hide it") — deliberately g.errorf, not g.logf, and
|
||||||
|
// deliberately its own line rather than folded into the
|
||||||
|
// generic "failing over" line above the next iteration
|
||||||
|
// already prints, and deliberately NOT gated by
|
||||||
|
// failoverLogGate (operator ruling: this one must always
|
||||||
|
// log).
|
||||||
|
g.errorf("%s: provider %q returned 404 for model %q; failing over to %q — this usually means a real provider/model configuration mistake, not a transient outage", logPrefix, providerName, requestedModel, candidates[i+1].providerName)
|
||||||
|
}
|
||||||
|
if !hasNext {
|
||||||
|
g.handleAdapterErrorEnvelope(sw, callErr, providerName, logPrefix, envelope, writeUpstream)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
// Falls through to the next candidate.
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -346,7 +346,37 @@ func run() error {
|
|||||||
}))
|
}))
|
||||||
defer brkUpstream.Close()
|
defer brkUpstream.Close()
|
||||||
|
|
||||||
attemptAccountingOverride := `{"breaker":{"failureThreshold":2,"openDuration":"3s","maxOpenDuration":"6s"},` +
|
// failoverAUpstream/failoverBUpstream back exerciseFailover (feat/
|
||||||
|
// failover, below): two providers configured with the IDENTICAL bare
|
||||||
|
// model id (failoverModelID) — the exact shape registry.go's bareWinner
|
||||||
|
// collision handles, and the shape production logs show for whisper-1/
|
||||||
|
// tts-1/tts-1-hd/the embeddings models across openai/openai-audio/
|
||||||
|
// copilot. A always answers 500; B always answers 200. This proves the
|
||||||
|
// failover candidate loop (runMeteredCall, routes_unified.go) falls
|
||||||
|
// through and serves correctly under the REAL interpreter, not merely
|
||||||
|
// compiled — this package's own doc comment is exactly why that
|
||||||
|
// distinction matters on this codebase.
|
||||||
|
failoverAUpstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
w.Header().Set("Content-Type", "application/json")
|
||||||
|
w.WriteHeader(http.StatusInternalServerError)
|
||||||
|
_, _ = w.Write([]byte(`{"error":{"message":"failover-a is down","type":"server_error"}}`))
|
||||||
|
}))
|
||||||
|
defer failoverAUpstream.Close()
|
||||||
|
failoverBHits := new(int64)
|
||||||
|
failoverBUpstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
atomic.AddInt64(failoverBHits, 1)
|
||||||
|
w.Header().Set("Content-Type", "application/json")
|
||||||
|
w.WriteHeader(http.StatusOK)
|
||||||
|
_, _ = w.Write([]byte(`{"id":"c-failover","object":"chat.completion","model":"` + failoverModelID + `","choices":[{"index":0,"message":{"role":"assistant","content":"served by failover-b"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1}}`))
|
||||||
|
}))
|
||||||
|
defer failoverBUpstream.Close()
|
||||||
|
|
||||||
|
// failover.enabled must be set explicitly here: FailoverConfig.Enabled
|
||||||
|
// defaults to false (coordinator ruling — a version upgrade with no
|
||||||
|
// config change must preserve prior behavior), so without this the
|
||||||
|
// exerciseFailover probe below would find failover-a's 500 returned
|
||||||
|
// to the client unchanged, never reaching failover-b at all.
|
||||||
|
attemptAccountingOverride := `{"failover":{"enabled":true},"breaker":{"failureThreshold":2,"openDuration":"3s","maxOpenDuration":"6s"},` +
|
||||||
`"providers":{"openai":{"type":"openai","baseUrl":"` + upstream.URL + `","apiKey":"sk-up","models":["` + testDataWantModel + `","` + builtinLookupModelID + `"]},` +
|
`"providers":{"openai":{"type":"openai","baseUrl":"` + upstream.URL + `","apiKey":"sk-up","models":["` + testDataWantModel + `","` + builtinLookupModelID + `"]},` +
|
||||||
`"brk":{"type":"openai","baseUrl":"` + brkUpstream.URL + `","apiKey":"sk-up","discovery":true,"discoveryInterval":"1ms"},"` +
|
`"brk":{"type":"openai","baseUrl":"` + brkUpstream.URL + `","apiKey":"sk-up","discovery":true,"discoveryInterval":"1ms"},"` +
|
||||||
slowProviderName + `":{"type":"openai","baseUrl":"` + slowUpstream.URL + `","apiKey":"sk-up","models":["` + slowProviderModel + `"]},` +
|
slowProviderName + `":{"type":"openai","baseUrl":"` + slowUpstream.URL + `","apiKey":"sk-up","models":["` + slowProviderModel + `"]},` +
|
||||||
@@ -354,7 +384,9 @@ func run() error {
|
|||||||
// (exerciseMessagesRoute, below): a real anthropic-type provider,
|
// (exerciseMessagesRoute, below): a real anthropic-type provider,
|
||||||
// interpreted end to end through
|
// interpreted end to end through
|
||||||
// callAnthropicMessagesPassthrough.
|
// callAnthropicMessagesPassthrough.
|
||||||
`"anthropic":{"type":"anthropic","baseUrl":"` + anthropicProbeUpstream.URL + `","apiKey":"sk-anth","models":["claude-test"]}},` + // #nosec G101 -- test fixture literal, not a real credential
|
`"anthropic":{"type":"anthropic","baseUrl":"` + anthropicProbeUpstream.URL + `","apiKey":"sk-anth","models":["claude-test"]},` + // #nosec G101 -- test fixture literal, not a real credential
|
||||||
|
`"failover-a":{"type":"openai","baseUrl":"` + failoverAUpstream.URL + `","apiKey":"sk-up","models":["` + failoverModelID + `"]},` +
|
||||||
|
`"failover-b":{"type":"openai","baseUrl":"` + failoverBUpstream.URL + `","apiKey":"sk-up","models":["` + failoverModelID + `"]}},` +
|
||||||
// modelMeta (feature v0.23): a config-override entry for
|
// modelMeta (feature v0.23): a config-override entry for
|
||||||
// testDataWantModel, so exerciseHandler's GET /v1/models
|
// testDataWantModel, so exerciseHandler's GET /v1/models
|
||||||
// assertion below proves resolveModelMeta's config-override
|
// assertion below proves resolveModelMeta's config-override
|
||||||
@@ -405,7 +437,7 @@ func run() error {
|
|||||||
return fmt.Errorf("%s.New's returned value does not implement http.Handler", pkgName)
|
return fmt.Errorf("%s.New's returned value does not implement http.Handler", pkgName)
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := exerciseHandler(handler, builtinLookupContextTokens); err != nil {
|
if err := exerciseHandler(handler, builtinLookupContextTokens, failoverBHits); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
// exerciseBreaker (feat/provider-health, adversarial-review round 2):
|
// exerciseBreaker (feat/provider-health, adversarial-review round 2):
|
||||||
@@ -429,6 +461,11 @@ func run() error {
|
|||||||
// the config-override path testDataWantModel already exercises.
|
// the config-override path testDataWantModel already exercises.
|
||||||
const builtinLookupModelID = "gpt-4o"
|
const builtinLookupModelID = "gpt-4o"
|
||||||
|
|
||||||
|
// failoverModelID is the bare model id both "failover-a" and "failover-b"
|
||||||
|
// (run(), above) are configured with — the run()-owned upstream servers
|
||||||
|
// exerciseFailover (below) drives.
|
||||||
|
const failoverModelID = "failover-shared"
|
||||||
|
|
||||||
// mcpProbeServerName is the federated MCP server run() configures against
|
// mcpProbeServerName is the federated MCP server run() configures against
|
||||||
// mcpProbeUpstream, and mcpProbeToolName is a tool id carrying its
|
// mcpProbeUpstream, and mcpProbeToolName is a tool id carrying its
|
||||||
// "<server>_" prefix so resolveFederatedTool routes POST /mcp's
|
// "<server>_" prefix so resolveFederatedTool routes POST /mcp's
|
||||||
@@ -634,8 +671,8 @@ func exerciseBreaker(handler http.Handler, hits, healthy *int64) error {
|
|||||||
// for real under Yaegi, not merely that the package's imports resolve.
|
// for real under Yaegi, not merely that the package's imports resolve.
|
||||||
// builtinContextTokens is builtinLookupModelID's own real, extracted
|
// builtinContextTokens is builtinLookupModelID's own real, extracted
|
||||||
// builtinModelMetaTable value (SHOULD-6) — see that constant's own doc
|
// builtinModelMetaTable value (SHOULD-6) — see that constant's own doc
|
||||||
// comment.
|
// comment. failoverBHits backs exerciseFailover's own call, below.
|
||||||
func exerciseHandler(handler http.Handler, builtinContextTokens int) error {
|
func exerciseHandler(handler http.Handler, builtinContextTokens int, failoverBHits *int64) error {
|
||||||
authedReq := httptest.NewRequest(http.MethodGet, "/v1/models", nil)
|
authedReq := httptest.NewRequest(http.MethodGet, "/v1/models", nil)
|
||||||
authedReq.Header.Set("Authorization", "Bearer "+testDataUserAPIKey)
|
authedReq.Header.Set("Authorization", "Bearer "+testDataUserAPIKey)
|
||||||
authedRec := httptest.NewRecorder()
|
authedRec := httptest.NewRecorder()
|
||||||
@@ -702,6 +739,10 @@ func exerciseHandler(handler http.Handler, builtinContextTokens int) error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if err := exerciseFailover(handler, failoverBHits); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
if err := exerciseFederatedTooLarge(handler); err != nil {
|
if err := exerciseFederatedTooLarge(handler); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -1013,6 +1054,38 @@ func exerciseFederatedTooLarge(handler http.Handler) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// exerciseFailover drives one real POST /v1/chat/completions against the
|
||||||
|
// interpreted handler for failoverModelID — a bare model id both
|
||||||
|
// "failover-a" (always 500s) and "failover-b" (always 200s) are
|
||||||
|
// configured with — proving the failover candidate loop (runMeteredCall,
|
||||||
|
// routes_unified.go/failover.go) falls through a genuine provider
|
||||||
|
// failure and serves the second provider's response under the REAL
|
||||||
|
// interpreter. Compiled tests already prove this (failover_test.go); this
|
||||||
|
// harness exists because compiled tests have repeatedly passed on this
|
||||||
|
// codebase while the interpreted shape failed (this package's own doc
|
||||||
|
// comment).
|
||||||
|
func exerciseFailover(handler http.Handler, bHits *int64) error {
|
||||||
|
req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(
|
||||||
|
`{"model":"`+failoverModelID+`","messages":[{"role":"user","content":"hi"}]}`,
|
||||||
|
))
|
||||||
|
req.Header.Set("Authorization", "Bearer "+testDataUserAPIKey)
|
||||||
|
req.Header.Set("Content-Type", "application/json")
|
||||||
|
rec := httptest.NewRecorder()
|
||||||
|
handler.ServeHTTP(rec, req)
|
||||||
|
|
||||||
|
if rec.Code != http.StatusOK {
|
||||||
|
return fmt.Errorf("POST /v1/chat/completions (feat/failover harness): status = %d, want 200 (failover-a's 500 must fall through to failover-b), body=%s", rec.Code, rec.Body.String())
|
||||||
|
}
|
||||||
|
if !strings.Contains(rec.Body.String(), "served by failover-b") {
|
||||||
|
return fmt.Errorf("POST /v1/chat/completions (feat/failover harness): body does not contain failover-b's own content, want the SECOND provider's response: %s", rec.Body.String())
|
||||||
|
}
|
||||||
|
if atomic.LoadInt64(bHits) != 1 {
|
||||||
|
return fmt.Errorf("feat/failover harness: failover-b upstream hit count = %d, want exactly 1", atomic.LoadInt64(bHits))
|
||||||
|
}
|
||||||
|
fmt.Println("yaegi-check: failover fell through from failover-a to failover-b and served its response")
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
func exerciseAttemptAccounting(handler http.Handler) error {
|
func exerciseAttemptAccounting(handler http.Handler) error {
|
||||||
chatReq := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(
|
chatReq := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(
|
||||||
`{"model":"`+testDataWantModel+`","messages":[{"role":"user","content":"hi"}]}`,
|
`{"model":"`+testDataWantModel+`","messages":[{"role":"user","content":"hi"}]}`,
|
||||||
|
|||||||
Reference in New Issue
Block a user