From 41e322c53d7b4b796eb377d0df9c29ecd10ba431 Mon Sep 17 00:00:00 2001 From: Chia Date: Thu, 6 Aug 2026 09:29:41 +1200 Subject: feat: complete commercial control plane, billing, auth, and model catalog - add PostgreSQL control-plane persistence with Redis-degraded hot reload - implement prepaid balance, usage ledger, Stripe top-up and reconciliation - add registration, email verification, password reset, invitations and RBAC - support TOTP, Passkey MFA, device sessions, quotas and rate limits - add tenant billing profiles, audit logs and operational readiness checks - build authenticated admin console, Quickstart, Playground and usage analytics - add public model catalog with pricing, filtering and cost estimation - support OpenAI Responses providers and provider health failover - validate real upstream usage reporting and balance settlement --- internal/providerhealth/tracker.go | 200 +++++++++++++++++++++++++++++++++++++ 1 file changed, 200 insertions(+) create mode 100644 internal/providerhealth/tracker.go (limited to 'internal/providerhealth/tracker.go') diff --git a/internal/providerhealth/tracker.go b/internal/providerhealth/tracker.go new file mode 100644 index 0000000..a5af2b2 --- /dev/null +++ b/internal/providerhealth/tracker.go @@ -0,0 +1,200 @@ +package providerhealth + +import ( + "sort" + "sync" + "time" +) + +const recentWindow = 100 + +type RouteKey struct { + ModelID string + ProviderID string + WireAPI string +} + +type Observation struct { + StatusCode int + Latency time.Duration + Failed bool + ObservedAt time.Time +} + +type Status struct { + ModelID string `json:"model_id"` + ProviderID string `json:"provider_id"` + WireAPI string `json:"wire_api"` + State string `json:"state"` + Attempts uint64 `json:"attempts"` + RecentSamples int `json:"recent_samples"` + AvailabilityPercent float64 `json:"availability_percent"` + HeaderLatencyEWMA int64 `json:"header_latency_ewma_ms"` + ConsecutiveFailures uint64 `json:"consecutive_failures"` + LastStatusCode int `json:"last_status_code,omitempty"` + LastObservedAt *time.Time `json:"last_observed_at,omitempty"` + LastHealthyAt *time.Time `json:"last_healthy_at,omitempty"` + CircuitOpenUntil *time.Time `json:"circuit_open_until,omitempty"` +} + +type Options struct { + FailureThreshold uint64 + OpenDuration time.Duration + Now func() time.Time +} + +type Tracker struct { + states sync.Map + failureThreshold uint64 + openDuration time.Duration + now func() time.Time +} + +type routeState struct { + mu sync.RWMutex + attempts uint64 + consecutiveFailures uint64 + lastStatusCode int + lastObservedAt time.Time + lastHealthyAt time.Time + openUntil time.Time + headerLatencyEWMA float64 + recent [recentWindow]bool + recentCount int + recentPosition int + recentHealthy int +} + +func New(options Options) *Tracker { + if options.FailureThreshold == 0 { + options.FailureThreshold = 3 + } + if options.OpenDuration <= 0 { + options.OpenDuration = 30 * time.Second + } + if options.Now == nil { + options.Now = time.Now + } + return &Tracker{failureThreshold: options.FailureThreshold, openDuration: options.OpenDuration, now: options.Now} +} + +func (t *Tracker) Observe(key RouteKey, observation Observation) { + if t == nil || key.ProviderID == "" { + return + } + if observation.ObservedAt.IsZero() { + observation.ObservedAt = t.now() + } + value, _ := t.states.LoadOrStore(key, &routeState{}) + state := value.(*routeState) + state.mu.Lock() + defer state.mu.Unlock() + + state.attempts++ + state.lastStatusCode = observation.StatusCode + state.lastObservedAt = observation.ObservedAt + if observation.Latency > 0 { + latency := float64(observation.Latency.Milliseconds()) + if latency < 1 { + latency = 1 + } + if state.headerLatencyEWMA == 0 { + state.headerLatencyEWMA = latency + } else { + state.headerLatencyEWMA = state.headerLatencyEWMA*0.8 + latency*0.2 + } + } + state.addRecent(!observation.Failed) + if observation.Failed { + state.consecutiveFailures++ + if state.consecutiveFailures >= t.failureThreshold { + state.openUntil = observation.ObservedAt.Add(t.openDuration) + } + return + } + state.consecutiveFailures = 0 + state.openUntil = time.Time{} + state.lastHealthyAt = observation.ObservedAt +} + +func (s *routeState) addRecent(healthy bool) { + if s.recentCount == recentWindow { + if s.recent[s.recentPosition] { + s.recentHealthy-- + } + } else { + s.recentCount++ + } + s.recent[s.recentPosition] = healthy + if healthy { + s.recentHealthy++ + } + s.recentPosition = (s.recentPosition + 1) % recentWindow +} + +func (t *Tracker) CircuitOpen(key RouteKey) bool { + if t == nil { + return false + } + value, ok := t.states.Load(key) + if !ok { + return false + } + state := value.(*routeState) + state.mu.RLock() + defer state.mu.RUnlock() + return state.openUntil.After(t.now()) +} + +func (t *Tracker) Snapshot() []Status { + if t == nil { + return []Status{} + } + now := t.now() + result := make([]Status, 0) + t.states.Range(func(rawKey, rawState any) bool { + key := rawKey.(RouteKey) + state := rawState.(*routeState) + state.mu.RLock() + item := statusFromState(key, state, now) + state.mu.RUnlock() + result = append(result, item) + return true + }) + sort.Slice(result, func(i, j int) bool { + if result[i].ModelID != result[j].ModelID { + return result[i].ModelID < result[j].ModelID + } + return result[i].ProviderID < result[j].ProviderID + }) + return result +} + +func statusFromState(key RouteKey, state *routeState, now time.Time) Status { + item := Status{ModelID: key.ModelID, ProviderID: key.ProviderID, WireAPI: key.WireAPI, + Attempts: state.attempts, RecentSamples: state.recentCount, HeaderLatencyEWMA: int64(state.headerLatencyEWMA + 0.5), + ConsecutiveFailures: state.consecutiveFailures, LastStatusCode: state.lastStatusCode} + if state.recentCount > 0 { + item.AvailabilityPercent = float64(state.recentHealthy) / float64(state.recentCount) * 100 + } + if !state.lastObservedAt.IsZero() { + value := state.lastObservedAt + item.LastObservedAt = &value + } + if !state.lastHealthyAt.IsZero() { + value := state.lastHealthyAt + item.LastHealthyAt = &value + } + if state.openUntil.After(now) { + value := state.openUntil + item.CircuitOpenUntil = &value + item.State = "open" + } else if state.attempts == 0 { + item.State = "unknown" + } else if state.consecutiveFailures > 0 || (state.recentCount >= 5 && item.AvailabilityPercent < 95) { + item.State = "degraded" + } else { + item.State = "healthy" + } + return item +} -- cgit v1.2.3