From a8d2b69b307d96affad508b654f9e72121bda952 Mon Sep 17 00:00:00 2001 From: sivaratrisrinivas Date: Tue, 18 Mar 2025 23:17:45 -0500 Subject: [PATCH] metrics: fix goroutine leak in meterTicker loop --- metrics/meter.go | 111 +++++++++++++++++++++++++++++++++++------- metrics/meter_test.go | 56 +++++++++++++++++++++ metrics/metrics.go | 9 ++++ 3 files changed, 158 insertions(+), 18 deletions(-) diff --git a/metrics/meter.go b/metrics/meter.go index 194bd1f304..8c09e552b4 100644 --- a/metrics/meter.go +++ b/metrics/meter.go @@ -124,46 +124,121 @@ func (m *Meter) tick() { m.a15.tick() } -var arbiter = meterTicker{meters: make(map[*Meter]struct{})} +var arbiter = newMeterTicker() // meterTicker ticks meters every 5s from a single goroutine. // meters are references in a set for future stopping. type meterTicker struct { - mu sync.RWMutex - - started bool - meters map[*Meter]struct{} + mu sync.RWMutex + quit chan struct{} + started bool + meters map[*Meter]struct{} + running bool + initOnce sync.Once } -// add adds another *Meter ot the arbiter, and starts the arbiter ticker. +// newMeterTicker creates a new meterTicker. +func newMeterTicker() *meterTicker { + return &meterTicker{ + meters: make(map[*Meter]struct{}), + quit: make(chan struct{}), + } +} + +// add adds another *Meter to the arbiter, and lazily starts the arbiter ticker +// only when metrics are enabled and at least one meter is added. func (ma *meterTicker) add(m *Meter) { ma.mu.Lock() defer ma.mu.Unlock() + ma.meters[m] = struct{}{} - if !ma.started { - ma.started = true - go ma.loop() + + // Only start the ticker if metrics are enabled and we haven't started yet + if metricsEnabled && !ma.started { + ma.initOnce.Do(func() { + ma.started = true + ma.running = true + go ma.loop() + }) } } // remove removes a meter from the set of ticked meters. func (ma *meterTicker) remove(m *Meter) { ma.mu.Lock() + defer ma.mu.Unlock() + delete(ma.meters, m) - ma.mu.Unlock() + + // If we have no more meters and the ticker is running, consider stopping + if len(ma.meters) == 0 && ma.running { + ma.stop() + } +} + +// stop stops the ticker goroutine. +func (ma *meterTicker) stop() { + if ma.running { + close(ma.quit) + ma.running = false + ma.started = false + // Create a new quit channel for future use + ma.quit = make(chan struct{}) + // Reset the init once so we can start again if needed + ma.initOnce = sync.Once{} + } +} + +// ensureStarted ensures the ticker is started if metrics are enabled +// and there are meters to process. +func (ma *meterTicker) ensureStarted() { + ma.mu.Lock() + defer ma.mu.Unlock() + + if metricsEnabled && len(ma.meters) > 0 && !ma.started { + ma.initOnce.Do(func() { + ma.started = true + ma.running = true + go ma.loop() + }) + } } // loop ticks meters on a 5 second interval. func (ma *meterTicker) loop() { ticker := time.NewTicker(5 * time.Second) - for range ticker.C { - if !metricsEnabled { - continue + defer ticker.Stop() + + for { + select { + case <-ticker.C: + // Skip processing if metrics are disabled + if !metricsEnabled { + continue + } + ma.mu.RLock() + for meter := range ma.meters { + meter.tick() + } + ma.mu.RUnlock() + case <-ma.quit: + return } - ma.mu.RLock() - for meter := range ma.meters { - meter.tick() - } - ma.mu.RUnlock() + } +} + +// EnableMetricsTicking ensures that the metrics ticker is running +// if metrics are enabled and meters exist. +func EnableMetricsTicking() { + arbiter.ensureStarted() +} + +// DisableMetricsTicking stops the metrics ticker. +func DisableMetricsTicking() { + arbiter.mu.Lock() + defer arbiter.mu.Unlock() + + if arbiter.running { + arbiter.stop() } } diff --git a/metrics/meter_test.go b/metrics/meter_test.go index e3f39684bd..448ecb74cd 100644 --- a/metrics/meter_test.go +++ b/metrics/meter_test.go @@ -81,3 +81,59 @@ func TestMeterRepeat(t *testing.T) { t.Errorf("m.Count(): 10100 != %v\n", count) } } + +func TestMeterTickerLifecycle(t *testing.T) { + // Make sure metrics are disabled initially + oldEnabled := metricsEnabled + metricsEnabled = false + defer func() { + metricsEnabled = oldEnabled + }() + + // Reset the arbiter + oldArbiter := arbiter + arbiter = newMeterTicker() + defer func() { + arbiter = oldArbiter + }() + + // Create a new meter but metrics are disabled + // This should not start the ticker + m := NewMeter() + if arbiter.started || arbiter.running { + t.Error("Ticker started when metrics are disabled") + } + + // Enable metrics, which should start the ticker + metricsEnabled = true + EnableMetricsTicking() + + // Check if the ticker is started + arbiter.mu.RLock() + started := arbiter.started + running := arbiter.running + arbiter.mu.RUnlock() + + if !started || !running { + t.Error("Ticker not started when metrics are enabled") + } + + // Stop the meter + m.Stop() + + // Ticker should be stopped as there are no more meters + arbiter.mu.RLock() + running = arbiter.running + arbiter.mu.RUnlock() + + if running { + t.Error("Ticker still running after all meters are stopped") + } + + // Disable metrics, which should stop the ticker + metricsEnabled = false + Disable() + + // Checking for any goroutine leaks would require a more complex test + // with runtime.NumGoroutine() checks +} diff --git a/metrics/metrics.go b/metrics/metrics.go index c4c43b7576..d9310b9a9d 100644 --- a/metrics/metrics.go +++ b/metrics/metrics.go @@ -30,6 +30,15 @@ func Enabled() bool { // the program, before any metrics collection will happen. func Enable() { metricsEnabled = true + // Ensure metrics ticker is started if there are any meters + EnableMetricsTicking() +} + +// Disable disables the metrics system and stops any running metric tickers. +func Disable() { + metricsEnabled = false + // Stop the metrics ticker + DisableMetricsTicking() } var threadCreateProfile = pprof.Lookup("threadcreate")