diff --git a/pkg/stats/monitor.go b/pkg/stats/monitor.go index 620aefaa..04413199 100644 --- a/pkg/stats/monitor.go +++ b/pkg/stats/monitor.go @@ -58,7 +58,7 @@ type Monitor struct { nodeID string clusterID string cpuCostConfig *config.CPUCostConfig - pulseSinkReapGrace time.Duration + pulseSinkReapGrace atomic.Duration promCPULoad prometheus.Gauge promCgroupMemory prometheus.Gauge @@ -122,10 +122,8 @@ func NewMonitor(conf *config.ServiceConfig, svc Service) (*Monitor, error) { m.initPrometheus() - if conf.PulseSinkReapGraceSec > 0 { - m.pulseSinkReapGrace = time.Duration(conf.PulseSinkReapGraceSec) * time.Second - go m.runPulseSinkReaper() - } + m.SetPulseSinkReapGraceSec(conf.PulseSinkReapGraceSec) + go m.runPulseSinkReaper() procStats, err := hwstats.NewProcMonitor(m.updateEgressStats) if err != nil { diff --git a/pkg/stats/pulse_reaper.go b/pkg/stats/pulse_reaper.go index ac8feaaa..82732051 100644 --- a/pkg/stats/pulse_reaper.go +++ b/pkg/stats/pulse_reaper.go @@ -30,17 +30,40 @@ const ( pulseReapCmdTimeout = 10 * time.Second ) +// SetPulseSinkReapGraceSec applies an updated reap grace, taking effect on the +// reaper's next tick. A non-positive grace disables reaping. +func (m *Monitor) SetPulseSinkReapGraceSec(graceSec int) { + m.pulseSinkReapGrace.Store(time.Duration(graceSec) * time.Second) +} + +// a grace shorter than the default interval would otherwise be quantized back up to it +func pulseReapTick(grace time.Duration) time.Duration { + if grace > 0 { + return min(pulseReapInterval, grace) + } + return pulseReapInterval +} + // runPulseSinkReaper unloads null-sinks left on the shared pulse daemon by handlers that // exited without cleanup, before they accumulate and hit the pulse client limit. +// The grace is re-read every tick, so reaping can be reconfigured or disabled at runtime. func (m *Monitor) runPulseSinkReaper() { firstOrphaned := make(map[string]time.Time) listFailing := false - // a grace shorter than the default interval would otherwise be quantized back up to it - ticker := time.NewTicker(min(pulseReapInterval, m.pulseSinkReapGrace)) + ticker := time.NewTicker(pulseReapTick(m.pulseSinkReapGrace.Load())) defer ticker.Stop() for range ticker.C { + grace := m.pulseSinkReapGrace.Load() + ticker.Reset(pulseReapTick(grace)) + + if grace <= 0 { + // re-enabling starts a fresh grace period instead of reaping instantly + clear(firstOrphaned) + continue + } + info, err := listPulse() if err != nil { if !listFailing { @@ -51,7 +74,7 @@ func (m *Monitor) runPulseSinkReaper() { } listFailing = false - toReap, egressSinks := planPulseSinkReaps(info.Sinks, m.activeEgressIDs(), firstOrphaned, m.pulseSinkReapGrace, time.Now()) + toReap, egressSinks := planPulseSinkReaps(info.Sinks, m.activeEgressIDs(), firstOrphaned, grace, time.Now()) m.promPulseSinks.Set(float64(egressSinks)) for _, sink := range toReap {