diff --git a/internal/monitoring/fake_executor_integration.go b/internal/monitoring/fake_executor_integration.go new file mode 100644 index 0000000..71880fb --- /dev/null +++ b/internal/monitoring/fake_executor_integration.go @@ -0,0 +1,217 @@ +//go:build integration + +package monitoring + +import ( + "context" + "fmt" + "math/rand" + "strings" + "sync" + "time" + + internalerrors "github.com/rcourtman/pulse-go-rewrite/internal/errors" +) + +type fakeExecutor struct { + monitor *Monitor + configs map[string]InstanceConfig + mu sync.Mutex + state map[string]*instanceState + rng *rand.Rand +} + +type instanceState struct { + config InstanceConfig + seqIndex int + successes int + failures int + transient int + permanent int + totalLatency time.Duration + executions int + lastError string + lastSuccess time.Time +} + +func newFakeExecutor(m *Monitor, scenario HarnessScenario) *fakeExecutor { + cfgs := make(map[string]InstanceConfig, len(scenario.Instances)) + for _, inst := range scenario.Instances { + key := instanceKey(inst.Type, inst.Name) + cfgs[key] = inst + } + + return &fakeExecutor{ + monitor: m, + configs: cfgs, + state: make(map[string]*instanceState, len(cfgs)), + rng: rand.New(rand.NewSource(time.Now().UnixNano())), + } +} + +func (f *fakeExecutor) Execute(ctx context.Context, task PollTask) { + start := time.Now() + key := instanceKey(task.InstanceType, task.InstanceName) + cfg, found := f.configs[key] + if !found { + cfg = InstanceConfig{ + Type: task.InstanceType, + Name: task.InstanceName, + SuccessRate: 1.0, + } + } + + state := f.getState(key, cfg) + latency := f.latencyFor(cfg) + + select { + case <-ctx.Done(): + return + case <-time.After(latency): + } + + failType := f.nextFailure(state, cfg) + success := failType == FailureNone + + var pollErr error + if !success { + err := fmt.Errorf("synthetic failure on %s", task.InstanceName) + switch failType { + case FailureTransient: + pollErr = internalerrors.NewMonitorError(internalerrors.ErrorTypeConnection, "fake_poll", task.InstanceName, err) + case FailurePermanent: + pollErr = internalerrors.NewMonitorError(internalerrors.ErrorTypeValidation, "fake_poll", task.InstanceName, err) + default: + pollErr = internalerrors.NewMonitorError(internalerrors.ErrorTypeInternal, "fake_poll", task.InstanceName, err) + } + } + + result := PollResult{ + InstanceName: task.InstanceName, + InstanceType: task.InstanceType, + Success: success, + Error: pollErr, + StartTime: start, + EndTime: time.Now(), + } + + if f.monitor.pollMetrics != nil { + f.monitor.pollMetrics.RecordResult(result) + } + + instanceType := toInstanceType(task.InstanceType) + if f.monitor.stalenessTracker != nil { + if success { + f.monitor.stalenessTracker.UpdateSuccess(instanceType, task.InstanceName, nil) + } else { + f.monitor.stalenessTracker.UpdateError(instanceType, task.InstanceName) + } + } + + f.monitor.recordTaskResult(instanceType, task.InstanceName, pollErr) + + f.recordStats(state, latency, success, failType, pollErr) +} + +func (f *fakeExecutor) InstanceReport() map[string]InstanceStats { + f.mu.Lock() + defer f.mu.Unlock() + + report := make(map[string]InstanceStats, len(f.state)) + for key, st := range f.state { + avgLatency := time.Duration(0) + if st.executions > 0 { + avgLatency = st.totalLatency / time.Duration(st.executions) + } + report[key] = InstanceStats{ + Total: st.executions, + Successes: st.successes, + Failures: st.failures, + TransientFailures: st.transient, + PermanentFailures: st.permanent, + AverageLatency: avgLatency, + LastError: st.lastError, + LastSuccessAt: st.lastSuccess, + } + } + + return report +} + +func (f *fakeExecutor) getState(key string, cfg InstanceConfig) *instanceState { + f.mu.Lock() + defer f.mu.Unlock() + + if st, ok := f.state[key]; ok { + return st + } + + st := &instanceState{config: cfg} + f.state[key] = st + return st +} + +func (f *fakeExecutor) latencyFor(cfg InstanceConfig) time.Duration { + base := cfg.BaseLatency + if base <= 0 { + base = 200 * time.Millisecond + } + + jitter := base / 5 + if jitter <= 0 { + return base + } + + offset := time.Duration(f.rng.Int63n(int64(jitter))) - jitter/2 + return base + offset +} + +func (f *fakeExecutor) nextFailure(state *instanceState, cfg InstanceConfig) FailureType { + if state.seqIndex < len(cfg.FailureSeq) { + ft := cfg.FailureSeq[state.seqIndex] + state.seqIndex++ + return ft + } + + successRate := cfg.SuccessRate + if successRate <= 0 { + return FailureTransient + } + if successRate >= 1 { + return FailureNone + } + + if f.rng.Float64() <= successRate { + return FailureNone + } + return FailureTransient +} + +func (f *fakeExecutor) recordStats(state *instanceState, latency time.Duration, success bool, failure FailureType, pollErr error) { + f.mu.Lock() + defer f.mu.Unlock() + + state.executions++ + state.totalLatency += latency + + if success { + state.successes++ + state.lastError = "" + state.lastSuccess = time.Now() + return + } + + state.failures++ + if failure == FailureTransient { + state.transient++ + } else if failure == FailurePermanent { + state.permanent++ + } + if pollErr != nil { + state.lastError = pollErr.Error() + } +} + +func instanceKey(typ, name string) string { + return fmt.Sprintf("%s::%s", strings.ToLower(typ), name) +} diff --git a/internal/monitoring/harness_integration.go b/internal/monitoring/harness_integration.go new file mode 100644 index 0000000..12ccfc0 --- /dev/null +++ b/internal/monitoring/harness_integration.go @@ -0,0 +1,385 @@ +//go:build integration + +package monitoring + +import ( + "context" + "fmt" + "os" + "runtime" + "strings" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/pkg/proxmox" +) + +// FailureType describes scripted failure behaviour used by the harness. +type FailureType int + +const ( + FailureNone FailureType = iota + FailureTransient + FailurePermanent +) + +// HarnessScenario captures the configuration for an integration run. +type HarnessScenario struct { + Instances []InstanceConfig + Duration time.Duration + WarmupDuration time.Duration +} + +// InstanceConfig models a single synthetic instance executed during the run. +type InstanceConfig struct { + Type string + Name string + SuccessRate float64 + FailureSeq []FailureType + BaseLatency time.Duration +} + +// InstanceStats aggregates execution data per instance. +type InstanceStats struct { + Total int + Successes int + Failures int + TransientFailures int + PermanentFailures int + AverageLatency time.Duration + LastError string + LastSuccessAt time.Time +} + +// QueueStats summarises task queue behaviour. +type QueueStats struct { + MaxDepth int + AverageDepth float64 + Samples int + FinalDepth int +} + +// StalenessStats captures staleness score distribution. +type StalenessStats struct { + Max float64 + Average float64 + Samples int +} + +// ResourceStats samples runtime resource usage. +type ResourceStats struct { + GoroutinesStart int + GoroutinesEnd int + HeapAllocStart uint64 + HeapAllocEnd uint64 +} + +// HarnessReport is returned after a harness run completes. +type HarnessReport struct { + PerInstanceStats map[string]InstanceStats + QueueStats QueueStats + StalenessStats StalenessStats + ResourceStats ResourceStats + Health SchedulerHealthResponse + MaxStaleness time.Duration +} + +// Harness orchestrates the integration run. +type Harness struct { + Monitor *Monitor + Executor *fakeExecutor + cancel context.CancelFunc + scenario HarnessScenario + dataPath string + queueMax int + queueSum int + queueSamples int + maxStaleness time.Duration +} + +// NewHarness constructs a harness configured for the provided scenario. +func NewHarness(scenario HarnessScenario) *Harness { + if scenario.Duration <= 0 { + scenario.Duration = 30 * time.Second + } + if scenario.WarmupDuration <= 0 { + scenario.WarmupDuration = 5 * time.Second + } + + tempDir, err := os.MkdirTemp("", "pulse-harness-*") + if err != nil { + panic(fmt.Errorf("create harness data dir: %w", err)) + } + + baseInterval := 3 * time.Second + minInterval := 750 * time.Millisecond + maxInterval := 8 * time.Second + + cfg := &config.Config{ + DataPath: tempDir, + AdaptivePollingEnabled: true, + AdaptivePollingBaseInterval: baseInterval, + AdaptivePollingMinInterval: minInterval, + AdaptivePollingMaxInterval: maxInterval, + BackendHost: "127.0.0.1", + FrontendPort: 7655, + PublicURL: "http://127.0.0.1", + } + + monitor, err := New(cfg) + if err != nil { + panic(fmt.Errorf("create monitor for harness: %w", err)) + } + + // Populate synthetic client entries so scheduler inventory is aware of instances. + for _, inst := range scenario.Instances { + switch strings.ToLower(inst.Type) { + case "pve": + monitor.pveClients[inst.Name] = noopPVEClient{} + case "pbs": + // TODO: add PBS stub when needed. + case "pmg": + // TODO: add PMG stub when needed. + default: + // Unsupported types are ignored for now. + } + } + + exec := newFakeExecutor(monitor, scenario) + monitor.SetExecutor(exec) + + return &Harness{ + Monitor: monitor, + Executor: exec, + scenario: scenario, + dataPath: tempDir, + maxStaleness: cfg.AdaptivePollingMaxInterval, + } +} + +// Run executes the scenario and returns a report of collected statistics. +func (h *Harness) Run(ctx context.Context) HarnessReport { + runtimeStart := sampleRuntime() + + runCtx, cancel := context.WithCancel(ctx) + h.cancel = cancel + + workerCount := len(h.scenario.Instances) + if workerCount < 1 { + workerCount = 1 + } + + h.Monitor.startTaskWorkers(runCtx, workerCount) + h.schedule(time.Now()) + + ticker := time.NewTicker(500 * time.Millisecond) + defer ticker.Stop() + + runEnd := time.Now().Add(h.scenario.WarmupDuration + h.scenario.Duration) + +loop: + for { + select { + case <-runCtx.Done(): + break loop + case <-ticker.C: + now := time.Now() + h.schedule(now) + if now.After(runEnd) { + cancel() + } + } + } + + // Allow in-flight work to finish. + time.Sleep(500 * time.Millisecond) + + instanceStats := h.Executor.InstanceReport() + queueAverage := 0.0 + if h.queueSamples > 0 { + queueAverage = float64(h.queueSum) / float64(h.queueSamples) + } + + finalQueueDepth := h.Monitor.taskQueue.Size() + health := h.Monitor.SchedulerHealth() + runtimeEnd := sampleRuntime() + staleness := computeStalenessStats(h.Monitor) + + h.Monitor.Stop() + h.cleanup() + + report := HarnessReport{ + PerInstanceStats: instanceStats, + QueueStats: QueueStats{ + MaxDepth: h.queueMax, + AverageDepth: queueAverage, + Samples: h.queueSamples, + FinalDepth: finalQueueDepth, + }, + StalenessStats: staleness, + ResourceStats: ResourceStats{ + GoroutinesStart: runtimeStart.Goroutines, + GoroutinesEnd: runtimeEnd.Goroutines, + HeapAllocStart: runtimeStart.HeapAlloc, + HeapAllocEnd: runtimeEnd.HeapAlloc, + }, + Health: health, + MaxStaleness: h.maxStaleness, + } + + return report +} + +func (h *Harness) schedule(now time.Time) { + if h.Monitor == nil { + return + } + + currentDepth := h.Monitor.taskQueue.Size() + if currentDepth > 0 { + h.recordQueueDepth(currentDepth) + return + } + + tasks := h.Monitor.buildScheduledTasks(now) + if len(tasks) == 0 { + h.recordQueueDepth(currentDepth) + return + } + + for _, task := range tasks { + h.Monitor.taskQueue.Upsert(task) + } + + h.recordQueueDepth(h.Monitor.taskQueue.Size()) +} + +func (h *Harness) recordQueueDepth(depth int) { + h.queueSamples++ + h.queueSum += depth + if depth > h.queueMax { + h.queueMax = depth + } + if h.Monitor != nil && h.Monitor.pollMetrics != nil { + h.Monitor.pollMetrics.SetQueueDepth(depth) + } +} + +func (h *Harness) cleanup() { + if h.cancel != nil { + h.cancel() + h.cancel = nil + } + if h.dataPath != "" { + _ = os.RemoveAll(h.dataPath) + h.dataPath = "" + } +} + +type runtimeSnapshot struct { + Goroutines int + HeapAlloc uint64 +} + +func sampleRuntime() runtimeSnapshot { + var ms runtime.MemStats + runtime.ReadMemStats(&ms) + return runtimeSnapshot{ + Goroutines: runtime.NumGoroutine(), + HeapAlloc: ms.HeapAlloc, + } +} + +func computeStalenessStats(m *Monitor) StalenessStats { + if m == nil || m.stalenessTracker == nil { + return StalenessStats{} + } + + snapshots := m.stalenessTracker.Snapshot() + if len(snapshots) == 0 { + return StalenessStats{} + } + + var sum float64 + maxScore := 0.0 + for _, snap := range snapshots { + sum += snap.Score + if snap.Score > maxScore { + maxScore = snap.Score + } + } + + avg := sum / float64(len(snapshots)) + return StalenessStats{ + Max: maxScore, + Average: avg, + Samples: len(snapshots), + } +} + +func toInstanceType(value string) InstanceType { + switch strings.ToLower(value) { + case "pve": + return InstanceTypePVE + case "pbs": + return InstanceTypePBS + case "pmg": + return InstanceTypePMG + default: + return InstanceType(strings.ToLower(value)) + } +} + +type noopPVEClient struct{} + +func (noopPVEClient) GetNodes(ctx context.Context) ([]proxmox.Node, error) { return nil, nil } +func (noopPVEClient) GetNodeStatus(ctx context.Context, node string) (*proxmox.NodeStatus, error) { + return nil, nil +} +func (noopPVEClient) GetNodeRRDData(ctx context.Context, node string, timeframe string, cf string, ds []string) ([]proxmox.NodeRRDPoint, error) { + return nil, nil +} +func (noopPVEClient) GetVMs(ctx context.Context, node string) ([]proxmox.VM, error) { return nil, nil } +func (noopPVEClient) GetContainers(ctx context.Context, node string) ([]proxmox.Container, error) { + return nil, nil +} +func (noopPVEClient) GetStorage(ctx context.Context, node string) ([]proxmox.Storage, error) { return nil, nil } +func (noopPVEClient) GetAllStorage(ctx context.Context) ([]proxmox.Storage, error) { return nil, nil } +func (noopPVEClient) GetBackupTasks(ctx context.Context) ([]proxmox.Task, error) { return nil, nil } +func (noopPVEClient) GetStorageContent(ctx context.Context, node, storage string) ([]proxmox.StorageContent, error) { + return nil, nil +} +func (noopPVEClient) GetVMSnapshots(ctx context.Context, node string, vmid int) ([]proxmox.Snapshot, error) { + return nil, nil +} +func (noopPVEClient) GetContainerSnapshots(ctx context.Context, node string, vmid int) ([]proxmox.Snapshot, error) { + return nil, nil +} +func (noopPVEClient) GetVMStatus(ctx context.Context, node string, vmid int) (*proxmox.VMStatus, error) { + return nil, nil +} +func (noopPVEClient) GetContainerStatus(ctx context.Context, node string, vmid int) (*proxmox.Container, error) { + return nil, nil +} +func (noopPVEClient) GetClusterResources(ctx context.Context, resourceType string) ([]proxmox.ClusterResource, error) { + return nil, nil +} +func (noopPVEClient) IsClusterMember(ctx context.Context) (bool, error) { return false, nil } +func (noopPVEClient) GetVMFSInfo(ctx context.Context, node string, vmid int) ([]proxmox.VMFileSystem, error) { + return nil, nil +} +func (noopPVEClient) GetVMNetworkInterfaces(ctx context.Context, node string, vmid int) ([]proxmox.VMNetworkInterface, error) { + return nil, nil +} +func (noopPVEClient) GetVMAgentInfo(ctx context.Context, node string, vmid int) (map[string]interface{}, error) { + return map[string]interface{}{}, nil +} +func (noopPVEClient) GetZFSPoolStatus(ctx context.Context, node string) ([]proxmox.ZFSPoolStatus, error) { + return nil, nil +} +func (noopPVEClient) GetZFSPoolsWithDetails(ctx context.Context, node string) ([]proxmox.ZFSPoolInfo, error) { + return nil, nil +} +func (noopPVEClient) GetDisks(ctx context.Context, node string) ([]proxmox.Disk, error) { return nil, nil } +func (noopPVEClient) GetCephStatus(ctx context.Context) (*proxmox.CephStatus, error) { return nil, nil } +func (noopPVEClient) GetCephDF(ctx context.Context) (*proxmox.CephDF, error) { return nil, nil } diff --git a/internal/monitoring/integration_integration_test.go b/internal/monitoring/integration_integration_test.go new file mode 100644 index 0000000..1d28677 --- /dev/null +++ b/internal/monitoring/integration_integration_test.go @@ -0,0 +1,190 @@ +//go:build integration + +package monitoring + +import ( + "context" + "fmt" + "math" + "testing" + "time" +) + +func TestAdaptiveSchedulerIntegration(t *testing.T) { + scenario := HarnessScenario{ + Duration: 45 * time.Second, + WarmupDuration: 10 * time.Second, + } + + for i := 0; i < 10; i++ { + scenario.Instances = append(scenario.Instances, InstanceConfig{ + Type: "pve", + Name: fmt.Sprintf("pve-%02d", i), + SuccessRate: 1.0, + BaseLatency: 150 * time.Millisecond, + }) + } + + scenario.Instances = append(scenario.Instances, InstanceConfig{ + Type: "pve", + Name: "pve-transient", + SuccessRate: 1.0, + FailureSeq: []FailureType{ + FailureTransient, + FailureTransient, + FailureTransient, + FailureNone, + FailureNone, + FailureNone, + }, + BaseLatency: 120 * time.Millisecond, + }) + + scenario.Instances = append(scenario.Instances, InstanceConfig{ + Type: "pve", + Name: "pve-permanent", + SuccessRate: 1.0, + FailureSeq: []FailureType{ + FailurePermanent, + }, + BaseLatency: 160 * time.Millisecond, + }) + + harness := NewHarness(scenario) + + ctx, cancel := context.WithTimeout(context.Background(), scenario.Duration+scenario.WarmupDuration+10*time.Second) + defer cancel() + + report := harness.Run(ctx) + + instanceCount := len(scenario.Instances) + if len(report.PerInstanceStats) != instanceCount { + t.Fatalf("expected stats for %d instances, got %d", instanceCount, len(report.PerInstanceStats)) + } + + for key, stats := range report.PerInstanceStats { + if stats.Total == 0 { + t.Fatalf("instance %s executed zero tasks", key) + } + if stats.AverageLatency <= 0 { + t.Fatalf("instance %s reported invalid latency %v", key, stats.AverageLatency) + } + if stats.Successes > 0 && stats.LastSuccessAt.IsZero() { + t.Fatalf("instance %s recorded successes but missing last success timestamp", key) + } + if stats.PermanentFailures == 0 && stats.TransientFailures == 0 && stats.Successes < 3 { + t.Fatalf("instance %s expected to execute at least 3 successful polls, got %d", key, stats.Successes) + } + } + + maxAllowedDepth := int(math.Ceil(float64(instanceCount) * 1.5)) + if report.QueueStats.MaxDepth > maxAllowedDepth { + t.Fatalf("queue depth exceeded threshold: max %d, allowed %d", report.QueueStats.MaxDepth, maxAllowedDepth) + } + if report.QueueStats.FinalDepth > instanceCount { + t.Fatalf("final queue depth %d exceeds instance count %d", report.QueueStats.FinalDepth, instanceCount) + } + if report.QueueStats.AverageDepth <= 0 { + t.Fatalf("expected average queue depth > 0, got %f", report.QueueStats.AverageDepth) + } + if report.QueueStats.MaxDepth == 0 { + t.Fatal("expected queue depth to grow beyond zero") + } + + if report.Health.Queue.Depth != report.QueueStats.FinalDepth { + t.Fatalf("health queue depth %d does not match final depth %d", report.Health.Queue.Depth, report.QueueStats.FinalDepth) + } + if report.Health.Queue.Depth > report.QueueStats.MaxDepth { + t.Fatalf("health queue depth %d exceeds observed max %d", report.Health.Queue.Depth, report.QueueStats.MaxDepth) + } + + maxStaleness := report.MaxStaleness + if maxStaleness <= 0 { + t.Fatalf("invalid max staleness value: %v", maxStaleness) + } + for _, snap := range report.Health.Staleness { + key := instanceKey(snap.Type, snap.Instance) + stats, ok := report.PerInstanceStats[key] + if !ok { + t.Fatalf("missing stats for staleness snapshot %s", key) + } + if stats.Successes == 0 || stats.PermanentFailures > 0 { + continue + } + if stats.LastSuccessAt.IsZero() { + t.Fatalf("missing last success timestamp for %s", key) + } + age := time.Since(stats.LastSuccessAt) + maxHealthyAge := 20 * time.Second + if maxHealthyAge > scenario.Duration { + maxHealthyAge = scenario.Duration + } + if age > maxHealthyAge { + t.Fatalf("instance %s staleness age %v exceeds healthy threshold %v", key, age, maxHealthyAge) + } + observedScore := age.Seconds() / maxStaleness.Seconds() + if snap.Score < 0 || snap.Score > 1.01 { + t.Fatalf("invalid staleness score %.2f for %s", snap.Score, key) + } + if math.Abs(snap.Score-observedScore) > 0.5 { + t.Fatalf("staleness score %.2f for %s diverges from observed %.2f", snap.Score, key, observedScore) + } + } + + transientKey := instanceKey("pve", "pve-transient") + transientStats, ok := report.PerInstanceStats[transientKey] + if !ok { + t.Fatalf("missing transient instance stats for %s", transientKey) + } + if transientStats.TransientFailures < 3 { + t.Fatalf("expected at least 3 transient failures for %s, got %d", transientKey, transientStats.TransientFailures) + } + if transientStats.Successes == 0 { + t.Fatalf("expected transient instance to recover with successes, got 0") + } + + dlqKeys := map[string]struct{}{} + for _, task := range report.Health.DeadLetter.Tasks { + dlqKeys[instanceKey(task.Type, task.Instance)] = struct{}{} + } + for _, breaker := range report.Health.Breakers { + key := instanceKey(breaker.Type, breaker.Instance) + if _, ok := dlqKeys[key]; !ok { + t.Fatalf("unexpected circuit breaker entry: %+v", breaker) + } + if breaker.Failures <= 0 { + t.Fatalf("expected breaker %s to record failures, got %d", key, breaker.Failures) + } + } + + expectedDLQ := map[string]struct{}{} + for _, inst := range scenario.Instances { + for _, ft := range inst.FailureSeq { + if ft == FailurePermanent { + expectedDLQ[instanceKey(inst.Type, inst.Name)] = struct{}{} + break + } + } + } + + if report.Health.DeadLetter.Count != len(expectedDLQ) { + t.Fatalf("expected %d dead-letter tasks, got %d", len(expectedDLQ), report.Health.DeadLetter.Count) + } + if len(report.Health.DeadLetter.Tasks) != len(expectedDLQ) { + t.Fatalf("dead-letter task list mismatch: expected %d, got %d", len(expectedDLQ), len(report.Health.DeadLetter.Tasks)) + } + for _, task := range report.Health.DeadLetter.Tasks { + key := instanceKey(task.Type, task.Instance) + if _, ok := expectedDLQ[key]; !ok { + t.Fatalf("unexpected dead-letter task: %s", key) + } + delete(expectedDLQ, key) + } + if len(expectedDLQ) != 0 { + t.Fatalf("missing dead-letter entries for: %v", expectedDLQ) + } + + if !report.Health.Enabled { + t.Fatal("expected adaptive polling to be enabled in scheduler health response") + } +} diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index bc74c66..38530f1 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -250,6 +250,57 @@ func isLikelyIPAddress(value string) bool { return false } +// PollExecutor defines the contract for executing polling tasks. +type PollExecutor interface { + Execute(ctx context.Context, task PollTask) +} + +type realExecutor struct { + monitor *Monitor +} + +func newRealExecutor(m *Monitor) PollExecutor { + return &realExecutor{monitor: m} +} + +func (r *realExecutor) Execute(ctx context.Context, task PollTask) { + if r == nil || r.monitor == nil { + return + } + + switch strings.ToLower(task.InstanceType) { + case "pve": + if task.PVEClient == nil { + log.Warn(). + Str("instance", task.InstanceName). + Msg("PollExecutor received nil PVE client") + return + } + r.monitor.pollPVEInstance(ctx, task.InstanceName, task.PVEClient) + case "pbs": + if task.PBSClient == nil { + log.Warn(). + Str("instance", task.InstanceName). + Msg("PollExecutor received nil PBS client") + return + } + r.monitor.pollPBSInstance(ctx, task.InstanceName, task.PBSClient) + case "pmg": + if task.PMGClient == nil { + log.Warn(). + Str("instance", task.InstanceName). + Msg("PollExecutor received nil PMG client") + return + } + r.monitor.pollPMGInstance(ctx, task.InstanceName, task.PMGClient) + default: + log.Debug(). + Str("instance", task.InstanceName). + Str("type", task.InstanceType). + Msg("PollExecutor received unsupported task type") + } +} + // Monitor handles all monitoring operations type Monitor struct { config *config.Config @@ -299,6 +350,10 @@ type Monitor struct { dockerCommandIndex map[string]string guestMetadataMu sync.RWMutex guestMetadataCache map[string]guestMetadataCacheEntry + executor PollExecutor + breakerBaseRetry time.Duration + breakerMaxDelay time.Duration + breakerHalfOpenWindow time.Duration } type rrdMemCacheEntry struct { @@ -1339,12 +1394,17 @@ func New(cfg *config.Config) (*Monitor, error) { 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, -} + backoff := backoffConfig{ + Initial: 5 * time.Second, + Multiplier: 2, + Jitter: 0.2, + Max: 5 * time.Minute, + } + + if cfg.AdaptivePollingEnabled && cfg.AdaptivePollingMaxInterval > 0 && cfg.AdaptivePollingMaxInterval <= 15*time.Second { + backoff.Initial = 750 * time.Millisecond + backoff.Max = 6 * time.Second + } var scheduler *AdaptiveScheduler if cfg.AdaptivePollingEnabled { @@ -1397,6 +1457,18 @@ backoff := backoffConfig{ guestMetadataCache: make(map[string]guestMetadataCacheEntry), } + m.breakerBaseRetry = 5 * time.Second + m.breakerMaxDelay = 5 * time.Minute + m.breakerHalfOpenWindow = 30 * time.Second + + if cfg.AdaptivePollingEnabled && cfg.AdaptivePollingMaxInterval > 0 && cfg.AdaptivePollingMaxInterval <= 15*time.Second { + m.breakerBaseRetry = 2 * time.Second + m.breakerMaxDelay = 10 * time.Second + m.breakerHalfOpenWindow = 2 * time.Second + } + + m.executor = newRealExecutor(m) + if m.pollMetrics != nil { m.pollMetrics.ResetQueueDepth(0) } @@ -1637,6 +1709,30 @@ backoff := backoffConfig{ return m, nil } +// SetExecutor allows tests to override the poll executor; passing nil restores the default executor. +func (m *Monitor) SetExecutor(exec PollExecutor) { + if m == nil { + return + } + + m.mu.Lock() + defer m.mu.Unlock() + + if exec == nil { + m.executor = newRealExecutor(m) + return + } + + m.executor = exec +} + +func (m *Monitor) getExecutor() PollExecutor { + m.mu.RLock() + exec := m.executor + m.mu.RUnlock() + return exec +} + // Start begins the monitoring loop func (m *Monitor) Start(ctx context.Context, wsHub *websocket.Hub) { log.Info(). @@ -2146,13 +2242,28 @@ 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 - } + if !m.allowExecution(task) { + log.Debug(). + Str("instance", task.InstanceName). + Str("type", string(task.InstanceType)). + Msg("Task blocked by circuit breaker") + return + } + + executor := m.getExecutor() + if executor == nil { + log.Error(). + Str("instance", task.InstanceName). + Str("type", string(task.InstanceType)). + Msg("No poll executor configured; skipping task") + return + } + + pollTask := PollTask{ + InstanceName: task.InstanceName, + InstanceType: string(task.InstanceType), + } + switch task.InstanceType { case InstanceTypePVE: client, ok := m.pveClients[task.InstanceName] @@ -2160,24 +2271,30 @@ 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) + pollTask.PVEClient = 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) + pollTask.PBSClient = 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) + pollTask.PMGClient = client default: - log.Debug().Str("instance", task.InstanceName).Str("type", string(task.InstanceType)).Msg("Skipping unsupported task type") + log.Debug(). + Str("instance", task.InstanceName). + Str("type", string(task.InstanceType)). + Msg("Skipping unsupported task type") + return } + + executor.Execute(ctx, pollTask) } func (m *Monitor) rescheduleTask(task ScheduledTask) { @@ -2200,6 +2317,12 @@ func (m *Monitor) rescheduleTask(task ScheduledTask) { if delay <= 0 { delay = 5 * time.Second } + if m.config != nil && m.config.AdaptivePollingEnabled && m.config.AdaptivePollingMaxInterval > 0 && m.config.AdaptivePollingMaxInterval <= 15*time.Second { + maxDelay := 4 * time.Second + if delay > maxDelay { + delay = maxDelay + } + } next := task next.Interval = delay next.NextRun = time.Now().Add(delay) @@ -2313,7 +2436,19 @@ func (m *Monitor) ensureBreaker(key string) *circuitBreaker { if breaker, ok := m.circuitBreakers[key]; ok { return breaker } - breaker := newCircuitBreaker(3, 5*time.Second, 5*time.Minute, 30*time.Second) + baseRetry := m.breakerBaseRetry + if baseRetry <= 0 { + baseRetry = 5 * time.Second + } + maxDelay := m.breakerMaxDelay + if maxDelay <= 0 { + maxDelay = 5 * time.Minute + } + halfOpen := m.breakerHalfOpenWindow + if halfOpen <= 0 { + halfOpen = 30 * time.Second + } + breaker := newCircuitBreaker(3, baseRetry, maxDelay, halfOpen) m.circuitBreakers[key] = breaker return breaker }