diff --git a/agent/agent.go b/agent/agent.go index 48000c953..7c0104d2a 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -60,6 +60,7 @@ type Agent struct { monitorManager *MonitorManager // Manages network monitors storagePoolManager *StoragePoolManager // Manages storage pool and dataset data packageUpdates *packageUpdatesManager // Checks for pending package updates + lastHubRequest atomic.Int64 // Wall clock Unix nanoseconds of the last hub stats request } // NewAgent creates a new agent with the given data directory for persisting data. @@ -141,7 +142,7 @@ func NewAgent(dataDir ...string) (agent *Agent, err error) { agent.handlerRegistry = NewHandlerRegistry() // initialize monitor manager - agent.monitorManager = newMonitorManager() + agent.monitorManager = newMonitorManager(agent) agent.storagePoolManager = newStoragePoolManager() @@ -191,6 +192,18 @@ func NewAgent(dataDir ...string) (agent *Agent, err error) { return agent, nil } +// markHubRequest records a hub stats request. It doesn't use the agent lock, +// which gatherStats holds through collection. +func (a *Agent) markHubRequest() { + a.lastHubRequest.Store(time.Now().UnixNano()) +} + +// hubIdleFor reports whether the hub hasn't requested stats for at least d. +// Wall time counts host suspend toward idle. +func (a *Agent) hubIdleFor(d time.Duration) bool { + return time.Now().UnixNano()-a.lastHubRequest.Load() >= int64(d) +} + func (a *Agent) gatherStats(options common.DataRequestOptions) *system.CombinedData { a.Lock() defer a.Unlock() diff --git a/agent/handlers.go b/agent/handlers.go index df5b651af..1e5f07413 100644 --- a/agent/handlers.go +++ b/agent/handlers.go @@ -103,6 +103,7 @@ func (h *GetDataHandler) Handle(hctx *HandlerContext) error { var options common.DataRequestOptions _ = cbor.Unmarshal(hctx.Request.Data, &options) + hctx.Agent.markHubRequest() sysStats := hctx.Agent.gatherStats(options) // Cached stats may be shared by concurrent SSH and WebSocket requests. // Set the transport on the response copy, not on the cached data. diff --git a/agent/network_monitor.go b/agent/network_monitor.go index b0486181c..6931d940f 100644 --- a/agent/network_monitor.go +++ b/agent/network_monitor.go @@ -4,13 +4,20 @@ import ( "crypto/tls" "errors" "fmt" + "log/slog" "net/http" "sync" + "sync/atomic" "time" "github.com/henrygd/beszel/internal/entities/monitor" ) +// monitorIdleTimeout is three default hub update intervals. The hub collects +// monitor results only with stats, which it stops requesting while the system is +// paused or the agent is disconnected, so scheduled probes stop after this long. +const monitorIdleTimeout = 3 * time.Minute + // MonitorManager manages network monitor configurations and task lifetimes. type MonitorManager struct { mu sync.RWMutex @@ -18,14 +25,18 @@ type MonitorManager struct { probe monitorProbe certCheck certChecker resumeGuard monitorResumeGuard + agent *Agent // Reference to the parent agent; nil means the hub is never idle + idle atomic.Bool } -func newMonitorManager() *MonitorManager { +func newMonitorManager(agent *Agent) *MonitorManager { client := &http.Client{Timeout: monitor.MaxProbeTimeout} insecureTransport := http.DefaultTransport.(*http.Transport).Clone() insecureTransport.TLSClientConfig = &tls.Config{InsecureSkipVerify: true} insecureClient := &http.Client{Timeout: monitor.MaxProbeTimeout, Transport: insecureTransport} - return newMonitorManagerWithProbe(networkMonitorProbe(client, insecureClient)) + pm := newMonitorManagerWithProbe(networkMonitorProbe(client, insecureClient)) + pm.agent = agent + return pm } func newMonitorManagerWithProbe(probe monitorProbe) *MonitorManager { @@ -188,6 +199,20 @@ func (pm *MonitorManager) GetResults(durationMs uint16) map[string]monitor.Resul return results } +// hubIdle reports whether scheduled probes should be skipped because the hub +// hasn't requested stats recently. Immediate probes requested by the hub still run. +func (pm *MonitorManager) hubIdle() bool { + idle := pm.agent != nil && pm.agent.hubIdleFor(monitorIdleTimeout) + if pm.idle.Swap(idle) != idle { + if idle { + slog.Debug("pausing network monitors until the hub requests stats") + } else { + slog.Debug("resuming network monitors") + } + } + return idle +} + // Stop stops all monitor tasks. func (pm *MonitorManager) Stop() { pm.mu.Lock() diff --git a/agent/network_monitor_schedule.go b/agent/network_monitor_schedule.go index 4394215b0..d5d590846 100644 --- a/agent/network_monitor_schedule.go +++ b/agent/network_monitor_schedule.go @@ -17,6 +17,9 @@ func (pm *MonitorManager) startMonitor(task *monitorTask) { // Certificate checks piggyback on probe ticks, so they run at most once per // probe interval after they become due. go runMonitorSchedule(task.ctx, interval, delay, func() { + if pm.hubIdle() { + return + } if _, allowed := task.resumeGuard.snapshot(); allowed { task.runProbe(pm.probe) task.refreshCert(pm.certCheck) diff --git a/agent/network_monitor_schedule_test.go b/agent/network_monitor_schedule_test.go index a66a59abe..9348b4240 100644 --- a/agent/network_monitor_schedule_test.go +++ b/agent/network_monitor_schedule_test.go @@ -165,3 +165,50 @@ func TestMonitorInjectedProbeTimeoutRecordsLoss(t *testing.T) { assert.NoError(t, pm.monitors["test"].ctx.Err()) }) } + +func TestMonitorScheduleSkipsProbesWhileHubIdle(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + var calls atomic.Int32 + pm := newMonitorManagerWithProbe(func(context.Context, monitor.Config) (int64, error) { + calls.Add(1) + return 42, nil + }) + defer pm.Stop() + a := &Agent{} + pm.agent = a + cfg := monitor.Config{ID: "test", Interval: 10} + + // Configs synced before the first stats request don't probe on schedule. + pm.SyncMonitors([]monitor.Config{cfg}) + time.Sleep(time.Minute) + synctest.Wait() + assert.Zero(t, calls.Load(), "monitors must not probe before the hub requests stats") + + // Immediate probes requested by the hub still run. + result, err := pm.UpsertMonitor(cfg, true) + require.NoError(t, err) + require.NotNil(t, result) + assert.Equal(t, 1, int(calls.Load())) + + a.markHubRequest() + time.Sleep(time.Minute) + synctest.Wait() + active := calls.Load() + assert.GreaterOrEqual(t, int(active), 6, "monitors must probe while the hub requests stats") + + // Without further requests, probing stops once the idle timeout passes. + time.Sleep(monitorIdleTimeout) + synctest.Wait() + idle := calls.Load() + time.Sleep(10 * time.Minute) + synctest.Wait() + assert.Equal(t, idle, calls.Load(), "monitors must not probe while the hub is idle") + assert.NotNil(t, pm.monitors[cfg.ID], "idle monitors keep their config") + + // The next stats request resumes the schedule without a resync. + a.markHubRequest() + time.Sleep(time.Minute) + synctest.Wait() + assert.Greater(t, calls.Load(), idle, "monitors must resume when the hub requests stats") + }) +} diff --git a/agent/network_monitor_test.go b/agent/network_monitor_test.go index d1e37e591..119abd2e0 100644 --- a/agent/network_monitor_test.go +++ b/agent/network_monitor_test.go @@ -26,7 +26,7 @@ func TestMonitorManagerGetResultsIncludesHourResponseRange(t *testing.T) { task.history.addSampleLocked(monitorSample{responseUs: 30, timestamp: now.Add(-50 * time.Second)}) task.history.addSampleLocked(monitorSample{responseUs: -1, timestamp: now.Add(-30 * time.Second)}) - pm := newMonitorManager() + pm := newMonitorManager(nil) pm.monitors = map[string]*monitorTask{"icmp:example.com": task} results := pm.GetResults(uint16(time.Minute / time.Millisecond)) @@ -48,7 +48,7 @@ func TestMonitorManagerGetResultsIncludesLossOnlyHourData(t *testing.T) { task.history.addSampleLocked(monitorSample{responseUs: -1, timestamp: now.Add(-30 * time.Second)}) task.history.addSampleLocked(monitorSample{responseUs: -1, timestamp: now.Add(-10 * time.Second)}) - pm := newMonitorManager() + pm := newMonitorManager(nil) pm.monitors = map[string]*monitorTask{"icmp:example.com": task} results := pm.GetResults(uint16(time.Minute / time.Millisecond)) @@ -73,7 +73,7 @@ func TestMonitorManagerSyncMonitorsSkipsConfigsWithoutStableID(t *testing.T) { validCfg := monitor.Config{ID: "monitor-1", Target: "ignored", Protocol: "noop", Interval: 10} invalidCfg := monitor.Config{Target: "ignored", Protocol: "noop", Interval: 10} - pm := newMonitorManager() + pm := newMonitorManager(nil) pm.SyncMonitors([]monitor.Config{validCfg, invalidCfg}) defer pm.Stop() @@ -89,7 +89,7 @@ func TestMonitorManagerSyncMonitorsStopsRemovedTasksButKeepsExisting(t *testing. keptTask := newMonitorTask(keepCfg) removedTask := newMonitorTask(removeCfg) - pm := newMonitorManager() + pm := newMonitorManager(nil) pm.monitors = map[string]*monitorTask{ keepCfg.ID: keptTask, removeCfg.ID: removedTask, @@ -118,7 +118,7 @@ func TestMonitorManagerSyncMonitorsRestartsChangedConfig(t *testing.T) { originalCfg := monitor.Config{ID: "monitor-1", Target: "ignored-a", Protocol: "noop", Interval: 10} updatedCfg := monitor.Config{ID: "monitor-1", Target: "ignored-b", Protocol: "noop", Interval: 10} originalTask := newMonitorTask(originalCfg) - pm := newMonitorManager() + pm := newMonitorManager(nil) pm.monitors = map[string]*monitorTask{ originalCfg.ID: originalTask, } @@ -176,7 +176,7 @@ func TestMonitorManagerUpsertMonitorKeepsHistoryWhenOnlyIntervalChanges(t *testi existingTask.history.addSampleLocked(monitorSample{responseUs: 12, timestamp: now.Add(-50 * time.Minute)}) existingTask.history.addSampleLocked(monitorSample{responseUs: 24, timestamp: now.Add(-30 * time.Second)}) - pm := newMonitorManager() + pm := newMonitorManager(nil) pm.monitors = map[string]*monitorTask{originalCfg.ID: existingTask} result, err := pm.UpsertMonitor(updatedCfg, false) @@ -211,7 +211,7 @@ func TestMonitorManagerUpsertMonitorKeepsHistoryWhenOnlyIntervalChanges(t *testi func TestMonitorManagerApplySyncDeleteRemovesTask(t *testing.T) { config := monitor.Config{ID: "monitor-1", Target: "1.1.1.1", Protocol: "icmp", Interval: 10} task := newMonitorTask(config) - pm := newMonitorManager() + pm := newMonitorManager(nil) pm.monitors = map[string]*monitorTask{config.ID: task} _, err := pm.HandleSyncRequest(monitor.SyncRequest{ @@ -267,7 +267,7 @@ func TestMonitorHTTP(t *testing.T) { w.WriteHeader(http.StatusNoContent) })) defer server.Close() - probe := newMonitorManager().probe + probe := newMonitorManager(nil).probe config := monitor.Config{Protocol: "http", Target: server.URL} responseUs, err := probe(context.Background(), config) @@ -488,7 +488,7 @@ func TestMonitorManagerCancelsActiveProbe(t *testing.T) { })) defer server.Close() defer close(release) - pm := newMonitorManager() + pm := newMonitorManager(nil) defer pm.Stop() cfg := monitor.Config{ID: "test", Protocol: "http", Target: server.URL, Interval: 3600} task := newMonitorTask(cfg) @@ -593,7 +593,7 @@ func TestMonitorProbeTimeoutRecordsLoss(t *testing.T) { })) defer server.Close() defer close(release) - pm := newMonitorManager() + pm := newMonitorManager(nil) pm.probe = networkMonitorProbe(&http.Client{Timeout: 20 * time.Millisecond}, nil) task := newMonitorTask(monitor.Config{ID: "timeout", Protocol: "http", Target: server.URL}) defer task.cancel()