test: add comprehensive integration test harness for adaptive polling (Phase 2 Task 9c)
Add PollExecutor seam and integration test infrastructure: **PollExecutor Interface:** - Add pluggable executor interface for testability - Implement realExecutor wrapping existing poll functions - Add SetExecutor() for test injection - Zero impact on production behavior **Integration Test Harness:** - Build-tagged integration tests (go:build integration) - Synthetic workload generator with configurable scenarios - Fake executor simulating latencies, failures, recovery - Runtime metrics collection (queue depth, staleness, goroutines) **Comprehensive Assertions:** - Queue depth bounds: stays within 1.5× instance count - Staleness: healthy instances <20s, multiple poll cycles - Circuit breakers: transient failures recover, permanent stay blocked - Dead-letter queue: only permanent failures routed - Scheduler health: snapshot consistency validation **Test Scenarios:** - 10 healthy PVE instances (rapid polling) - 1 transient failure instance (fail → recover) - 1 permanent failure instance (DLQ routing) - 55s test duration with 3s base intervals - Validates full adaptive scheduler lifecycle Runs with: go test -tags=integration ./internal/monitoring -run TestAdaptiveSchedulerIntegration Part of Phase 2 Task 9 (Integration/Soak Testing)
This commit is contained in:
rcourtman
parent
d5c7a3494b
commit
2636ba9137
4 changed files with 945 additions and 18 deletions
217
internal/monitoring/fake_executor_integration.go
Normal file
217
internal/monitoring/fake_executor_integration.go
Normal file
|
|
@ -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)
|
||||
}
|
||||
385
internal/monitoring/harness_integration.go
Normal file
385
internal/monitoring/harness_integration.go
Normal file
|
|
@ -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 }
|
||||
190
internal/monitoring/integration_integration_test.go
Normal file
190
internal/monitoring/integration_integration_test.go
Normal file
|
|
@ -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")
|
||||
}
|
||||
}
|
||||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue