Jeffrey Smith
11fd8c1e57
- go.mod module name
- All 714 import references across 289 Go files
- VERSION: 0.1.0
- CI DB names: switchboard_core_{ci,dev,test}
- Docker image: gobha/switchboard-core
- Test fixtures: JWT issuer, repo names
- .env.example, docker-compose container name
- Compiles clean (go build exit 0)
448 lines
13 KiB
Go
448 lines
13 KiB
Go
// Package health provides in-memory accumulation of provider health
|
|
// metrics with periodic flush to the database. The accumulator is
|
|
// goroutine-safe and designed for high-throughput recording from
|
|
// concurrent completion handlers.
|
|
package health
|
|
|
|
import (
|
|
"context"
|
|
"log"
|
|
"sync"
|
|
"time"
|
|
|
|
"switchboard-core/metrics"
|
|
"switchboard-core/models"
|
|
)
|
|
|
|
// ── Thresholds ──────────────────────────────
|
|
|
|
const (
|
|
// DegradedThreshold — error rate above this = degraded.
|
|
DegradedThreshold = 0.05 // 5%
|
|
// DownThreshold — error rate above this = down.
|
|
DownThreshold = 0.25 // 25%
|
|
// FlushInterval — how often in-memory counters are flushed to DB.
|
|
FlushInterval = 60 * time.Second
|
|
// PruneAge — health windows older than this are deleted.
|
|
PruneAge = 7 * 24 * time.Hour
|
|
)
|
|
|
|
// ── Store Interface ─────────────────────────
|
|
|
|
// Store is the persistence layer for health data.
|
|
type Store interface {
|
|
// UpsertWindow merges in-memory counters into the hourly bucket row.
|
|
UpsertWindow(ctx context.Context, w *models.ProviderHealthWindow) error
|
|
|
|
// GetCurrentWindow returns the current hourly bucket for a provider.
|
|
GetCurrentWindow(ctx context.Context, providerConfigID string) (*models.ProviderHealthWindow, error)
|
|
|
|
// ListWindows returns the last N hourly buckets for a provider, newest first.
|
|
ListWindows(ctx context.Context, providerConfigID string, hours int) ([]models.ProviderHealthWindow, error)
|
|
|
|
// ListAllCurrent returns the current-hour bucket for every provider.
|
|
ListAllCurrent(ctx context.Context) ([]models.ProviderHealthWindow, error)
|
|
|
|
// Prune deletes rows older than the given time.
|
|
Prune(ctx context.Context, before time.Time) (int64, error)
|
|
|
|
// ── Tool Health (v0.22.4) ──────────────
|
|
// UpsertToolWindow merges tool health counters into the hourly bucket.
|
|
UpsertToolWindow(ctx context.Context, w *models.ToolHealthWindow) error
|
|
|
|
// ListAllToolCurrent returns the current-hour bucket for every tool.
|
|
ListAllToolCurrent(ctx context.Context) ([]models.ToolHealthWindow, error)
|
|
}
|
|
|
|
// ── In-Memory Bucket ────────────────────────
|
|
|
|
// bucket holds counters for one provider within one flush interval.
|
|
type bucket struct {
|
|
providerConfigID string
|
|
requestCount int
|
|
errorCount int
|
|
timeoutCount int
|
|
rateLimitCount int
|
|
totalLatencyMs int64
|
|
maxLatencyMs int
|
|
lastError string
|
|
lastErrorAt time.Time
|
|
}
|
|
|
|
// ── Accumulator ─────────────────────────────
|
|
|
|
// Accumulator collects health metrics in memory and periodically
|
|
// flushes them to the database. One instance per process.
|
|
type Accumulator struct {
|
|
mu sync.Mutex
|
|
buckets map[string]*bucket // keyed by provider_config_id
|
|
|
|
// Tool health (v0.22.4)
|
|
toolMu sync.Mutex
|
|
toolBuckets map[string]*toolBucket // keyed by tool_name
|
|
|
|
store Store
|
|
stopCh chan struct{}
|
|
wg sync.WaitGroup
|
|
|
|
// Auto-disable (v0.22.4): if set, providers that are "down" for
|
|
// N consecutive windows get auto-deactivated.
|
|
autoDisabler AutoDisabler
|
|
autoDisableThreshold int // consecutive down windows to trigger (0 = disabled)
|
|
}
|
|
|
|
// AutoDisabler marks a provider config as inactive. Implemented by the
|
|
// provider store so the accumulator doesn't need to import the store package.
|
|
type AutoDisabler interface {
|
|
DeactivateProvider(ctx context.Context, configID string) error
|
|
}
|
|
|
|
// toolBucket holds counters for one tool within one flush interval.
|
|
type toolBucket struct {
|
|
toolName string
|
|
requestCount int
|
|
errorCount int
|
|
totalLatencyMs int64
|
|
maxLatencyMs int
|
|
lastError string
|
|
lastErrorAt time.Time
|
|
}
|
|
|
|
// NewAccumulator creates and starts the background flush loop.
|
|
func NewAccumulator(store Store) *Accumulator {
|
|
a := &Accumulator{
|
|
buckets: make(map[string]*bucket),
|
|
toolBuckets: make(map[string]*toolBucket),
|
|
store: store,
|
|
stopCh: make(chan struct{}),
|
|
}
|
|
a.wg.Add(1)
|
|
go a.flushLoop()
|
|
return a
|
|
}
|
|
|
|
// SetAutoDisable configures the auto-disable policy. When a provider
|
|
// has been "down" for N consecutive hourly windows, it is deactivated.
|
|
// N=0 disables the feature.
|
|
func (a *Accumulator) SetAutoDisable(disabler AutoDisabler, threshold int) {
|
|
a.autoDisabler = disabler
|
|
a.autoDisableThreshold = threshold
|
|
}
|
|
|
|
// RecordToolSuccess records a successful tool execution.
|
|
func (a *Accumulator) RecordToolSuccess(toolName string, latencyMs int) {
|
|
a.toolMu.Lock()
|
|
defer a.toolMu.Unlock()
|
|
b := a.getToolBucket(toolName)
|
|
b.requestCount++
|
|
b.totalLatencyMs += int64(latencyMs)
|
|
if latencyMs > b.maxLatencyMs {
|
|
b.maxLatencyMs = latencyMs
|
|
}
|
|
}
|
|
|
|
// RecordToolError records a failed tool execution.
|
|
func (a *Accumulator) RecordToolError(toolName string, latencyMs int, errMsg string) {
|
|
a.toolMu.Lock()
|
|
defer a.toolMu.Unlock()
|
|
b := a.getToolBucket(toolName)
|
|
b.requestCount++
|
|
b.errorCount++
|
|
b.totalLatencyMs += int64(latencyMs)
|
|
if latencyMs > b.maxLatencyMs {
|
|
b.maxLatencyMs = latencyMs
|
|
}
|
|
b.lastError = errMsg
|
|
b.lastErrorAt = time.Now().UTC()
|
|
}
|
|
|
|
// RecordSuccess records a successful provider call.
|
|
func (a *Accumulator) RecordSuccess(providerConfigID string, latencyMs int) {
|
|
a.mu.Lock()
|
|
defer a.mu.Unlock()
|
|
b := a.getBucket(providerConfigID)
|
|
b.requestCount++
|
|
b.totalLatencyMs += int64(latencyMs)
|
|
if latencyMs > b.maxLatencyMs {
|
|
b.maxLatencyMs = latencyMs
|
|
}
|
|
}
|
|
|
|
// RecordError records a failed provider call.
|
|
func (a *Accumulator) RecordError(providerConfigID string, latencyMs int, errMsg string) {
|
|
a.mu.Lock()
|
|
defer a.mu.Unlock()
|
|
b := a.getBucket(providerConfigID)
|
|
b.requestCount++
|
|
b.errorCount++
|
|
b.totalLatencyMs += int64(latencyMs)
|
|
if latencyMs > b.maxLatencyMs {
|
|
b.maxLatencyMs = latencyMs
|
|
}
|
|
b.lastError = errMsg
|
|
b.lastErrorAt = time.Now().UTC()
|
|
}
|
|
|
|
// RecordTimeout records a timed-out provider call (counted as both error and timeout).
|
|
func (a *Accumulator) RecordTimeout(providerConfigID string, latencyMs int, errMsg string) {
|
|
a.mu.Lock()
|
|
defer a.mu.Unlock()
|
|
b := a.getBucket(providerConfigID)
|
|
b.requestCount++
|
|
b.errorCount++
|
|
b.timeoutCount++
|
|
b.totalLatencyMs += int64(latencyMs)
|
|
if latencyMs > b.maxLatencyMs {
|
|
b.maxLatencyMs = latencyMs
|
|
}
|
|
b.lastError = errMsg
|
|
b.lastErrorAt = time.Now().UTC()
|
|
}
|
|
|
|
// RecordRateLimit records a rate-limited provider call (HTTP 429).
|
|
// Counted as both an error and a rate limit event.
|
|
func (a *Accumulator) RecordRateLimit(providerConfigID string, latencyMs int, errMsg string) {
|
|
a.mu.Lock()
|
|
defer a.mu.Unlock()
|
|
b := a.getBucket(providerConfigID)
|
|
b.requestCount++
|
|
b.errorCount++
|
|
b.rateLimitCount++
|
|
b.totalLatencyMs += int64(latencyMs)
|
|
if latencyMs > b.maxLatencyMs {
|
|
b.maxLatencyMs = latencyMs
|
|
}
|
|
b.lastError = errMsg
|
|
b.lastErrorAt = time.Now().UTC()
|
|
}
|
|
|
|
// IsRateLimitError returns true if the error message indicates an HTTP 429.
|
|
func IsRateLimitError(errMsg string) bool {
|
|
return len(errMsg) > 0 && (contains(errMsg, "HTTP 429") || contains(errMsg, "rate limit") || contains(errMsg, "too many requests"))
|
|
}
|
|
|
|
func contains(s, sub string) bool {
|
|
return len(s) >= len(sub) && (s == sub || len(s) > 0 && containsLower(s, sub))
|
|
}
|
|
|
|
func containsLower(s, sub string) bool {
|
|
// Simple case-insensitive contains for short substrings.
|
|
for i := 0; i <= len(s)-len(sub); i++ {
|
|
match := true
|
|
for j := 0; j < len(sub); j++ {
|
|
sc, tc := s[i+j], sub[j]
|
|
if sc >= 'A' && sc <= 'Z' { sc += 32 }
|
|
if tc >= 'A' && tc <= 'Z' { tc += 32 }
|
|
if sc != tc { match = false; break }
|
|
}
|
|
if match { return true }
|
|
}
|
|
return false
|
|
}
|
|
|
|
// Stop halts the flush loop and performs a final flush.
|
|
func (a *Accumulator) Stop() {
|
|
close(a.stopCh)
|
|
a.wg.Wait()
|
|
}
|
|
|
|
// DeriveStatus computes the provider status from an error rate.
|
|
func DeriveStatus(errorRate float64, requestCount int) models.ProviderStatus {
|
|
if requestCount == 0 {
|
|
return models.StatusUnknown
|
|
}
|
|
if errorRate >= DownThreshold {
|
|
return models.StatusDown
|
|
}
|
|
if errorRate >= DegradedThreshold {
|
|
return models.StatusDegraded
|
|
}
|
|
return models.StatusHealthy
|
|
}
|
|
|
|
// ── Internal ────────────────────────────────
|
|
|
|
func (a *Accumulator) getBucket(providerConfigID string) *bucket {
|
|
b, ok := a.buckets[providerConfigID]
|
|
if !ok {
|
|
b = &bucket{providerConfigID: providerConfigID}
|
|
a.buckets[providerConfigID] = b
|
|
}
|
|
return b
|
|
}
|
|
|
|
func (a *Accumulator) getToolBucket(toolName string) *toolBucket {
|
|
b, ok := a.toolBuckets[toolName]
|
|
if !ok {
|
|
b = &toolBucket{toolName: toolName}
|
|
a.toolBuckets[toolName] = b
|
|
}
|
|
return b
|
|
}
|
|
|
|
func (a *Accumulator) flushLoop() {
|
|
defer a.wg.Done()
|
|
ticker := time.NewTicker(FlushInterval)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-ticker.C:
|
|
a.flush()
|
|
a.flushTools()
|
|
a.checkAutoDisable()
|
|
case <-a.stopCh:
|
|
a.flush() // final flush
|
|
a.flushTools()
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
func (a *Accumulator) flush() {
|
|
a.mu.Lock()
|
|
if len(a.buckets) == 0 {
|
|
a.mu.Unlock()
|
|
return
|
|
}
|
|
// Swap out current buckets so we don't hold the lock during DB writes.
|
|
snap := a.buckets
|
|
a.buckets = make(map[string]*bucket, len(snap))
|
|
a.mu.Unlock()
|
|
|
|
windowStart := hourFloor(time.Now().UTC())
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
|
|
for _, b := range snap {
|
|
if b.requestCount == 0 {
|
|
continue
|
|
}
|
|
w := &models.ProviderHealthWindow{
|
|
ProviderConfigID: b.providerConfigID,
|
|
WindowStart: windowStart,
|
|
RequestCount: b.requestCount,
|
|
ErrorCount: b.errorCount,
|
|
TimeoutCount: b.timeoutCount,
|
|
RateLimitCount: b.rateLimitCount,
|
|
TotalLatencyMs: b.totalLatencyMs,
|
|
MaxLatencyMs: b.maxLatencyMs,
|
|
}
|
|
if b.lastError != "" {
|
|
w.LastError = &b.lastError
|
|
ts := b.lastErrorAt.Format(time.RFC3339)
|
|
w.LastErrorAt = &ts
|
|
}
|
|
if err := a.store.UpsertWindow(ctx, w); err != nil {
|
|
log.Printf("⚠ health: flush failed for provider %s: %v", b.providerConfigID, err)
|
|
}
|
|
|
|
// v0.33.0: Update Prometheus provider status gauge
|
|
errorRate := float64(0)
|
|
if b.requestCount > 0 {
|
|
errorRate = float64(b.errorCount) / float64(b.requestCount)
|
|
}
|
|
status := DeriveStatus(errorRate, b.requestCount)
|
|
var statusVal float64
|
|
switch status {
|
|
case models.StatusHealthy:
|
|
statusVal = 1
|
|
case models.StatusDegraded:
|
|
statusVal = 2
|
|
case models.StatusDown:
|
|
statusVal = 3
|
|
default:
|
|
statusVal = 0
|
|
}
|
|
metrics.ProviderStatus.WithLabelValues(b.providerConfigID).Set(statusVal)
|
|
}
|
|
}
|
|
|
|
// flushTools writes tool health buckets to the database.
|
|
func (a *Accumulator) flushTools() {
|
|
a.toolMu.Lock()
|
|
if len(a.toolBuckets) == 0 {
|
|
a.toolMu.Unlock()
|
|
return
|
|
}
|
|
snap := a.toolBuckets
|
|
a.toolBuckets = make(map[string]*toolBucket, len(snap))
|
|
a.toolMu.Unlock()
|
|
|
|
windowStart := hourFloor(time.Now().UTC())
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
|
|
for _, b := range snap {
|
|
if b.requestCount == 0 {
|
|
continue
|
|
}
|
|
w := &models.ToolHealthWindow{
|
|
ToolName: b.toolName,
|
|
WindowStart: windowStart,
|
|
RequestCount: b.requestCount,
|
|
ErrorCount: b.errorCount,
|
|
TotalLatencyMs: b.totalLatencyMs,
|
|
MaxLatencyMs: b.maxLatencyMs,
|
|
}
|
|
if b.lastError != "" {
|
|
w.LastError = &b.lastError
|
|
ts := b.lastErrorAt.Format(time.RFC3339)
|
|
w.LastErrorAt = &ts
|
|
}
|
|
if err := a.store.UpsertToolWindow(ctx, w); err != nil {
|
|
log.Printf("⚠ health: tool flush failed for %s: %v", b.toolName, err)
|
|
}
|
|
}
|
|
}
|
|
|
|
// checkAutoDisable inspects recent health windows and deactivates providers
|
|
// that have been "down" for too many consecutive hours.
|
|
func (a *Accumulator) checkAutoDisable() {
|
|
if a.autoDisabler == nil || a.autoDisableThreshold <= 0 {
|
|
return
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
|
|
// Get current windows for all providers
|
|
windows, err := a.store.ListAllCurrent(ctx)
|
|
if err != nil {
|
|
return
|
|
}
|
|
|
|
for _, w := range windows {
|
|
status := DeriveStatus(w.ErrorRate(), w.RequestCount)
|
|
if status != models.StatusDown {
|
|
continue
|
|
}
|
|
|
|
// Check last N windows — are they all down?
|
|
recent, err := a.store.ListWindows(ctx, w.ProviderConfigID, a.autoDisableThreshold)
|
|
if err != nil || len(recent) < a.autoDisableThreshold {
|
|
continue // not enough data
|
|
}
|
|
|
|
allDown := true
|
|
for _, rw := range recent {
|
|
if DeriveStatus(rw.ErrorRate(), rw.RequestCount) != models.StatusDown {
|
|
allDown = false
|
|
break
|
|
}
|
|
}
|
|
|
|
if allDown {
|
|
log.Printf("⚠ health: auto-disabling provider %s — down for %d consecutive windows", w.ProviderConfigID, a.autoDisableThreshold)
|
|
if err := a.autoDisabler.DeactivateProvider(ctx, w.ProviderConfigID); err != nil {
|
|
log.Printf("⚠ health: auto-disable failed for %s: %v", w.ProviderConfigID, err)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// hourFloor truncates a time to the start of its hour.
|
|
func hourFloor(t time.Time) time.Time {
|
|
return t.Truncate(time.Hour)
|
|
}
|