diff --git a/interface.go b/interface.go index 2b9ff5eb..5320b4ce 100644 --- a/interface.go +++ b/interface.go @@ -639,17 +639,15 @@ func (f *Interface) reloadMisc(c *config.C) { } } -func (f *Interface) emitStats(ctx context.Context, i time.Duration) { - ticker := time.NewTicker(i) - defer ticker.Stop() - +// statsEmitter updates the interface's gauges, the stats server runs it on its capture loop. +func (f *Interface) statsEmitter() func() { udpStats := udp.NewUDPStatsEmitter(f.writers) certExpirationGauge := metrics.GetOrRegisterGauge("certificate.ttl_seconds", nil) certInitiatingVersion := metrics.GetOrRegisterGauge("certificate.initiating_version", nil) certMaxVersion := metrics.GetOrRegisterGauge("certificate.max_version", nil) - emit := func() { + return func() { f.firewall.EmitStats() f.handshakeManager.EmitStats() udpStats() @@ -666,19 +664,6 @@ func (f *Interface) emitStats(ctx context.Context, i time.Duration) { certMaxVersion.Update(int64(certState.v1Cert.Version())) } } - - // Prime gauges so a Prometheus scrape that lands before the first tick - // sees real values instead of the zero defaults (issue #907). - emit() - - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - emit() - } - } } func (f *Interface) GetHostInfo(vpnIp netip.Addr) *HostInfo { diff --git a/interface_emit_test.go b/interface_emit_test.go index b0a9d025..f78e02ca 100644 --- a/interface_emit_test.go +++ b/interface_emit_test.go @@ -3,7 +3,6 @@ package nebula import ( - "context" "net/netip" "testing" "time" @@ -18,11 +17,8 @@ import ( "github.com/stretchr/testify/require" ) -// Test_emitStats_primesGauges covers issue #907: a Prometheus scrape that -// landed before the first ticker fire used to read 0 for the cert gauges. -// emitStats now primes the gauges before entering the ticker loop. We assert -// the gauge is zero before the first call and non-zero after. -func Test_emitStats_primesGauges(t *testing.T) { +// Priming before the first tick (issue #907) is Start's job, see TestStatsServer_Start_primes. +func Test_statsEmitter_setsGauges(t *testing.T) { defer metrics.DefaultRegistry.UnregisterAll() l := test.NewLogger() @@ -56,17 +52,12 @@ func Test_emitStats_primesGauges(t *testing.T) { ifce.pki.cs.Store(cs) ttlGauge := metrics.GetOrRegisterGauge("certificate.ttl_seconds", nil) - require.Zero(t, ttlGauge.Value(), "gauge should be zero before emitStats runs") + require.Zero(t, ttlGauge.Value(), "gauge should be zero before the emitter runs") - // Pre-cancel the context so emitStats returns after priming the gauges - // without ever reading from ticker.C. The one hour interval is just a - // belt-and-suspenders, the test does not expect the ticker to fire. - ctx, cancel := context.WithCancel(context.Background()) - cancel() - ifce.emitStats(ctx, time.Hour) + ifce.statsEmitter()() ttl := ttlGauge.Value() - assert.Positive(t, ttl, "ttl gauge should be primed by emitStats before its first tick") + assert.Positive(t, ttl, "ttl gauge should be set by one run of the emitter") assert.LessOrEqual(t, ttl, int64(3600)) assert.Equal(t, int64(cert.Version1), metrics.GetOrRegisterGauge("certificate.initiating_version", nil).Value()) assert.Equal(t, int64(cert.Version1), metrics.GetOrRegisterGauge("certificate.max_version", nil).Value()) diff --git a/main.go b/main.go index 52f7bea8..a8d01779 100644 --- a/main.go +++ b/main.go @@ -300,6 +300,7 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev } var ifce *Interface + var emitStats func() if !configTest { ifce, err = NewInterface(ctx, ifConfig) if err != nil { @@ -318,9 +319,12 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev go handshakeManager.Run(ctx) punchy.Start(ctx, ifce, hostMap, lightHouse) + + // On the stats server's loop, so the gauges follow the interval it accepted + emitStats = ifce.statsEmitter() } - stats, err := newStatsServerFromConfig(ctx, l, c, buildVersion, configTest) + stats, err := newStatsServerFromConfig(ctx, l, c, buildVersion, configTest, emitStats) if err != nil { return nil, util.ContextualizeIfNeeded("Failed to start stats emitter", err) } @@ -329,8 +333,6 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev return nil, nil } - go ifce.emitStats(ctx, c.GetDuration("stats.interval", time.Second*10)) - attachCommands(l, c, ssh, ifce) networkChanges := udp.NewNetworkChangeMonitor(ctx, l, c) diff --git a/stats.go b/stats.go index c6895eef..7e42d44b 100644 --- a/stats.go +++ b/stats.go @@ -38,6 +38,11 @@ type statsServer struct { runMu sync.Mutex runCfg *statsConfig run *statsRuntime // non-nil while a runtime is live + // Closed when the last runtime's capture loop has returned. Stop waits on it so nothing captures once it returns, + // and a reload's new runtime never overlaps the old one, go-metrics keeps its GC capture state in unguarded globals + loopDone chan struct{} + // See newStatsServerFromConfig, never changes after it + emitStats func() } // statsRuntime is the live state owned by a single Start invocation. Start @@ -82,12 +87,16 @@ type promConfig struct { // // Start is safe to call unconditionally: it no-ops when stats are disabled. // The returned pointer is always non-nil, even on error. -func newStatsServerFromConfig(ctx context.Context, l *slog.Logger, c *config.C, buildVersion string, configTest bool) (*statsServer, error) { +// +// emitStats, if set, updates metrics on every capture, ahead of the exporters so each export sees what it just set. It +// is fixed here, before the reload callback exists, so every runtime has it. +func newStatsServerFromConfig(ctx context.Context, l *slog.Logger, c *config.C, buildVersion string, configTest bool, emitStats func()) (*statsServer, error) { s := &statsServer{ l: l, ctx: ctx, buildVersion: buildVersion, configTest: configTest, + emitStats: emitStats, } c.RegisterReloadCallback(func(c *config.C) { @@ -120,7 +129,6 @@ func (s *statsServer) reload(c *config.C, initial bool) error { s.runMu.Lock() sameCfg := s.runCfg != nil && *s.runCfg == newCfg s.runCfg = &newCfg - running := s.run != nil s.runMu.Unlock() s.enabled.Store(enabled) @@ -129,9 +137,8 @@ func (s *statsServer) reload(c *config.C, initial bool) error { return nil } - if running { - s.Stop() - } + // Even with nothing running, a runtime whose listener failed can still be finishing its last pass + s.Stop() if enabled && !s.configTest { go s.Start() } @@ -157,9 +164,15 @@ func (s *statsServer) Start() { captureFns, listener := s.buildRuntime(runCtx, cfg) rt := &statsRuntime{cancel: cancel, listener: listener} s.run = rt + done := make(chan struct{}) + s.loopDone = done s.runMu.Unlock() - go captureStatsLoop(runCtx, cfg.interval, captureFns) + // Prime before serving, so a scrape that lands before the first tick sees real values instead of zeros (issue #907) + for _, fn := range captureFns { + fn() + } + go captureStatsLoop(runCtx, cfg.interval, captureFns, done) cleanExit := true if listener == nil { @@ -218,20 +231,25 @@ func (s *statsServer) serveListener(listener *http.Server) bool { // Stop tears down the active runtime, if any. Idempotent. func (s *statsServer) Stop() { + // Held until the last pass is done, a Start that slipped in after s.run cleared would capture alongside it s.runMu.Lock() + defer s.runMu.Unlock() rt := s.run s.run = nil - s.runMu.Unlock() - if rt == nil { - return - } - rt.cancel() - if rt.listener != nil { - shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - if err := rt.listener.Shutdown(shutdownCtx); err != nil { - s.l.Warn("Failed to shut down prometheus stats listener", "error", err) + if rt != nil { + rt.cancel() + if rt.listener != nil { + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + if err := rt.listener.Shutdown(shutdownCtx); err != nil { + s.l.Warn("Failed to shut down prometheus stats listener", "error", err) + } + cancel() } - cancel() + } + // Nothing captures once Stop returns. The last runtime is cancelled by now, whoever cleared it, so this is at most + // the pass it was in. Nothing in a pass takes runMu + if s.loopDone != nil { + <-s.loopDone } } @@ -248,6 +266,9 @@ func (s *statsServer) buildRuntime(ctx context.Context, cfg statsConfig) ([]func func() { metrics.CaptureDebugGCStatsOnce(metrics.DefaultRegistry) }, func() { metrics.CaptureRuntimeMemStatsOnce(metrics.DefaultRegistry) }, } + if s.emitStats != nil { + captureFns = append(captureFns, s.emitStats) + } switch cfg.typ { case "graphite": @@ -310,8 +331,10 @@ func (s *statsServer) buildRuntime(ctx context.Context, cfg statsConfig) ([]func return captureFns, nil } -// captureStatsLoop runs each fn on every tick of d until ctx is cancelled. -func captureStatsLoop(ctx context.Context, d time.Duration, fns []func()) { +// captureStatsLoop runs each fn on every tick of d until ctx is cancelled, then +// closes done. +func captureStatsLoop(ctx context.Context, d time.Duration, fns []func(), done chan struct{}) { + defer close(done) t := time.NewTicker(d) defer t.Stop() for { diff --git a/stats_test.go b/stats_test.go index 58bfe417..a89ac9bb 100644 --- a/stats_test.go +++ b/stats_test.go @@ -2,12 +2,18 @@ package nebula import ( "context" + "fmt" + "io" "log/slog" "net" + "net/http" + "net/http/httptest" "strconv" + "sync/atomic" "testing" "time" + "github.com/rcrowley/go-metrics" "github.com/slackhq/nebula/config" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -369,6 +375,19 @@ func waitForListening(t *testing.T, addr string) { }) } +// stopAtCleanup returns the ctx to build *s with, and stops it when the test ends. Cancelling first turns away a Start +// a reload spawned that hasn't run yet, then Stop waits out the last pass, so nothing captures into the next test +func stopAtCleanup(t *testing.T, s **statsServer) context.Context { + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(func() { + cancel() + if *s != nil { + (*s).Stop() + } + }) + return ctx +} + func freeTCPPort(t *testing.T) string { t.Helper() ln, err := net.Listen("tcp", "127.0.0.1:0") @@ -377,3 +396,280 @@ func freeTCPPort(t *testing.T) string { require.NoError(t, ln.Close()) return strconv.Itoa(port) } + +// Each export has to see what emitStats set in the same pass +func TestStatsServer_emitStatsRunsBeforeExport(t *testing.T) { + defer metrics.DefaultRegistry.Unregister("emitter.order.test") + + l := slog.New(slog.DiscardHandler) + c := config.NewC(l) + setStatsConfig(c, map[string]any{ + "type": "prometheus", + "interval": "1s", + "listen": "127.0.0.1:0", + "path": "/metrics", + }) + gauge := metrics.GetOrRegisterGauge("emitter.order.test", nil) + s, err := newStatsServerFromConfig(t.Context(), l, c, "", false, func() { gauge.Update(42) }) + require.NoError(t, err) + + s.runMu.Lock() + fns, srv := s.buildRuntime(t.Context(), *s.runCfg) + s.runMu.Unlock() + require.NotNil(t, srv) + for _, fn := range fns { + fn() + } + + rec := httptest.NewRecorder() + srv.Handler.ServeHTTP(rec, httptest.NewRequest("GET", "/metrics", nil)) + assert.Contains(t, rec.Body.String(), "emitter_order_test 42") +} + +// A scrape that lands the moment the listener is up already sees what emitStats set (issue #907) +func TestStatsServer_Start_primesBeforeServing(t *testing.T) { + defer metrics.DefaultRegistry.Unregister("prime.before.serving") + + port := freeTCPPort(t) + l := slog.New(slog.DiscardHandler) + c := config.NewC(l) + setStatsConfig(c, map[string]any{ + "type": "prometheus", + "interval": "1h", + "listen": "127.0.0.1:" + port, + "path": "/metrics", + }) + gauge := metrics.GetOrRegisterGauge("prime.before.serving", nil) + // The pause would let a listener that came up before the prime serve the zero value + var s *statsServer + ctx := stopAtCleanup(t, &s) + s, err := newStatsServerFromConfig(ctx, l, c, "", false, func() { + time.Sleep(200 * time.Millisecond) + gauge.Update(42) + }) + require.NoError(t, err) + + go s.Start() + waitForListening(t, "127.0.0.1:"+port) + + resp, err := http.Get("http://127.0.0.1:" + port + "/metrics") + require.NoError(t, err) + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + require.NoError(t, err) + assert.Contains(t, string(body), "prime_before_serving 42") +} + +// Start captures once before the first tick, the interval here is an hour +func TestStatsServer_Start_primes(t *testing.T) { + ln, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + defer ln.Close() + + l := slog.New(slog.DiscardHandler) + c := config.NewC(l) + setStatsConfig(c, map[string]any{"type": "graphite", "interval": "1h", "host": ln.Addr().String()}) + primed := make(chan struct{}, 1) + var s *statsServer + ctx := stopAtCleanup(t, &s) + s, err = newStatsServerFromConfig(ctx, l, c, "", false, func() { + select { + case primed <- struct{}{}: + default: + } + }) + require.NoError(t, err) + + go s.Start() + select { + case <-primed: + case <-time.After(5 * time.Second): + t.Fatal("nothing was captured before the first tick") + } +} + +// Graphite has to ship what emitStats set in the same pass, not the pass before +func TestStatsServer_emitStatsRunsBeforeGraphiteExport(t *testing.T) { + defer metrics.DefaultRegistry.Unregister("emitter.graphite.test") + + ln, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + defer ln.Close() + received := make(chan string, 1) + go func() { + conn, err := ln.Accept() + if err != nil { + return + } + defer conn.Close() + b, _ := io.ReadAll(conn) + received <- string(b) + }() + + l := slog.New(slog.DiscardHandler) + c := config.NewC(l) + setStatsConfig(c, map[string]any{"type": "graphite", "interval": "1h", "host": ln.Addr().String()}) + gauge := metrics.GetOrRegisterGauge("emitter.graphite.test", nil) + s, err := newStatsServerFromConfig(t.Context(), l, c, "", false, func() { gauge.Update(42) }) + require.NoError(t, err) + + s.runMu.Lock() + fns, _ := s.buildRuntime(t.Context(), *s.runCfg) + s.runMu.Unlock() + for _, fn := range fns { + fn() + } + + select { + case body := <-received: + assert.Contains(t, body, ".emitter.graphite.test.value 42 ") + case <-time.After(5 * time.Second): + t.Fatal("graphite sent nothing") + } +} + +// stats.interval reloads, and the runtime a reload starts still has emitStats +func TestStatsServer_reloadMovesTheInterval(t *testing.T) { + var passes atomic.Int64 + l := slog.New(slog.DiscardHandler) + c := config.NewC(l) + // Nothing listens here, each export fails at once + host := "127.0.0.1:" + freeTCPPort(t) + setStatsConfig(c, map[string]any{"type": "graphite", "interval": "1h", "host": host}) + var s *statsServer + ctx := stopAtCleanup(t, &s) + s, err := newStatsServerFromConfig(ctx, l, c, "", false, func() { passes.Add(1) }) + require.NoError(t, err) + go s.Start() + waitFor(t, func() bool { return passes.Load() == 1 }) + + setStatsConfig(c, map[string]any{"type": "graphite", "interval": "20ms", "host": host}) + require.NoError(t, s.reload(c, false)) + waitFor(t, func() bool { return passes.Load() >= 6 }) + + setStatsConfig(c, map[string]any{"type": "graphite", "interval": "1h", "host": host}) + require.NoError(t, s.reload(c, false)) + // The reload's Stop waited out the 20ms loop. The new runtime's prime may land either side of this read, it is the + // only pass left to come + stopped := passes.Load() + time.Sleep(300 * time.Millisecond) + assert.LessOrEqual(t, passes.Load()-stopped, int64(1), "the 20ms loop kept running after the reload") +} + +// A reload's new runtime never captures while the old one is still in a pass, go-metrics keeps its GC capture state +// in unguarded globals +func TestStatsServer_reloadWaitsForTheLastPass(t *testing.T) { + var active, most atomic.Int64 + l := slog.New(slog.DiscardHandler) + c := config.NewC(l) + host := "127.0.0.1:" + freeTCPPort(t) + setStatsConfig(c, map[string]any{"type": "graphite", "interval": "10ms", "host": host}) + var s *statsServer + ctx := stopAtCleanup(t, &s) + s, err := newStatsServerFromConfig(ctx, l, c, "", false, func() { + n := active.Add(1) + for { + m := most.Load() + if n <= m || most.CompareAndSwap(m, n) { + break + } + } + // Long enough that a reload almost always lands mid pass + time.Sleep(50 * time.Millisecond) + active.Add(-1) + }) + require.NoError(t, err) + go s.Start() + + for i := range 6 { + setStatsConfig(c, map[string]any{"type": "graphite", "interval": fmt.Sprintf("%dms", 11-i%2), "host": host}) + require.NoError(t, s.reload(c, false)) + time.Sleep(30 * time.Millisecond) + } + assert.Equal(t, int64(1), most.Load()) +} + +// Nothing captures once Stop returns, a pass in progress is waited out +func TestStatsServer_Stop_waitsForThePass(t *testing.T) { + var inPass atomic.Bool + entered := make(chan struct{}, 1) + l := slog.New(slog.DiscardHandler) + c := config.NewC(l) + setStatsConfig(c, map[string]any{"type": "graphite", "interval": "1h", "host": "127.0.0.1:" + freeTCPPort(t)}) + var s *statsServer + ctx := stopAtCleanup(t, &s) + s, err := newStatsServerFromConfig(ctx, l, c, "", false, func() { + inPass.Store(true) + select { + case entered <- struct{}{}: + default: + } + time.Sleep(100 * time.Millisecond) + inPass.Store(false) + }) + require.NoError(t, err) + + go s.Start() + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("never captured") + } + s.Stop() + assert.False(t, inPass.Load(), "Stop returned with a pass still running") +} + +// A Start that arrives while Stop waits out the last pass doesn't capture alongside it +func TestStatsServer_Start_duringStopWaitsForThePass(t *testing.T) { + var active, most atomic.Int64 + entered := make(chan struct{}, 1) + l := slog.New(slog.DiscardHandler) + c := config.NewC(l) + setStatsConfig(c, map[string]any{"type": "graphite", "interval": "1h", "host": "127.0.0.1:" + freeTCPPort(t)}) + var s *statsServer + ctx := stopAtCleanup(t, &s) + s, err := newStatsServerFromConfig(ctx, l, c, "", false, func() { + n := active.Add(1) + for { + m := most.Load() + if n <= m || most.CompareAndSwap(m, n) { + break + } + } + select { + case entered <- struct{}{}: + default: + } + time.Sleep(200 * time.Millisecond) + active.Add(-1) + }) + require.NoError(t, err) + + go s.Start() + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("never captured") + } + stopped := make(chan struct{}) + go func() { + s.Stop() + close(stopped) + }() + // Stop holds runMu for its whole wait, the next Start has to arrive after it took it + waitFor(t, func() bool { + if s.runMu.TryLock() { + s.runMu.Unlock() + return false + } + return true + }) + go s.Start() + select { + case <-stopped: + case <-time.After(5 * time.Second): + t.Fatal("Stop did not return") + } + waitFor(t, func() bool { return len(entered) == 1 }) + assert.Equal(t, int64(1), most.Load(), "a Start captured alongside the pass Stop was waiting out") +}