From 52489e89263fbf8969b1cd619451b9d1d485d5c3 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Tue, 11 Aug 2026 14:05:06 +0100 Subject: [PATCH] Scale connection stale cutoff by the adaptive planned poll interval Adaptive polling deliberately stretches an instance's cadence toward its max interval (5 minutes by default) while data is fresh, but the connections aggregator judged staleness against the configured cadence with a 2-minute floor. Any adaptive-enabled install therefore cycled healthy PVE/PBS/PMG connections into stale for the back half of every stretched poll gap: the Infrastructure page dropped the source badge from API + Agent to Agent and connection-degraded alerts fired against a schedule the poller was honoring. The aggregator now scales the active-to-stale cutoff by the scheduler's currently planned interval when that exceeds the configured cadence, via Monitor.PlannedPollInterval and per-instance planned intervals in the aggregator inputs. A plan tighter than the configured cadence never tightens the cutoff, so genuine poll outages still trip the floor on time. Connection-degraded alerts and the runtime inventory gate consume the same derived state and inherit the fix. Refs #1437 Contract-Neutral: behavioral fix: stale cutoff follows adaptive planned interval (#1437), no public contract delta --- internal/api/connections_aggregator.go | 30 +++++++++- ...nections_aggregator_adaptive_stale_test.go | 58 +++++++++++++++++++ internal/api/connections_alerts.go | 27 +++++++++ internal/monitoring/monitor.go | 16 +++++ .../monitor_planned_interval_test.go | 40 +++++++++++++ 5 files changed, 168 insertions(+), 3 deletions(-) create mode 100644 internal/api/connections_aggregator_adaptive_stale_test.go create mode 100644 internal/monitoring/monitor_planned_interval_test.go diff --git a/internal/api/connections_aggregator.go b/internal/api/connections_aggregator.go index f119b5b95..8bc7423c2 100644 --- a/internal/api/connections_aggregator.go +++ b/internal/api/connections_aggregator.go @@ -139,6 +139,27 @@ type aggregatorInputs struct { pvePollingInterval time.Duration pbsPollingInterval time.Duration pmgPollingInterval time.Duration + + // plannedPollIntervals maps instanceHealth keys ("pve::") to the + // adaptive scheduler's currently planned interval for that instance. + // Adaptive polling deliberately stretches cadence past the configured + // interval while data is fresh, so the active→stale cutoff must follow + // the schedule actually in force or a healthy connection reads as stale + // for the back half of every stretched cycle (#1437). Absent or zero + // entries fall back to the configured cadence. + plannedPollIntervals map[string]time.Duration +} + +// effectivePollInterval picks the cadence the stale cutoff should scale by: +// the adaptive scheduler's planned interval when it stretched beyond the +// configured one, otherwise the configured interval. A planned interval that +// shrank below the configured cadence never tightens the cutoff, so a genuine +// poll outage still trips the connectionStaleThreshold floor on time. +func effectivePollInterval(configured, planned time.Duration) time.Duration { + if planned > configured { + return planned + } + return configured } type connectionAgentDesiredConfig struct { @@ -160,13 +181,16 @@ func buildConnections(in aggregatorInputs) []Connection { len(in.vmwareInstances)+len(in.truenasInstances)+len(in.availabilityTargets)+len(in.hosts)) for _, pve := range in.pveInstances { - out = append(out, buildPVEConnection(pve, in.instanceHealth, now, in.pvePollingInterval)) + interval := effectivePollInterval(in.pvePollingInterval, in.plannedPollIntervals["pve::"+pve.Name]) + out = append(out, buildPVEConnection(pve, in.instanceHealth, now, interval)) } for _, pbs := range in.pbsInstances { - out = append(out, buildPBSConnection(pbs, in.instanceHealth, now, in.pbsPollingInterval, in.pbsReportedNodeNames[pbs.Name])) + interval := effectivePollInterval(in.pbsPollingInterval, in.plannedPollIntervals["pbs::"+pbs.Name]) + out = append(out, buildPBSConnection(pbs, in.instanceHealth, now, interval, in.pbsReportedNodeNames[pbs.Name])) } for _, pmg := range in.pmgInstances { - out = append(out, buildPMGConnection(pmg, in.instanceHealth, now, in.pmgPollingInterval)) + interval := effectivePollInterval(in.pmgPollingInterval, in.plannedPollIntervals["pmg::"+pmg.Name]) + out = append(out, buildPMGConnection(pmg, in.instanceHealth, now, interval)) } for _, vmw := range in.vmwareInstances { out = append(out, buildVMwareConnection(vmw, in.instanceHealth, in.vmwareSummaries, now)) diff --git a/internal/api/connections_aggregator_adaptive_stale_test.go b/internal/api/connections_aggregator_adaptive_stale_test.go new file mode 100644 index 000000000..8a89ec5a7 --- /dev/null +++ b/internal/api/connections_aggregator_adaptive_stale_test.go @@ -0,0 +1,58 @@ +package api + +import ( + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/internal/monitoring" +) + +// Regression coverage for #1437: adaptive polling stretches an instance's +// cadence past the configured interval while data is fresh, so the +// active→stale cutoff has to follow the planned schedule. Before the fix a +// healthy connection on a stretched (e.g. 5-minute) cadence read as stale for +// the back half of every cycle and the platform page dropped to Agent-only. +func TestBuildConnectionsAdaptiveIntervalScalesStaleCutoff(t *testing.T) { + now := time.Date(2026, 8, 11, 12, 0, 0, 0, time.UTC) + lastSuccess := now.Add(-4 * time.Minute) + base := aggregatorInputs{ + pveInstances: []config.PVEInstance{{Name: "lab", Host: "https://lab:8006"}}, + instanceHealth: map[string]monitoring.InstanceHealth{ + "pve::lab": {PollStatus: monitoring.InstancePollStatus{LastSuccess: &lastSuccess}}, + }, + pvePollingInterval: 30 * time.Second, + now: now, + } + + pveState := func(in aggregatorInputs) ConnectionState { + t.Helper() + conns := buildConnections(in) + if len(conns) != 1 { + t.Fatalf("connections = %d, want 1", len(conns)) + } + return conns[0].State + } + + // No planned interval known: a 4-minute-old poll on a 30s configured + // cadence is past the floor and correctly reads stale. + if state := pveState(base); state != ConnectionStateStale { + t.Fatalf("state without plan = %q, want %q", state, ConnectionStateStale) + } + + // The scheduler planned a 5-minute cadence, so 4 minutes old is on + // schedule and must stay active. + onSchedule := base + onSchedule.plannedPollIntervals = map[string]time.Duration{"pve::lab": 5 * time.Minute} + if state := pveState(onSchedule); state != ConnectionStateActive { + t.Fatalf("state with 5m plan = %q, want %q", state, ConnectionStateActive) + } + + // A plan tighter than the configured cadence never tightens the cutoff: + // a genuine outage still trips against the configured interval and floor. + tightPlan := base + tightPlan.plannedPollIntervals = map[string]time.Duration{"pve::lab": 5 * time.Second} + if state := pveState(tightPlan); state != ConnectionStateStale { + t.Fatalf("state with 5s plan = %q, want %q", state, ConnectionStateStale) + } +} diff --git a/internal/api/connections_alerts.go b/internal/api/connections_alerts.go index 36089aa5d..dc8e519a5 100644 --- a/internal/api/connections_alerts.go +++ b/internal/api/connections_alerts.go @@ -66,6 +66,7 @@ func buildAggregatorInputsWithRuntimeSources( inputs.instanceHealth = instanceHealthByKey(monitor.SchedulerHealth()) inputs.availabilityStatuses = monitor.AvailabilityStatusSnapshot() inputs.pbsReportedNodeNames = pbsReportedNodeNamesByInstance(monitor.PBSInstancesSnapshot()) + inputs.plannedPollIntervals = plannedPollIntervalsForConfig(monitor, cfg) } else { inputs.hosts = []models.Host{} inputs.instanceHealth = map[string]monitoring.InstanceHealth{} @@ -143,3 +144,29 @@ func buildAlertConnectionSnapshotsWithRuntimeSources( inputs := buildAggregatorInputsWithRuntimeSources(ctx, cfg, persistence, monitor, runtime) return snapshotConnectionsForAlerts(buildConnections(inputs)) } + +// plannedPollIntervalsForConfig collects the adaptive scheduler's currently +// planned interval for every configured PVE/PBS/PMG instance, keyed the same +// way as instanceHealth ("pve::"). Instances the scheduler has no plan +// for are omitted so the aggregator falls back to the configured cadence. +func plannedPollIntervalsForConfig(monitor *monitoring.Monitor, cfg *config.Config) map[string]time.Duration { + if monitor == nil || cfg == nil { + return nil + } + out := make(map[string]time.Duration) + record := func(instanceType monitoring.InstanceType, keyPrefix, name string) { + if interval := monitor.PlannedPollInterval(instanceType, name); interval > 0 { + out[keyPrefix+name] = interval + } + } + for _, inst := range cfg.PVEInstances { + record(monitoring.InstanceTypePVE, "pve::", inst.Name) + } + for _, inst := range cfg.PBSInstances { + record(monitoring.InstanceTypePBS, "pbs::", inst.Name) + } + for _, inst := range cfg.PMGInstances { + record(monitoring.InstanceTypePMG, "pmg::", inst.Name) + } + return out +} diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 33c82f30d..e50f2f007 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -2636,6 +2636,22 @@ func (m *Monitor) DeadLetterCount() int { return m.deadLetterQueue.Size() } +// PlannedPollInterval reports the adaptive scheduler's currently planned +// interval for one instance, or zero when no plan exists (adaptive polling +// disabled, unknown instance, or startup before the first plan). Consumers +// use it to judge poll freshness against the schedule actually in force +// rather than the configured cadence, which adaptive polling deliberately +// stretches while data is fresh. +func (m *Monitor) PlannedPollInterval(instanceType InstanceType, instanceName string) time.Duration { + if m == nil || m.scheduler == nil { + return 0 + } + if task, ok := m.scheduler.LastScheduled(instanceType, instanceName); ok { + return task.Interval + } + return 0 +} + func (m *Monitor) SchedulerHealth() SchedulerHealthResponse { response := emptySchedulerHealthResponse(m.config != nil && m.config.AdaptivePollingEnabled) diff --git a/internal/monitoring/monitor_planned_interval_test.go b/internal/monitoring/monitor_planned_interval_test.go new file mode 100644 index 000000000..89fb7a6b3 --- /dev/null +++ b/internal/monitoring/monitor_planned_interval_test.go @@ -0,0 +1,40 @@ +package monitoring + +import ( + "testing" + "time" +) + +// PlannedPollInterval feeds the connections aggregator the cadence the +// adaptive scheduler actually promised, so the active→stale cutoff can track +// stretched schedules (#1437). +func TestPlannedPollIntervalReportsLastScheduled(t *testing.T) { + sched := NewAdaptiveScheduler(DefaultSchedulerConfig(), nil, nil, nil) + now := time.Now() + tasks := sched.BuildPlan(now, []InstanceDescriptor{{ + Name: "pve-a", + Type: InstanceTypePVE, + LastSuccess: now, + }}, 0) + if len(tasks) != 1 { + t.Fatalf("planned tasks = %d, want 1", len(tasks)) + } + if tasks[0].Interval <= 0 { + t.Fatalf("planned interval = %v, want > 0", tasks[0].Interval) + } + + m := &Monitor{scheduler: sched} + if got := m.PlannedPollInterval(InstanceTypePVE, "pve-a"); got != tasks[0].Interval { + t.Fatalf("PlannedPollInterval = %v, want %v", got, tasks[0].Interval) + } + if got := m.PlannedPollInterval(InstanceTypePVE, "unknown"); got != 0 { + t.Fatalf("unknown instance interval = %v, want 0", got) + } + if got := (&Monitor{}).PlannedPollInterval(InstanceTypePVE, "pve-a"); got != 0 { + t.Fatalf("schedulerless monitor interval = %v, want 0", got) + } + var nilMonitor *Monitor + if got := nilMonitor.PlannedPollInterval(InstanceTypePVE, "pve-a"); got != 0 { + t.Fatalf("nil monitor interval = %v, want 0", got) + } +}