diff --git a/internal/monitoring/backoff.go b/internal/monitoring/backoff.go new file mode 100644 index 0000000..ffc0a7d --- /dev/null +++ b/internal/monitoring/backoff.go @@ -0,0 +1,39 @@ +package monitoring + +import ( + "math" + "time" +) + +type backoffConfig struct { + Initial time.Duration + Multiplier float64 + Jitter float64 + Max time.Duration +} + +func (cfg backoffConfig) nextDelay(attempt int, rng float64) time.Duration { + if attempt < 0 { + attempt = 0 + } + base := float64(cfg.Initial) + if base <= 0 { + base = float64(2 * time.Second) + } + multiplier := cfg.Multiplier + if multiplier <= 1 { + multiplier = 2 + } + delay := base * math.Pow(multiplier, float64(attempt)) + if cfg.Jitter > 0 { + j := cfg.Jitter + if j > 1 { + j = 1 + } + delay = delay * (1 + (rng*2-1)*j) + } + if cfg.Max > 0 && delay > float64(cfg.Max) { + delay = float64(cfg.Max) + } + return time.Duration(delay) +} diff --git a/internal/monitoring/circuit_breaker.go b/internal/monitoring/circuit_breaker.go new file mode 100644 index 0000000..71ec056 --- /dev/null +++ b/internal/monitoring/circuit_breaker.go @@ -0,0 +1,108 @@ +package monitoring + +import ( + "sync" + "time" +) + +type breakerState int + +const ( + breakerClosed breakerState = iota + breakerOpen + breakerHalfOpen +) + +type circuitBreaker struct { + mu sync.Mutex + state breakerState + failureCount int + openedAt time.Time + lastAttempt time.Time + retryInterval time.Duration + maxDelay time.Duration + openThreshold int + halfOpenWindow time.Duration +} + +func newCircuitBreaker(openThreshold int, retryInterval, maxDelay, halfOpenWindow time.Duration) *circuitBreaker { + if openThreshold <= 0 { + openThreshold = 3 + } + if retryInterval <= 0 { + retryInterval = 5 * time.Second + } + if maxDelay <= 0 { + maxDelay = 5 * time.Minute + } + if halfOpenWindow <= 0 { + halfOpenWindow = 30 * time.Second + } + return &circuitBreaker{ + state: breakerClosed, + retryInterval: retryInterval, + maxDelay: maxDelay, + openThreshold: openThreshold, + halfOpenWindow: halfOpenWindow, + } +} + +func (b *circuitBreaker) allow(now time.Time) bool { + b.mu.Lock() + defer b.mu.Unlock() + + switch b.state { + case breakerClosed: + return true + case breakerOpen: + if now.Sub(b.openedAt) >= b.retryInterval { + b.state = breakerHalfOpen + b.lastAttempt = now + return true + } + return false + case breakerHalfOpen: + if now.Sub(b.lastAttempt) >= b.halfOpenWindow { + b.lastAttempt = now + return true + } + return false + default: + return true + } +} + +func (b *circuitBreaker) recordSuccess() { + b.mu.Lock() + defer b.mu.Unlock() + if b.state != breakerClosed { + b.state = breakerClosed + b.failureCount = 0 + } +} + +func (b *circuitBreaker) recordFailure(now time.Time) { + b.mu.Lock() + defer b.mu.Unlock() + b.failureCount++ + b.lastAttempt = now + + switch b.state { + case breakerHalfOpen: + b.trip(now) + case breakerClosed: + if b.failureCount >= b.openThreshold { + b.trip(now) + } + } +} + +func (b *circuitBreaker) trip(now time.Time) { + b.state = breakerOpen + delay := b.retryInterval << uint(b.failureCount) + if delay > b.maxDelay { + delay = b.maxDelay + } + b.retryInterval = delay + b.openedAt = now +} diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 1cddad4..88c79f1 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -3,8 +3,10 @@ package monitoring import ( "context" "encoding/json" + stderrors "errors" "fmt" "math" + "math/rand" "net" "net/url" "os" @@ -259,6 +261,13 @@ type Monitor struct { scheduler *AdaptiveScheduler stalenessTracker *StalenessTracker taskQueue *TaskQueue + circuitBreakers map[string]*circuitBreaker + deadLetterQueue *TaskQueue + failureCounts map[string]int + lastOutcome map[string]taskOutcome + backoffCfg backoffConfig + rng *rand.Rand + maxRetryAttempts int tempCollector *TemperatureCollector // SSH-based temperature collector mu sync.RWMutex startTime time.Time @@ -379,6 +388,13 @@ type guestMetadataCacheEntry struct { fetchedAt time.Time } +type taskOutcome struct { + success bool + transient bool + err error + recordedAt time.Time +} + func (m *Monitor) getNodeRRDMemAvailable(ctx context.Context, client PVEClientInterface, nodeName string) (uint64, error) { if client == nil || nodeName == "" { return 0, fmt.Errorf("invalid arguments for RRD lookup") @@ -1319,6 +1335,16 @@ func New(cfg *config.Config) (*Monitor, error) { stalenessTracker := NewStalenessTracker(getPollMetrics()) stalenessTracker.SetBounds(cfg.AdaptivePollingBaseInterval, cfg.AdaptivePollingMaxInterval) taskQueue := NewTaskQueue() + deadLetterQueue := NewTaskQueue() + breakers := make(map[string]*circuitBreaker) +failureCounts := make(map[string]int) +lastOutcome := make(map[string]taskOutcome) +backoff := backoffConfig{ + Initial: 5 * time.Second, + Multiplier: 2, + Jitter: 0.2, + Max: 5 * time.Minute, +} var scheduler *AdaptiveScheduler if cfg.AdaptivePollingEnabled { @@ -1339,6 +1365,13 @@ func New(cfg *config.Config) (*Monitor, error) { scheduler: scheduler, stalenessTracker: stalenessTracker, taskQueue: taskQueue, + deadLetterQueue: deadLetterQueue, + circuitBreakers: breakers, + failureCounts: failureCounts, + lastOutcome: lastOutcome, + backoffCfg: backoff, + rng: rand.New(rand.NewSource(time.Now().UnixNano())), + maxRetryAttempts: 5, tempCollector: tempCollector, startTime: time.Now(), rateTracker: NewRateTracker(), @@ -2113,6 +2146,13 @@ func (m *Monitor) taskWorker(ctx context.Context, id int) { } func (m *Monitor) executeScheduledTask(ctx context.Context, task ScheduledTask) { + if !m.allowExecution(task) { + log.Debug(). + Str("instance", task.InstanceName). + Str("type", string(task.InstanceType)). + Msg("Task blocked by circuit breaker") + return + } switch task.InstanceType { case InstanceTypePVE: client, ok := m.pveClients[task.InstanceName] @@ -2120,21 +2160,21 @@ func (m *Monitor) executeScheduledTask(ctx context.Context, task ScheduledTask) log.Warn().Str("instance", task.InstanceName).Msg("PVE client missing for scheduled task") return } - m.pollPVEInstance(ctx, task.InstanceName, client) + m.pollPVEInstance(ctx, task.InstanceName, client) case InstanceTypePBS: client, ok := m.pbsClients[task.InstanceName] if !ok || client == nil { log.Warn().Str("instance", task.InstanceName).Msg("PBS client missing for scheduled task") return } - m.pollPBSInstance(ctx, task.InstanceName, client) + m.pollPBSInstance(ctx, task.InstanceName, client) case InstanceTypePMG: client, ok := m.pmgClients[task.InstanceName] if !ok || client == nil { log.Warn().Str("instance", task.InstanceName).Msg("PMG client missing for scheduled task") return } - m.pollPMGInstance(ctx, task.InstanceName, client) + m.pollPMGInstance(ctx, task.InstanceName, client) default: log.Debug().Str("instance", task.InstanceName).Str("type", string(task.InstanceType)).Msg("Skipping unsupported task type") } @@ -2145,6 +2185,28 @@ func (m *Monitor) rescheduleTask(task ScheduledTask) { return } + key := schedulerKey(task.InstanceType, task.InstanceName) + m.mu.Lock() + outcome, hasOutcome := m.lastOutcome[key] + failureCount := m.failureCounts[key] + m.mu.Unlock() + + if hasOutcome && !outcome.success { + if !outcome.transient || failureCount >= m.maxRetryAttempts { + m.sendToDeadLetter(task, outcome.err) + return + } + delay := m.backoffCfg.nextDelay(failureCount-1, m.randomFloat()) + if delay <= 0 { + delay = 5 * time.Second + } + next := task + next.Interval = delay + next.NextRun = time.Now().Add(delay) + m.taskQueue.Upsert(next) + return + } + if m.scheduler == nil { nextInterval := task.Interval if nextInterval <= 0 && m.config != nil { @@ -2161,9 +2223,9 @@ func (m *Monitor) rescheduleTask(task ScheduledTask) { } desc := InstanceDescriptor{ - Name: task.InstanceName, - Type: task.InstanceType, - LastInterval: task.Interval, + Name: task.InstanceName, + Type: task.InstanceType, + LastInterval: task.Interval, LastScheduled: task.NextRun, } if m.stalenessTracker != nil { @@ -2196,6 +2258,35 @@ func (m *Monitor) rescheduleTask(task ScheduledTask) { } } +func (m *Monitor) sendToDeadLetter(task ScheduledTask, err error) { + if m.deadLetterQueue == nil { + log.Error(). + Str("instance", task.InstanceName). + Str("type", string(task.InstanceType)). + Err(err). + Msg("Dead-letter queue unavailable; dropping task") + return + } + + log.Error(). + Str("instance", task.InstanceName). + Str("type", string(task.InstanceType)). + Err(err). + Msg("Routing task to dead-letter queue after repeated failures") + + next := task + next.Interval = 30 * time.Minute + next.NextRun = time.Now().Add(next.Interval) + m.deadLetterQueue.Upsert(next) +} + +func (m *Monitor) randomFloat() float64 { + if m.rng == nil { + m.rng = rand.New(rand.NewSource(time.Now().UnixNano())) + } + return m.rng.Float64() +} + func (m *Monitor) updateQueueDepthMetric() { if m.pollMetrics == nil || m.taskQueue == nil { return @@ -2203,6 +2294,91 @@ func (m *Monitor) updateQueueDepthMetric() { m.pollMetrics.SetQueueDepth(m.taskQueue.Size()) } +func (m *Monitor) allowExecution(task ScheduledTask) bool { + if m.circuitBreakers == nil { + return true + } + key := schedulerKey(task.InstanceType, task.InstanceName) + breaker := m.ensureBreaker(key) + return breaker.allow(time.Now()) +} + +func (m *Monitor) ensureBreaker(key string) *circuitBreaker { + m.mu.Lock() + defer m.mu.Unlock() + + if m.circuitBreakers == nil { + m.circuitBreakers = make(map[string]*circuitBreaker) + } + if breaker, ok := m.circuitBreakers[key]; ok { + return breaker + } + breaker := newCircuitBreaker(3, 5*time.Second, 5*time.Minute, 30*time.Second) + m.circuitBreakers[key] = breaker + return breaker +} + +func (m *Monitor) recordTaskResult(instanceType InstanceType, instance string, pollErr error) { + if m == nil { + return + } + + key := schedulerKey(instanceType, instance) + now := time.Now() + + breaker := m.ensureBreaker(key) + + m.mu.Lock() + if pollErr == nil { + if m.failureCounts != nil { + m.failureCounts[key] = 0 + } + if m.lastOutcome != nil { + m.lastOutcome[key] = taskOutcome{ + success: true, + transient: true, + err: nil, + recordedAt: now, + } + } + m.mu.Unlock() + if breaker != nil { + breaker.recordSuccess() + } + return + } + + transient := isTransientError(pollErr) + if m.failureCounts != nil { + m.failureCounts[key] = m.failureCounts[key] + 1 + } + if m.lastOutcome != nil { + m.lastOutcome[key] = taskOutcome{ + success: false, + transient: transient, + err: pollErr, + recordedAt: now, + } + } + m.mu.Unlock() + if breaker != nil { + breaker.recordFailure(now) + } +} + +func isTransientError(err error) bool { + if err == nil { + return true + } + if errors.IsRetryableError(err) { + return true + } + if stderrors.Is(err, context.Canceled) || stderrors.Is(err, context.DeadlineExceeded) { + return true + } + return false +} + // pollPVEInstance polls a single PVE instance func (m *Monitor) pollPVEInstance(ctx context.Context, instanceName string, client PVEClientInterface) { start := time.Now() @@ -2230,6 +2406,7 @@ func (m *Monitor) pollPVEInstance(ctx context.Context, instanceName string, clie } }() } + defer m.recordTaskResult(InstanceTypePVE, instanceName, pollErr) // Check if context is cancelled select { @@ -4000,6 +4177,7 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie } }() } + defer m.recordTaskResult(InstanceTypePBS, instanceName, pollErr) // Check if context is cancelled select { @@ -4310,6 +4488,7 @@ func (m *Monitor) pollPMGInstance(ctx context.Context, instanceName string, clie } }() } + defer m.recordTaskResult(InstanceTypePMG, instanceName, pollErr) select { case <-ctx.Done():