lukaszraczylo/traefik-llmgateway
1
0
Fork 0
mirror of https://github.com/lukaszraczylo/traefik-llmgateway.git synced 2026-10-14 05:18:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(registry): add per-provider discovery circuit breaker

Browse Source 
Add a three-state circuit breaker to providerState: closed, open, and
half-open. It is driven only by discovery-refresh outcomes, never by
request-path traffic.

A provider whose refresh fails breakerThreshold times in a row opens
the breaker and backs off before its next half-open probe, instead of
retrying every discovery interval forever. A failed probe reopens the
breaker with a doubled backoff, capped at breakerOpenMax. A successful
probe closes the breaker again, so a recovered provider never stays
dead.

The failure classification reuses limits.go's isTransient and
isDeadlineExceeded, the same rule recordProviderAttempt already
applies to request-path attempts. It does not add a second
classification.

Add Config.Breaker (BreakerConfig), validated and defaulted by
validateBreakerConfig. Every default preserves current behavior for a
provider that never fails: a closed breaker never changes the existing
interval-based refresh gating.

Export modelRegistry.providerUsable(name) as the queryable health
predicate. A sibling branch (feat/failover) will use it to skip
unusable providers before routing. This commit does not touch the
request path or routes_unified.go.
This commit is contained in:
lukaszraczylo
2026-08-22 22:13:52 +01:00
parent 9dc67dedbb
commit e9ce6b873d
4 changed files with 771 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
llmgateway.go
+58 -12
View File
@@ -18,19 +18,23 @@ import (
// Config is the plugin's dynamic configuration, populated by Traefik from
// the middleware's YAML/testData. Every limit or list field follows the
// convention: zero/empty/omitted means unlimited/all.
// 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 pointer-first/scalar-last convention this follows.
type Config struct {
Providers map[string]*ProviderConfig `json:"providers,omitempty"`
Groups map[string]*GroupConfig `json:"groups,omitempty"`
Pricing map[string]*ModelPricing `json:"pricing,omitempty"`
MCPServers map[string]*TargetConfig `json:"mcpServers,omitempty"`
Agents map[string]*AgentConfig `json:"agents,omitempty"`
Users *UsersConfig `json:"users,omitempty"`
Redis *RedisConfig `json:"redis,omitempty"`
Redis *RedisConfig `json:"redis,omitempty"`
// Admin gates the read-only admin dashboard (spec §4, v0.2): nil or
// Admin.Enabled false means the /admin* routes are not registered at
// all — ServeHTTP falls through to its existing 404/passthroughUnknown
// handling for those paths, preserving v0.1 behavior exactly.
Admin *AdminConfig `json:"admin,omitempty"`
Admin *AdminConfig `json:"admin,omitempty"`
Pricing map[string]*ModelPricing `json:"pricing,omitempty"`
MCPServers map[string]*TargetConfig `json:"mcpServers,omitempty"`
Agents map[string]*AgentConfig `json:"agents,omitempty"`
Users *UsersConfig `json:"users,omitempty"`
Groups map[string]*GroupConfig `json:"groups,omitempty"`
// ModelAliases maps an operator-defined alias id to a target model id
// (spec §5, v0.2) — e.g. {"aliased/coding": "anthropic/claude-sonnet-4-5"}.
// An exact alias match wins resolution before any other rule
@@ -39,7 +43,8 @@ type Config struct {
// exactly: resolve never consults an empty alias map, so no existing
// model id's resolution changes. Validated at construction
// (validateModelAliases, registry.go).
ModelAliases map[string]string `json:"modelAliases,omitempty"`
ModelAliases map[string]string `json:"modelAliases,omitempty"`
Providers map[string]*ProviderConfig `json:"providers,omitempty"`
// ModelMeta declares per-model metadata overrides (feature v0.23):
// context window size and per-token cost, keyed by an exact
// "provider/model" id or a bare model id (applying wherever that bare
@@ -54,6 +59,18 @@ type Config struct {
// discovery/builtin/absent, same as before this feature existed.
// Validated at construction (validateModelMeta, modelmeta.go).
ModelMeta map[string]*ModelMetaConfig `json:"modelMeta,omitempty"`
// Breaker configures the discovery circuit breaker (feat/provider-
// health): how many consecutive discovery-refresh failures open a
// provider's breaker, and how long it then backs off before probing
// again. A struct value, not a pointer, for the same reason as Retry/
// Cache below — its fields are individually zero-means-default
// (validateBreakerConfig, registry.go), not gated by one Enabled
// flag, so "omitempty" on this tag would be a no-op that misleadingly
// implies otherwise. The zero value (Config{} with no breaker block
// at all) resolves to every documented default and behaves exactly
// like a deployment with no circuit breaker at all for a provider
// that never fails — see validateBreakerConfig's own doc comment.
Breaker BreakerConfig `json:"breaker"`
// 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
// mere presence is the signal) — so its tag omits "omitempty":
@@ -63,8 +80,7 @@ type Config struct {
Retry RetryConfig `json:"retry"`
// Cache is a struct value, not a pointer, for the same reason as Retry
// above: its own Enabled field is the on/off signal.
Cache CacheConfig `json:"cache"`
PassthroughUnknown bool `json:"passthroughUnknown,omitempty"`
Cache CacheConfig `json:"cache"`
// MaxInFlightBodyRequests caps how many unified and media JSON/
// multipart requests (routes_unified.go's runUnified via
// readAndDecodeUnifiedBody; routes_media.go's decodeMediaJSONRequest/
@@ -85,7 +101,8 @@ type Config struct {
// maxExplicitBodyAdmissionCap (llmgateway.go): "wins" means it
// overrides the self-tuned value, not that an operator can silently
// disable the bound entirely with an unbounded number.
MaxInFlightBodyRequests int `json:"maxInFlightBodyRequests,omitempty"`
MaxInFlightBodyRequests int `json:"maxInFlightBodyRequests,omitempty"`
PassthroughUnknown bool `json:"passthroughUnknown,omitempty"`
}
// ProviderConfig describes one upstream LLM provider.
@@ -278,6 +295,35 @@ type CacheConfig struct {
Enabled bool `json:"enabled,omitempty"`
}
// BreakerConfig configures the per-provider discovery circuit breaker
// (feat/provider-health): closed -> open -> half-open, driven entirely by
// discovery-refresh outcomes (registry.go's providerState.finishRefresh),
// never by request-path traffic. Every field is individually zero-means-
// default (validateBreakerConfig, registry.go) — there is no Enabled
// flag, unlike Retry/Cache above, because the breaker is always live:
// its defaults are chosen so a provider that never fails never notices
// it exists, so there is nothing meaningful to "disable". Field order
// (both strings before the int) is fieldalignment-sensitive, the same
// convention providerState's own doc comment explains.
type BreakerConfig struct {
// OpenDuration is the base backoff a newly opened breaker waits
// before its first half-open probe (a Go duration string, e.g.
// "1m"). Doubles on every further failed probe, capped at
// MaxOpenDuration. Empty (the default) uses
// defaultBreakerOpenDuration (registry.go).
OpenDuration string `json:"openDuration,omitempty"`
// MaxOpenDuration caps the exponential backoff OpenDuration above
// doubles into. Empty (the default) uses
// defaultBreakerMaxOpenDuration (registry.go). Must be >=
// OpenDuration when both are set.
MaxOpenDuration string `json:"maxOpenDuration,omitempty"`
// FailureThreshold is how many CONSECUTIVE failed discovery refreshes
// open the breaker. 0 (the default) uses defaultBreakerFailureThreshold
// (registry.go). A value outside 1..maxBreakerFailureThreshold is a
// construction error.
FailureThreshold int `json:"failureThreshold,omitempty"`
}
// AdminConfig configures the read-only admin dashboard (spec §4, v0.2).
// The zero value (Enabled: false) disables it, preserving v0.1 behavior
// exactly: no /admin* route is registered, so those paths fall through
registry.go
+346 -32
View File
@@ -28,6 +28,28 @@ const warmFillTimeout = 5 * time.Second
// goroutine's lifetime if an upstream hangs.
const backgroundRefreshTimeout = 30 * time.Second
// defaultBreakerFailureThreshold is how many CONSECUTIVE failed discovery
// refreshes open a provider's circuit breaker (feat/provider-health),
// applied when BreakerConfig.FailureThreshold is left at 0. Chosen so a
// single transient blip (one bad refresh) never trips it — only a
// provider that is persistently broken, like xiaomi's ongoing 401s, does.
const defaultBreakerFailureThreshold = 3
// maxBreakerFailureThreshold bounds BreakerConfig.FailureThreshold — a
// generous ceiling; there is no legitimate reason to require more
// consecutive failures than this before backing off at all.
const maxBreakerFailureThreshold = 100
// defaultBreakerOpenDuration is the base backoff a newly opened breaker
// waits before its first half-open probe, applied when BreakerConfig.
// OpenDuration is left empty.
const defaultBreakerOpenDuration = time.Minute
// defaultBreakerMaxOpenDuration caps the exponential backoff the breaker
// doubles into on every further failed probe, applied when BreakerConfig.
// MaxOpenDuration is left empty.
const defaultBreakerMaxOpenDuration = 30 * time.Minute
// errModelUnknown is returned by modelRegistry.resolve when id does not
// match any provider's known model set (explicit config plus the last
// successful discovery fetch) — 404 semantics for the caller.
@@ -39,15 +61,59 @@ var errModelUnknown = errors.New("llmgateway: unknown model")
// such model" from "that model exists, you cannot use it".
var errModelDenied = errors.New("llmgateway: model access denied")
// breakerState is one provider's discovery circuit breaker state
// (feat/provider-health): closed (normal — refreshes run on the
// configured interval, exactly like before this feature existed), open
// (discovery refreshes are skipped until openUntil, backing off
// exponentially on repeated failure), or halfOpen (exactly one probe
// refresh is in flight, deciding whether to close the breaker again).
// The zero value is closed, so a providerState built without ever
// touching health fields (every existing construction path, and every
// provider that never fails) behaves exactly as it did before this type
// existed.
type breakerState int32
const (
breakerClosed breakerState = iota
breakerOpen
breakerHalfOpen
)
// String renders bs for logs and the admin dashboard (feat/
// provider-health) — lower-case, hyphenated for the two-word state, to
// read naturally as a status word in either place.
func (bs breakerState) String() string {
switch bs {
case breakerOpen:
return "open"
case breakerHalfOpen:
return "half-open"
default:
return "closed"
}
}
// providerState tracks one provider's known model ids and discovery
// bookkeeping. explicit is set once at construction from
// ProviderConfig.Models and never mutated afterward; discovered is
// stale-while-error — a failed refresh leaves the previous discovered set
// in place rather than clearing it. mu guards every mutable field below it.
//
// Field order below is fieldalignment-sensitive (golangci-lint's govet
// enable-all): every pointer-containing field (time.Time — its loc
// *Location field carries a pointer — plus every map/string) is grouped
// first, every pointer-free field (durations, sync.Mutex, the breaker's
// own int/int32 counters, and the two trailing bools) last. Keep new
// fields in the matching group rather than appending at the very end.
type providerState struct {
lastRefresh time.Time
explicit map[string]bool
discovered map[string]bool
// openUntil is when an open breaker (feat/provider-health) next
// allows a half-open probe — see recordHealthLocked/tryBeginRefresh.
// Grouped with lastRefresh above (both time.Time) rather than beside
// the rest of the breaker fields further down, which are pointer-free.
openUntil time.Time
explicit map[string]bool
discovered map[string]bool
// discoveredContext holds discovery-captured per-model context
// lengths (feature v0.23, ProviderConfig.MetadataPath — currently
// only openai-type adapters ever populate this, via
@@ -66,11 +132,31 @@ type providerState struct {
// outcome, not the stale-while-error discovered set: a provider can
// show a non-empty lastErr while modelCount still reflects its last
// successful discovery.
lastErr string
interval time.Duration
mu sync.Mutex
discoveryEnabled bool
inFlight bool
lastErr string
interval time.Duration
mu sync.Mutex
// --- feat/provider-health: discovery circuit breaker, guarded by mu
// like every other mutable field above. consecutiveFailures counts
// unbroken failures while closed, toward breakerThreshold. backoff is
// the duration that produced openUntil above (doubled on the next
// failed probe, capped at breakerOpenMax). breakerThreshold/
// breakerOpenBase/breakerOpenMax are resolved once, at construction
// (newModelRegistry, from Config.Breaker via validateBreakerConfig),
// and never change afterward — copied onto every providerState
// rather than shared from one place, the same "baked in at
// construction" convention interval above already uses, so
// recordHealthLocked/tryBeginRefresh never need a second parameter or
// a pointer back to the registry. health is the breaker's current
// state, declared last among these since breakerState is a 4-byte
// int32, smaller than its 8-byte neighbors above.
consecutiveFailures int
backoff time.Duration
breakerThreshold int
breakerOpenBase time.Duration
breakerOpenMax time.Duration
health breakerState
discoveryEnabled bool
inFlight bool
}
// hasModel reports whether id is in this provider's explicit or discovered
@@ -132,14 +218,36 @@ func (st *providerState) setDiscoveredContext(m map[string]int) {
st.discoveredContext = m
}
// tryBeginRefresh reports whether now is far enough past lastRefresh (or
// this is the first refresh) to start a new one, and if so marks the
// provider inFlight so a concurrent caller cannot start a second one. The
// caller must pair a true result with a later finishRefresh call.
// tryBeginRefresh reports whether now permits starting a new refresh, and
// if so marks the provider inFlight so a concurrent caller cannot start a
// second one. The caller must pair a true result with a later finishRefresh
// call.
//
// A closed breaker (feat/provider-health) behaves EXACTLY as before this
// feature existed: now far enough past lastRefresh (or this is the first
// refresh) is the only gate — a provider that never fails never touches the
// breaker branch below at all. An open breaker replaces that interval gate
// with its own backoff window: refresh stays blocked until now reaches
// openUntil, regardless of interval, and reaching it starts exactly one
// half-open probe (state moves to halfOpen here, before the caller's fetch
// even runs) rather than resuming normal per-interval refreshing outright —
// recordHealthLocked decides whether that probe closes the breaker or
// reopens it with a longer backoff.
func (st *providerState) tryBeginRefresh(now time.Time) bool {
st.mu.Lock()
defer st.mu.Unlock()
if st.inFlight || now.Sub(st.lastRefresh) < st.interval {
if st.inFlight {
return false
}
if st.health == breakerOpen {
if now.Before(st.openUntil) {
return false
}
st.health = breakerHalfOpen
st.inFlight = true
return true
}
if now.Sub(st.lastRefresh) < st.interval {
return false
}
st.inFlight = true
@@ -152,7 +260,10 @@ func (st *providerState) tryBeginRefresh(now time.Time) bool {
// what throttles a failing provider to one attempt per interval instead of
// retrying on every call. On success (err == nil) ids replaces the
// discovered set; on failure the previous discovered set is kept
// (stale-while-error).
// (stale-while-error). Either way, recordHealthLocked (feat/provider-health)
// updates the discovery circuit breaker from this same outcome — a healthy
// provider's breaker never leaves closed, so this adds no new behavior for
// the common case, only for a provider that is actually failing.
func (st *providerState) finishRefresh(now time.Time, ids []string, err error) {
st.mu.Lock()
defer st.mu.Unlock()
@@ -160,14 +271,97 @@ func (st *providerState) finishRefresh(now time.Time, ids []string, err error) {
st.lastRefresh = now
if err != nil {
st.lastErr = err.Error()
} else {
st.lastErr = ""
discovered := make(map[string]bool, len(ids))
for _, id := range ids {
discovered[id] = true
}
st.discovered = discovered
}
st.recordHealthLocked(now, err)
}
// recordHealthLocked updates st's discovery circuit breaker (feat/
// provider-health) from one finished refresh attempt's outcome. Caller must
// already hold st.mu.
//
// failed reuses limits.go's existing recordProviderAttempt classification —
// isTransient(resp, err) with a nil *http.Response (a discovery fetch has no
// response to inspect on a transport-level failure, unlike a proxied
// request) plus isDeadlineExceeded(err) — rather than inventing a second
// one, per this feature's own design mandate. Net effect: any non-nil error
// except a caller-canceled context counts against the provider, the same
// line recordProviderAttempt already draws for request-path attempts.
//
// While closed, breakerThreshold consecutive failures open the breaker at
// its base backoff (breakerOpenBase); any success resets the counter to
// zero. While halfOpen (the only other state finishRefresh can observe —
// tryBeginRefresh never starts a refresh from open without first moving to
// halfOpen), a successful probe closes the breaker outright; a failed one
// reopens it with a DOUBLED backoff, capped at breakerOpenMax — the
// exponential escalation that keeps a persistently broken provider (like
// xiaomi's ongoing 401s) from being re-probed every single interval
// forever.
func (st *providerState) recordHealthLocked(now time.Time, err error) {
failed := isTransient(nil, err) || isDeadlineExceeded(err)
if st.health == breakerHalfOpen {
if failed {
st.openBreakerLocked(now, true)
} else {
st.closeBreakerLocked()
}
return
}
st.lastErr = ""
discovered := make(map[string]bool, len(ids))
for _, id := range ids {
discovered[id] = true
if !failed {
st.closeBreakerLocked()
return
}
st.discovered = discovered
st.consecutiveFailures++
if st.consecutiveFailures >= st.breakerThreshold {
st.openBreakerLocked(now, false)
}
}
// closeBreakerLocked resets st to a healthy, closed breaker — reachable
// from a successful attempt while closed (nothing to do in practice, since
// consecutiveFailures is already 0, but harmless to reset) or a successful
// half-open probe (the "a provider that recovers must never stay dead"
// requirement). Caller must already hold st.mu.
func (st *providerState) closeBreakerLocked() {
st.health = breakerClosed
st.consecutiveFailures = 0
st.backoff = 0
st.openUntil = time.Time{}
}
// openBreakerLocked opens st's breaker as of now, computing the backoff
// window a subsequent tryBeginRefresh must wait out before its next
// half-open probe. escalate true (a failed half-open probe re-opening)
// doubles the PREVIOUS backoff; escalate false (the closed->open trip)
// starts fresh at breakerOpenBase — either way capped at breakerOpenMax so
// a persistently broken provider's backoff grows but never exceeds it.
// Caller must already hold st.mu.
func (st *providerState) openBreakerLocked(now time.Time, escalate bool) {
if escalate && st.backoff > 0 {
st.backoff *= 2
} else {
st.backoff = st.breakerOpenBase
}
if st.backoff > st.breakerOpenMax {
st.backoff = st.breakerOpenMax
}
st.health = breakerOpen
st.openUntil = now.Add(st.backoff)
}
// usable reports whether st's discovery circuit breaker currently permits
// routing to this provider — see modelRegistry.providerUsable's doc comment
// for the full contract this backs.
func (st *providerState) usable() bool {
st.mu.Lock()
defer st.mu.Unlock()
return st.health != breakerOpen
}
// snapshot returns st's read-only view for the admin dashboard (spec §4,
@@ -191,10 +385,15 @@ func (st *providerState) finishRefresh(now time.Time, ids []string, err error) {
// previous discovered set rather than clearing it. This method applies
// no special handling for that case; it is simply the same set every
// other snapshot consumer (modelCount, listFor) already reads.
func (st *providerState) snapshot() (models []string, lastRefresh time.Time, lastErr string) {
// health and openUntil (feat/provider-health) join the same locked pass for
// the identical reason the doc comment above already gives for models/
// lastRefresh/lastErr: a concurrent finishRefresh must never be able to
// hand the admin dashboard a breaker state that disagrees with the
// lastErr/models it is shown alongside.
func (st *providerState) snapshot() (models []string, lastRefresh time.Time, lastErr string, health breakerState, openUntil time.Time) {
st.mu.Lock()
defer st.mu.Unlock()
return st.knownIDsLocked(), st.lastRefresh, st.lastErr
return st.knownIDsLocked(), st.lastRefresh, st.lastErr, st.health, st.openUntil
}
// modelRegistry aggregates every configured provider's explicit and
@@ -285,6 +484,65 @@ type modelsCacheEntry struct {
gen int64
}
// breakerConfig is Config.Breaker (feat/provider-health), validated and
// defaulted once by validateBreakerConfig — the resolved values
// newModelRegistry copies onto every provider's providerState at
// construction (providerState's own doc comment explains why copied,
// not shared).
type breakerConfig struct {
failureThreshold int
openBase time.Duration
openMax time.Duration
}
// validateBreakerConfig validates bc and returns the resolved breakerConfig
// (feat/provider-health) — mirrors newRetryPolicy (retry.go) and
// validateCacheConfig (cache.go)'s own zero-means-default, validated-and-
// resolved shape. bc's zero value (BreakerConfig{}, an operator who
// configures no breaker block at all) resolves to every default constant
// above unchanged: a provider that never fails never has a reason to
// consult any of these three numbers, so their defaults exist only to
// bound how a FAILING provider backs off, never to alter a healthy one.
func validateBreakerConfig(bc BreakerConfig) (breakerConfig, error) {
threshold := bc.FailureThreshold
if threshold == 0 {
threshold = defaultBreakerFailureThreshold
}
if threshold < 1 || threshold > maxBreakerFailureThreshold {
return breakerConfig{}, fmt.Errorf("llmgateway: breaker.failureThreshold must be between 1 and %d, got %d", maxBreakerFailureThreshold, threshold)
}
openBase := defaultBreakerOpenDuration
if bc.OpenDuration != "" {
d, err := time.ParseDuration(bc.OpenDuration)
if err != nil {
return breakerConfig{}, fmt.Errorf("llmgateway: breaker.openDuration %q is invalid: %w", bc.OpenDuration, err)
}
if d <= 0 {
return breakerConfig{}, fmt.Errorf("llmgateway: breaker.openDuration must be positive, got %q", bc.OpenDuration)
}
openBase = d
}
openMax := defaultBreakerMaxOpenDuration
if bc.MaxOpenDuration != "" {
d, err := time.ParseDuration(bc.MaxOpenDuration)
if err != nil {
return breakerConfig{}, fmt.Errorf("llmgateway: breaker.maxOpenDuration %q is invalid: %w", bc.MaxOpenDuration, err)
}
if d <= 0 {
return breakerConfig{}, fmt.Errorf("llmgateway: breaker.maxOpenDuration must be positive, got %q", bc.MaxOpenDuration)
}
openMax = d
}
if openMax < openBase {
return breakerConfig{}, fmt.Errorf("llmgateway: breaker.maxOpenDuration (%s) must be >= breaker.openDuration (%s)", openMax, openBase)
}
return breakerConfig{failureThreshold: threshold, openBase: openBase, openMax: openMax}, nil
}
// newModelRegistry builds a modelRegistry from adapters and cfg's matching
// ProviderConfig entries. It does not perform discovery itself — the
// caller (newGateway) runs the synchronous first fill separately via
@@ -292,6 +550,11 @@ type modelsCacheEntry struct {
// provider's ProviderConfig.DiscoveryInterval that fails time.ParseDuration
// is a constructor error; an empty one defaults to defaultDiscoveryInterval.
func newModelRegistry(adapters map[string]providerAdapter, cfg *Config, log func(string, ...any)) (*modelRegistry, error) {
breaker, err := validateBreakerConfig(cfg.Breaker)
if err != nil {
return nil, err
}
m := &modelRegistry{
adapters: adapters,
states: make(map[string]*providerState, len(adapters)),
@@ -320,9 +583,14 @@ func newModelRegistry(adapters map[string]providerAdapter, cfg *Config, log func
interval := defaultDiscoveryInterval
if pc.DiscoveryInterval != "" {
d, err := time.ParseDuration(pc.DiscoveryInterval)
if err != nil {
return nil, fmt.Errorf("llmgateway: provider %q: invalid discoveryInterval %q: %w", name, pc.DiscoveryInterval, err)
// parseErr, not err: newModelRegistry's own outer err (from
// validateBreakerConfig above) is already in scope here, and
// govet's shadow check (golangci-lint's govet enable-all)
// flags a nested ":=" reusing that identifier even though it
// is scoped to this if-block only.
d, parseErr := time.ParseDuration(pc.DiscoveryInterval)
if parseErr != nil {
return nil, fmt.Errorf("llmgateway: provider %q: invalid discoveryInterval %q: %w", name, pc.DiscoveryInterval, parseErr)
}
interval = d
}
@@ -338,6 +606,9 @@ func newModelRegistry(adapters map[string]providerAdapter, cfg *Config, log func
discovered: make(map[string]bool),
discoveryEnabled: pc.Discovery,
interval: interval,
breakerThreshold: breaker.failureThreshold,
breakerOpenBase: breaker.openBase,
breakerOpenMax: breaker.openMax,
}
}
@@ -433,6 +704,32 @@ func (m *modelRegistry) now() time.Time {
return m.nowFn()
}
// providerUsable reports whether name's discovery circuit breaker
// currently permits routing to it (feat/provider-health's queryable
// health signal — the predicate feat/failover consults before sending a
// request to a provider). false only while the breaker is OPEN: an
// unconfigured provider name, one with discovery disabled (it has no
// health signal to trip the breaker with, so it stays closed forever),
// and one whose breaker is closed OR half-open all report true — a
// half-open provider is actively being re-probed by the next discovery
// cycle and must stay routable, per this feature's own "a provider that
// recovers must never stay dead" requirement; only an OPEN breaker, still
// backing off, is unusable.
//
// This is a single mutex-guarded read of state finishRefresh/
// tryBeginRefresh already maintain — no round trip, no store lookup — per
// this feature's own design: per-pod, in-memory, a latency optimization
// rather than a correctness guarantee. Three pods learning independently
// (each with its own view of a provider's health) is accepted and
// intended, not a bug.
func (m *modelRegistry) providerUsable(name string) bool {
st, ok := m.states[name]
if !ok {
return true
}
return st.usable()
}
// finishRefresh records the outcome of a refresh attempt (st.finishRefresh)
// and invalidates modelsJSON's cache by bumping modelsGen — the ONLY two
// places that ever mutate st.discovered or st.discoveredContext both
@@ -1104,14 +1401,28 @@ func (m *modelRegistry) resolveMetaForAlias(alias, targetProvider, targetModel s
// model id names no credential, and this same set is already public
// through /v1/models (listFor) to any authenticated caller the group
// allows.
// Field order below is fieldalignment-sensitive, the same convention
// providerState's own doc comment explains: pointer-containing fields
// (time.Time, string, []string) grouped first, pointer-free fields
// (modelCount, health, discoveryEnabled) last.
type providerSnapshot struct {
lastRefresh time.Time
name string
typeName string
baseURL string
lastErr string
models []string
modelCount int
// openUntil mirrors providerState's own field of the same name
// (feat/provider-health) — see providerState.snapshot's doc comment
// for why it is read in the same locked pass as the rest of this
// struct's fields.
openUntil time.Time
name string
typeName string
baseURL string
lastErr string
models []string
modelCount int
// health mirrors providerState's own field of the same name (feat/
// provider-health): the admin dashboard's breaker-state column. Every
// provider reads closed until a discovery failure actually trips its
// breaker — a healthy deployment's snapshot is unaffected.
health breakerState
// discoveryEnabled mirrors providerState.discoveryEnabled (Feature B,
// v0.22 last-refresh label honesty): read directly rather than through
// providerState.snapshot(), because it is set once at construction
@@ -1164,13 +1475,14 @@ func (m *modelRegistry) snapshot() []providerSnapshot {
out := make([]providerSnapshot, 0, len(m.providerNames))
for _, name := range m.providerNames {
adapter := m.adapters[name]
// Three-value multi-assign straight from the call, no intermediate
// Five-value multi-assign straight from the call, no intermediate
// named variables reused across a loop iteration boundary — Yaegi
// has had multi-assign edge cases around reused/shadowed loop
// variables in other parts of this codebase's history; this shape
// (fresh locals every iteration, assigned once, read once) avoids
// that class of trap entirely.
models, lastRefresh, lastErr := m.states[name].snapshot()
// that class of trap entirely. health/openUntil (feat/provider-
// health) widen the same tuple rather than a second locked call.
models, lastRefresh, lastErr, health, openUntil := m.states[name].snapshot()
out = append(out, providerSnapshot{
name: adapter.name(),
typeName: adapter.typeName(),
@@ -1180,6 +1492,8 @@ func (m *modelRegistry) snapshot() []providerSnapshot {
lastRefresh: lastRefresh,
lastErr: lastErr,
discoveryEnabled: m.states[name].discoveryEnabled,
health: health,
openUntil: openUntil,
})
}
return out
registry_health_test.go
+366
View File
@@ -0,0 +1,366 @@
package traefikllmgateway
import (
"context"
"errors"
"testing"
"time"
)
// --- validateBreakerConfig: defaults and validation table ---
func TestValidateBreakerConfig_Table(t *testing.T) {
tests := []struct {
name string
bc BreakerConfig
wantThresh int
wantOpen time.Duration
wantMaxOpen time.Duration
wantErr bool
}{
{
name: "zero value uses every default",
bc: BreakerConfig{},
wantThresh: defaultBreakerFailureThreshold,
wantOpen: defaultBreakerOpenDuration,
wantMaxOpen: defaultBreakerMaxOpenDuration,
},
{
name: "explicit valid overrides used verbatim",
bc: BreakerConfig{FailureThreshold: 5, OpenDuration: "10s", MaxOpenDuration: "2m"},
wantThresh: 5,
wantOpen: 10 * time.Second,
wantMaxOpen: 2 * time.Minute,
},
{
name: "threshold at the max boundary is accepted",
bc: BreakerConfig{FailureThreshold: maxBreakerFailureThreshold},
wantThresh: maxBreakerFailureThreshold,
wantOpen: defaultBreakerOpenDuration, wantMaxOpen: defaultBreakerMaxOpenDuration,
},
{
name: "threshold above max is an error",
bc: BreakerConfig{FailureThreshold: maxBreakerFailureThreshold + 1},
wantErr: true,
},
{
name: "negative threshold is an error",
bc: BreakerConfig{FailureThreshold: -1},
wantErr: true,
},
{
name: "unparseable openDuration is an error",
bc: BreakerConfig{OpenDuration: "not-a-duration"},
wantErr: true,
},
{
name: "zero-valued openDuration string is an error",
bc: BreakerConfig{OpenDuration: "0s"},
wantErr: true,
},
{
name: "negative openDuration is an error",
bc: BreakerConfig{OpenDuration: "-1s"},
wantErr: true,
},
{
name: "unparseable maxOpenDuration is an error",
bc: BreakerConfig{MaxOpenDuration: "not-a-duration"},
wantErr: true,
},
{
name: "maxOpenDuration below openDuration is an error",
bc: BreakerConfig{OpenDuration: "5m", MaxOpenDuration: "1m"},
wantErr: true,
},
{
name: "maxOpenDuration equal to openDuration is accepted",
bc: BreakerConfig{OpenDuration: "5m", MaxOpenDuration: "5m"},
wantThresh: defaultBreakerFailureThreshold,
wantOpen: 5 * time.Minute,
wantMaxOpen: 5 * time.Minute,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got, err := validateBreakerConfig(tt.bc)
if (err != nil) != tt.wantErr {
t.Fatalf("validateBreakerConfig(%+v): err = %v, wantErr %v", tt.bc, err, tt.wantErr)
}
if tt.wantErr {
return
}
if got.failureThreshold != tt.wantThresh {
t.Errorf("failureThreshold = %d, want %d", got.failureThreshold, tt.wantThresh)
}
if got.openBase != tt.wantOpen {
t.Errorf("openBase = %v, want %v", got.openBase, tt.wantOpen)
}
if got.openMax != tt.wantMaxOpen {
t.Errorf("openMax = %v, want %v", got.openMax, tt.wantMaxOpen)
}
})
}
}
// newHealthTestState builds a providerState with the breaker fields set
// directly (no newModelRegistry round trip needed — the state machine
// under test is pure providerState methods), and every other field at a
// harmless zero/empty value. interval is set deliberately SHORT relative
// to openBase in the tests below so a passing assertion can only be
// explained by the breaker's own backoff window, never by the ordinary
// per-interval throttle it lives alongside.
func newHealthTestState(interval time.Duration, threshold int, openBase, openMax time.Duration) *providerState {
return &providerState{
explicit: make(map[string]bool),
discovered: make(map[string]bool),
interval: interval,
breakerThreshold: threshold,
breakerOpenBase: openBase,
breakerOpenMax: openMax,
}
}
// --- breaker: opens after the configured threshold ---
func TestProviderState_Breaker_OpensAfterThreshold(t *testing.T) {
t.Parallel()
st := newHealthTestState(time.Hour, 3, time.Minute, 10*time.Minute)
now := time.Now()
boom := errors.New("boom")
for i := 0; i < 2; i++ {
st.finishRefresh(now, nil, boom)
if !st.usable() {
t.Fatalf("after %d failure(s) (threshold 3): usable() = false, want true — must not open early", i+1)
}
}
st.finishRefresh(now, nil, boom) // 3rd consecutive failure reaches the threshold
if st.usable() {
t.Fatal("after reaching the failure threshold: usable() = true, want false (breaker must be open)")
}
}
// TestProviderState_Breaker_ASuccessResetsTheFailureCounter proves the
// counter is CONSECUTIVE, not cumulative: two failures, a success, then two
// more failures must not open a threshold-3 breaker, because the success
// in between reset the count back to zero.
func TestProviderState_Breaker_ASuccessResetsTheFailureCounter(t *testing.T) {
t.Parallel()
st := newHealthTestState(time.Hour, 3, time.Minute, 10*time.Minute)
now := time.Now()
boom := errors.New("boom")
st.finishRefresh(now, nil, boom)
st.finishRefresh(now, nil, boom)
st.finishRefresh(now, []string{"m1"}, nil) // success: resets the streak
st.finishRefresh(now, nil, boom)
st.finishRefresh(now, nil, boom)
if !st.usable() {
t.Error("usable() = false, want true — the intervening success must have reset the consecutive-failure streak")
}
}
// --- breaker: stays open for the backoff window, superseding interval ---
func TestProviderState_Breaker_StaysOpenForBackoffWindow(t *testing.T) {
t.Parallel()
// interval is deliberately much shorter than openBase: a bare pass
// through tryBeginRefresh would allow a retry after 5s. Only the
// breaker's own backoff should be blocking it past that point.
st := newHealthTestState(5*time.Second, 1, time.Minute, 10*time.Minute)
base := time.Now()
if ok := st.tryBeginRefresh(base); !ok {
t.Fatal("first tryBeginRefresh on a fresh provider = false, want true")
}
st.finishRefresh(base, nil, errors.New("boom")) // trips the breaker (threshold 1)
if st.usable() {
t.Fatal("usable() = true after the trip, want false")
}
if ok := st.tryBeginRefresh(base.Add(6 * time.Second)); ok {
t.Error("tryBeginRefresh past the plain interval (5s) but well inside the 1m backoff = true, want false — backoff must supersede interval")
}
if ok := st.tryBeginRefresh(base.Add(59 * time.Second)); ok {
t.Error("tryBeginRefresh just before the backoff window elapses = true, want false")
}
if ok := st.tryBeginRefresh(base.Add(61 * time.Second)); !ok {
t.Fatal("tryBeginRefresh once the backoff window has elapsed = false, want true (half-open probe should start)")
}
if !st.usable() {
t.Error("usable() during the half-open probe = false, want true — a provider being re-probed must stay routable")
}
}
// --- breaker: half-open probe success closes it again ---
func TestProviderState_Breaker_HalfOpenProbe_SuccessCloses(t *testing.T) {
t.Parallel()
st := newHealthTestState(5*time.Second, 1, time.Minute, 10*time.Minute)
base := time.Now()
st.tryBeginRefresh(base)
st.finishRefresh(base, nil, errors.New("boom")) // open
probeAt := base.Add(61 * time.Second)
if ok := st.tryBeginRefresh(probeAt); !ok {
t.Fatal("tryBeginRefresh after the backoff window = false, want true")
}
st.finishRefresh(probeAt, []string{"m1"}, nil) // probe succeeds
if !st.usable() {
t.Error("usable() after a successful half-open probe = false, want true — a recovered provider must not stay dead")
}
if !st.hasModel("m1") {
t.Error("hasModel(m1) = false after a successful probe, want true — discovery still updates the model set normally")
}
// Closed again means ORDINARY interval gating resumed, not a fresh
// backoff window: immediately after the probe, another attempt must
// still wait out the plain 5s interval, and no longer than that.
if ok := st.tryBeginRefresh(probeAt); ok {
t.Error("tryBeginRefresh at the same instant as the closing probe = true, want false (interval not yet elapsed)")
}
if ok := st.tryBeginRefresh(probeAt.Add(6 * time.Second)); !ok {
t.Error("tryBeginRefresh past the plain interval after closing = false, want true")
}
}
// --- breaker: half-open probe failure reopens it, with escalating backoff ---
func TestProviderState_Breaker_HalfOpenProbe_FailureReopens_Escalates(t *testing.T) {
t.Parallel()
st := newHealthTestState(5*time.Second, 1, time.Minute, 10*time.Minute)
base := time.Now()
st.tryBeginRefresh(base)
st.finishRefresh(base, nil, errors.New("boom")) // opens, backoff = 1m
probeAt := base.Add(61 * time.Second)
if ok := st.tryBeginRefresh(probeAt); !ok {
t.Fatal("tryBeginRefresh after the first backoff window = false, want true")
}
st.finishRefresh(probeAt, nil, errors.New("still broken")) // probe fails: reopen, backoff doubles to 2m
if st.usable() {
t.Fatal("usable() after a failed probe = true, want false")
}
if ok := st.tryBeginRefresh(probeAt.Add(61 * time.Second)); ok {
t.Error("tryBeginRefresh after only the ORIGINAL (1m) backoff = true, want false — a failed probe must double it to 2m")
}
if ok := st.tryBeginRefresh(probeAt.Add(121 * time.Second)); !ok {
t.Fatal("tryBeginRefresh after the doubled (2m) backoff = false, want true")
}
}
// TestProviderState_Breaker_BackoffCapsAtOpenMax proves repeated failed
// probes double the backoff only up to breakerOpenMax, never past it.
func TestProviderState_Breaker_BackoffCapsAtOpenMax(t *testing.T) {
t.Parallel()
st := newHealthTestState(time.Second, 1, time.Minute, 3*time.Minute) // base 1m, cap 3m: 1m -> 2m -> capped at 3m (not 4m)
base := time.Now()
st.tryBeginRefresh(base)
st.finishRefresh(base, nil, errors.New("boom")) // open, backoff 1m
at := base.Add(61 * time.Second)
st.tryBeginRefresh(at)
st.finishRefresh(at, nil, errors.New("boom")) // reopen, backoff doubles to 2m
at = at.Add(121 * time.Second)
st.tryBeginRefresh(at)
st.finishRefresh(at, nil, errors.New("boom")) // reopen, doubling to 4m would exceed the 3m cap
if ok := st.tryBeginRefresh(at.Add(2*time.Minute + 59*time.Second)); ok {
t.Error("tryBeginRefresh just before the capped 3m backoff = true, want false")
}
if ok := st.tryBeginRefresh(at.Add(3*time.Minute + 1*time.Second)); !ok {
t.Error("tryBeginRefresh just past the capped 3m backoff = false, want true — backoff must not exceed breakerOpenMax")
}
}
// --- breaker: a context.Canceled discovery error is not the provider's
// fault, and must not count toward the failure threshold (reuses limits.go's
// isTransient/isDeadlineExceeded classification, which draws this same
// line for request-path attempts) ---
func TestProviderState_Breaker_ContextCanceled_NeverCountsAsFailure(t *testing.T) {
t.Parallel()
st := newHealthTestState(time.Hour, 1, time.Minute, 10*time.Minute)
now := time.Now()
st.finishRefresh(now, nil, context.Canceled)
if !st.usable() {
t.Error("usable() = false after a context.Canceled discovery error, want true — a canceled attempt is not a provider fault")
}
}
// --- breaker: a provider that never fails is unaffected (default-
// preserving: today's interval-only gating, unchanged) ---
func TestProviderState_Breaker_HealthyProviderNeverAffected(t *testing.T) {
t.Parallel()
st := newHealthTestState(time.Hour, defaultBreakerFailureThreshold, defaultBreakerOpenDuration, defaultBreakerMaxOpenDuration)
now := time.Now()
for i := 0; i < 5; i++ {
if ok := st.tryBeginRefresh(now); !ok {
t.Fatalf("cycle %d: tryBeginRefresh = false, want true (interval already elapsed)", i)
}
st.finishRefresh(now, []string{"m1"}, nil)
if !st.usable() {
t.Fatalf("cycle %d: usable() = false, want true — a never-failing provider must never trip its breaker", i)
}
if !st.openUntil.IsZero() {
t.Fatalf("cycle %d: openUntil = %v, want zero — breaker must never have opened", i, st.openUntil)
}
// Immediately after: interval-gated exactly like before this
// feature existed, no breaker-induced extra wait.
if ok := st.tryBeginRefresh(now); ok {
t.Fatalf("cycle %d: tryBeginRefresh immediately after a fresh refresh = true, want false (interval not elapsed)", i)
}
now = now.Add(2 * time.Hour) // past the interval for the next cycle
}
}
// --- modelRegistry.providerUsable: the exported predicate feat/failover
// consumes ---
func TestModelRegistry_ProviderUsable(t *testing.T) {
t.Parallel()
adapters := map[string]providerAdapter{
"flaky": newFakeAdapter("flaky"),
"healthy": newFakeAdapter("healthy"),
}
cfg := &Config{
Providers: map[string]*ProviderConfig{
"flaky": {Models: []string{"m1"}},
"healthy": {Models: []string{"m2"}},
},
Breaker: BreakerConfig{FailureThreshold: 2, OpenDuration: "1m", MaxOpenDuration: "10m"},
}
reg, err := newModelRegistry(adapters, cfg, func(string, ...any) {})
if err != nil {
t.Fatalf("newModelRegistry: %v", err)
}
now := time.Now()
reg.nowFn = func() time.Time { return now }
if !reg.providerUsable("flaky") || !reg.providerUsable("healthy") {
t.Fatal("a freshly constructed registry's providers must both be usable")
}
if !reg.providerUsable("does-not-exist") {
t.Error(`providerUsable("does-not-exist") = false, want true — an unconfigured name is not a reason to block routing`)
}
boom := errors.New("boom")
reg.states["flaky"].finishRefresh(now, nil, boom)
reg.states["flaky"].finishRefresh(now, nil, boom) // reaches FailureThreshold: 2
if reg.providerUsable("flaky") {
t.Error(`providerUsable("flaky") = true after tripping its breaker, want false`)
}
if !reg.providerUsable("healthy") {
t.Error(`providerUsable("healthy") = false, want true — a sibling provider's breaker must not affect it`)
}
}
registry_modelmeta_test.go
+1 -1
View File
@@ -367,7 +367,7 @@ func TestModelRegistry_RefreshProvider_MetadataPanicDoesNotFailDiscovery(t *test
if !reg.states["lmstudio"].hasModel("m2") {
t.Error("a panic in metadata capture must not discard the successful discovery result")
}
_, _, lastErr := reg.states["lmstudio"].snapshot()
_, _, lastErr, _, _ := reg.states["lmstudio"].snapshot()
if lastErr != "" {
t.Errorf("lastErr = %q, want empty — discovery itself succeeded; only metadata capture panicked", lastErr)
}