@cryptotaxi247 / netdata-1 / commits / 54771625c

feat(go.d/logger): add conditional and rate-limited logging (#21813)

Ilya Mashchenko committed Feb 25, 2026 at 14:00 UTC 54771625c6f33ca9392bd6dc7b48a7fb053dfcb9
10 files changed +739 -19
src/go/logger/conditional.go new
+89
@@ -0,0 +1,89 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package logger
4 +
5 +import (
6 + "fmt"
7 + "log/slog"
8 +)
9 +
10 +type ConditionalLogger struct {
11 + l *Logger
12 + cond bool
13 + allowElse bool
14 +}
15 +
16 +type ElseBranch struct {
17 + l *Logger
18 + cond bool
19 +}
20 +
21 +func (l *Logger) When(cond bool) ConditionalLogger {
22 + return ConditionalLogger{l: l, cond: cond, allowElse: true}
23 +}
24 +
25 +func (b ElseBranch) Else() ConditionalLogger {
26 + return ConditionalLogger{l: b.l, cond: b.cond}
27 +}
28 +
29 +func (c ConditionalLogger) Error(a ...any) ElseBranch {
30 + return c.logArgs(slog.LevelError, a...)
31 +}
32 +
33 +func (c ConditionalLogger) Warning(a ...any) ElseBranch {
34 + return c.logArgs(slog.LevelWarn, a...)
35 +}
36 +
37 +func (c ConditionalLogger) Notice(a ...any) ElseBranch {
38 + return c.logArgs(levelNotice, a...)
39 +}
40 +
41 +func (c ConditionalLogger) Info(a ...any) ElseBranch {
42 + return c.logArgs(slog.LevelInfo, a...)
43 +}
44 +
45 +func (c ConditionalLogger) Debug(a ...any) ElseBranch {
46 + return c.logArgs(slog.LevelDebug, a...)
47 +}
48 +
49 +func (c ConditionalLogger) Errorf(format string, a ...any) ElseBranch {
50 + return c.logf(slog.LevelError, format, a...)
51 +}
52 +
53 +func (c ConditionalLogger) Warningf(format string, a ...any) ElseBranch {
54 + return c.logf(slog.LevelWarn, format, a...)
55 +}
56 +
57 +func (c ConditionalLogger) Noticef(format string, a ...any) ElseBranch {
58 + return c.logf(levelNotice, format, a...)
59 +}
60 +
61 +func (c ConditionalLogger) Infof(format string, a ...any) ElseBranch {
62 + return c.logf(slog.LevelInfo, format, a...)
63 +}
64 +
65 +func (c ConditionalLogger) Debugf(format string, a ...any) ElseBranch {
66 + return c.logf(slog.LevelDebug, format, a...)
67 +}
68 +
69 +func (c ConditionalLogger) logArgs(level slog.Level, a ...any) ElseBranch {
70 + if c.cond && c.l.canLog(level) {
71 + c.l.log(level, fmt.Sprint(a...))
72 + }
73 + return c.next()
74 +}
75 +
76 +func (c ConditionalLogger) logf(level slog.Level, format string, a ...any) ElseBranch {
77 + if c.cond && c.l.canLog(level) {
78 + c.l.log(level, fmt.Sprintf(format, a...))
79 + }
80 + return c.next()
81 +}
82 +
83 +func (c ConditionalLogger) next() ElseBranch {
84 + cond := false
85 + if c.allowElse {
86 + cond = !c.cond
87 + }
88 + return ElseBranch{l: c.l, cond: cond}
89 +}
src/go/logger/conditional_test.go new
+67
@@ -0,0 +1,67 @@
1 +package logger
2 +
3 +import (
4 + "log/slog"
5 + "sync/atomic"
6 + "testing"
7 +
8 + "github.com/stretchr/testify/assert"
9 +)
10 +
11 +type formatProbe struct {
12 + hits *atomic.Int32
13 +}
14 +
15 +func (p formatProbe) String() string {
16 + p.hits.Add(1)
17 + return "probe"
18 +}
19 +
20 +func TestWhenElseLogsOnlyChosenBranch(t *testing.T) {
21 + setTestLevel(t, slog.LevelDebug)
22 +
23 + l, h := newTestLogger(slog.LevelDebug)
24 +
25 + l.When(true).Warning("warn1").Else().Info("info1")
26 + assert.Equal(t, 1, h.count())
27 + assert.Equal(t, slog.LevelWarn, h.last().level)
28 + assert.Equal(t, "warn1", h.last().msg)
29 +
30 + l.When(false).Warning("warn2").Else().Info("info2")
31 + assert.Equal(t, 2, h.count())
32 + assert.Equal(t, slog.LevelInfo, h.last().level)
33 + assert.Equal(t, "info2", h.last().msg)
34 +}
35 +
36 +func TestWhenElseSkipsFormattingOnSuppressedBranch(t *testing.T) {
37 + setTestLevel(t, slog.LevelDebug)
38 +
39 + l, h := newTestLogger(slog.LevelDebug)
40 + var firstHits atomic.Int32
41 + var secondHits atomic.Int32
42 +
43 + l.When(true).Warningf("warn %s", formatProbe{hits: &firstHits}).Else().Infof("info %s", formatProbe{hits: &secondHits})
44 +
45 + assert.Equal(t, int32(1), firstHits.Load())
46 + assert.Equal(t, int32(0), secondHits.Load())
47 + assert.Equal(t, 1, h.count())
48 + assert.Equal(t, slog.LevelWarn, h.last().level)
49 +}
50 +
51 +func TestWhenWithoutElseDoesNotLogWhenConditionFalse(t *testing.T) {
52 + setTestLevel(t, slog.LevelDebug)
53 +
54 + l, h := newTestLogger(slog.LevelDebug)
55 + l.When(false).Infof("suppressed %d", 1)
56 +
57 + assert.Equal(t, 0, h.count())
58 +}
59 +
60 +func TestWhenNilLoggerDoesNotPanic(t *testing.T) {
61 + setTestLevel(t, slog.LevelDebug)
62 +
63 + var l *Logger
64 + assert.NotPanics(t, func() {
65 + l.When(true).Noticef("hello %s", "world").Else().Info("ignored")
66 + })
67 +}
src/go/logger/default.go
+11 -2
@@ -5,6 +5,7 @@ package logger
5 import (
6 "log/slog"
7 "os"
8 + "time"
9
10 "github.com/mattn/go-isatty"
11 )
@@ -12,19 +13,27 @@ import (
13 func newDefaultLogger() *Logger {
14 if isatty.IsTerminal(os.Stderr.Fd()) {
15 // skip 2 slog pkg calls, 3 this pkg calls
15 - return &Logger{sl: slog.New(withCallDepth(5, newTerminalHandler()))}
16 + return &Logger{sl: slog.New(withCallDepth(5, newTerminalHandler())), rl: newRateLimiter()}
17 }
17 - return &Logger{sl: slog.New(newTextHandler()).With(pluginAttr)}
18 + return &Logger{sl: slog.New(newTextHandler()).With(pluginAttr), rl: newRateLimiter()}
19 }
20
21 var defaultLogger = newDefaultLogger()
22
23 func Error(a ...any) { defaultLogger.Error(a...) }
24 func Warning(a ...any) { defaultLogger.Warning(a...) }
25 +func Notice(a ...any) { defaultLogger.Notice(a...) }
26 func Info(a ...any) { defaultLogger.Info(a...) }
27 func Debug(a ...any) { defaultLogger.Debug(a...) }
28 func Errorf(format string, a ...any) { defaultLogger.Errorf(format, a...) }
29 func Warningf(format string, a ...any) { defaultLogger.Warningf(format, a...) }
30 +func Noticef(format string, a ...any) { defaultLogger.Noticef(format, a...) }
31 func Infof(format string, a ...any) { defaultLogger.Infof(format, a...) }
32 func Debugf(format string, a ...any) { defaultLogger.Debugf(format, a...) }
33 func With(args ...any) *Logger { return defaultLogger.With(args...) }
34 +func When(cond bool) ConditionalLogger { return defaultLogger.When(cond) }
35 +func Once(key string) LimitedLogger { return defaultLogger.Once(key) }
36 +func Limit(key string, n int, d time.Duration) LimitedLogger {
37 + return defaultLogger.Limit(key, n, d)
38 +}
39 +func ResetAllOnce() { defaultLogger.ResetAllOnce() }
src/go/logger/logger.go
+88 -16
@@ -23,35 +23,97 @@ var pluginAttr = slog.String("plugin", executable.Name)
23 func New() *Logger {
24 if isTerm {
25 // skip 2 slog pkg calls, 2 this pkg calls
26 - return &Logger{sl: slog.New(withCallDepth(4, newTerminalHandler()))}
26 + return &Logger{sl: slog.New(withCallDepth(4, newTerminalHandler())), rl: newRateLimiter()}
27 }
28 - return &Logger{sl: slog.New(newTextHandler()).With(pluginAttr)}
28 + return &Logger{sl: slog.New(newTextHandler()).With(pluginAttr), rl: newRateLimiter()}
29 }
30
31 type Logger struct {
32 muted atomic.Bool
33 sl *slog.Logger
34 + rl *rateLimiter
35 }
36
36 -func (l *Logger) Error(a ...any) { l.log(slog.LevelError, fmt.Sprint(a...)) }
37 -func (l *Logger) Warning(a ...any) { l.log(slog.LevelWarn, fmt.Sprint(a...)) }
38 -func (l *Logger) Notice(a ...any) { l.log(levelNotice, fmt.Sprint(a...)) }
39 -func (l *Logger) Info(a ...any) { l.log(slog.LevelInfo, fmt.Sprint(a...)) }
40 -func (l *Logger) Debug(a ...any) { l.log(slog.LevelDebug, fmt.Sprint(a...)) }
41 -func (l *Logger) Errorf(format string, a ...any) { l.log(slog.LevelError, fmt.Sprintf(format, a...)) }
42 -func (l *Logger) Warningf(format string, a ...any) { l.log(slog.LevelWarn, fmt.Sprintf(format, a...)) }
43 -func (l *Logger) Noticef(format string, a ...any) { l.log(levelNotice, fmt.Sprintf(format, a...)) }
44 -func (l *Logger) Infof(format string, a ...any) { l.log(slog.LevelInfo, fmt.Sprintf(format, a...)) }
45 -func (l *Logger) Debugf(format string, a ...any) { l.log(slog.LevelDebug, fmt.Sprintf(format, a...)) }
46 -func (l *Logger) Mute() { l.mute(true) }
47 -func (l *Logger) Unmute() { l.mute(false) }
37 +func (l *Logger) Error(a ...any) {
38 + if !l.canLog(slog.LevelError) {
39 + return
40 + }
41 + l.log(slog.LevelError, fmt.Sprint(a...))
42 +}
43 +
44 +func (l *Logger) Warning(a ...any) {
45 + if !l.canLog(slog.LevelWarn) {
46 + return
47 + }
48 + l.log(slog.LevelWarn, fmt.Sprint(a...))
49 +}
50 +
51 +func (l *Logger) Notice(a ...any) {
52 + if !l.canLog(levelNotice) {
53 + return
54 + }
55 + l.log(levelNotice, fmt.Sprint(a...))
56 +}
57 +
58 +func (l *Logger) Info(a ...any) {
59 + if !l.canLog(slog.LevelInfo) {
60 + return
61 + }
62 + l.log(slog.LevelInfo, fmt.Sprint(a...))
63 +}
64 +
65 +func (l *Logger) Debug(a ...any) {
66 + if !l.canLog(slog.LevelDebug) {
67 + return
68 + }
69 + l.log(slog.LevelDebug, fmt.Sprint(a...))
70 +}
71 +
72 +func (l *Logger) Errorf(format string, a ...any) {
73 + if !l.canLog(slog.LevelError) {
74 + return
75 + }
76 + l.log(slog.LevelError, fmt.Sprintf(format, a...))
77 +}
78 +
79 +func (l *Logger) Warningf(format string, a ...any) {
80 + if !l.canLog(slog.LevelWarn) {
81 + return
82 + }
83 + l.log(slog.LevelWarn, fmt.Sprintf(format, a...))
84 +}
85 +
86 +func (l *Logger) Noticef(format string, a ...any) {
87 + if !l.canLog(levelNotice) {
88 + return
89 + }
90 + l.log(levelNotice, fmt.Sprintf(format, a...))
91 +}
92 +
93 +func (l *Logger) Infof(format string, a ...any) {
94 + if !l.canLog(slog.LevelInfo) {
95 + return
96 + }
97 + l.log(slog.LevelInfo, fmt.Sprintf(format, a...))
98 +}
99 +
100 +func (l *Logger) Debugf(format string, a ...any) {
101 + if !l.canLog(slog.LevelDebug) {
102 + return
103 + }
104 + l.log(slog.LevelDebug, fmt.Sprintf(format, a...))
105 +}
106 +
107 +func (l *Logger) Mute() { l.mute(true) }
108 +func (l *Logger) Unmute() { l.mute(false) }
109
110 func (l *Logger) With(args ...any) *Logger {
111 if l.isNil() {
51 - return &Logger{sl: New().sl.With(args...)}
112 + ll := New()
113 + return &Logger{sl: ll.sl.With(args...), rl: ll.rl}
114 }
115
54 - ll := &Logger{sl: l.sl.With(args...)}
116 + ll := &Logger{sl: l.sl.With(args...), rl: l.rl}
117 ll.muted.Store(l.muted.Load())
118
119 return ll
@@ -68,6 +130,16 @@ func (l *Logger) log(level slog.Level, msg string) {
130 }
131 }
132
133 +func (l *Logger) canLog(level slog.Level) bool {
134 + if !Level.Enabled(level) {
135 + return false
136 + }
137 + if l.isNil() {
138 + return true
139 + }
140 + return !l.muted.Load()
141 +}
142 +
143 func (l *Logger) mute(v bool) {
144 if l.isNil() || isTerm && Level.Enabled(slog.LevelDebug) {
145 return
src/go/logger/logger_test.go
+8 -1
@@ -2,6 +2,7 @@ package logger
2
3 import (
4 "testing"
5 + "time"
6
7 "github.com/stretchr/testify/assert"
8 )
@@ -14,7 +15,13 @@ func TestNew(t *testing.T) {
15
16 for name, logger := range tests {
17 t.Run(name, func(t *testing.T) {
17 - f := func() { logger.Infof("test %s", "test") }
18 + f := func() {
19 + logger.Infof("test %s", "test")
20 + logger.When(true).Warning("warn").Else().Info("info")
21 + logger.Once("k").Info("once")
22 + logger.Limit("k", 1, time.Second).Info("limit")
23 + logger.ResetAllOnce()
24 + }
25 assert.NotPanics(t, f)
26 })
27 }
src/go/logger/ratelimit.go new
+215
@@ -0,0 +1,215 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package logger
4 +
5 +import (
6 + "fmt"
7 + "log/slog"
8 + "sync"
9 + "time"
10 +)
11 +
12 +const (
13 + rateLimiterSweepEvery = 4096
14 + rateLimiterTTL = time.Hour
15 +)
16 +
17 +type limitMode uint8
18 +
19 +const (
20 + modeOnce limitMode = iota + 1
21 + modeLimit
22 +)
23 +
24 +type limitKey struct {
25 + mode limitMode
26 + key string
27 +}
28 +
29 +type limitEntry struct {
30 + limit int
31 + window time.Duration
32 + count int
33 + windowStart time.Time
34 + lastSeen time.Time
35 +}
36 +
37 +type rateLimiter struct {
38 + mu sync.Mutex
39 + entries map[limitKey]*limitEntry
40 + sweepEvery uint64
41 + ttl time.Duration
42 + calls uint64
43 + now func() time.Time
44 +}
45 +
46 +func newRateLimiter() *rateLimiter {
47 + return &rateLimiter{
48 + entries: make(map[limitKey]*limitEntry),
49 + sweepEvery: rateLimiterSweepEvery,
50 + ttl: rateLimiterTTL,
51 + now: time.Now,
52 + }
53 +}
54 +
55 +type LimitedLogger struct {
56 + l *Logger
57 + mode limitMode
58 + key string
59 + limit int
60 + window time.Duration
61 +}
62 +
63 +func (l *Logger) Once(key string) LimitedLogger {
64 + return LimitedLogger{
65 + l: l,
66 + mode: modeOnce,
67 + key: key,
68 + limit: 1,
69 + window: 0,
70 + }
71 +}
72 +
73 +func (l *Logger) Limit(key string, n int, d time.Duration) LimitedLogger {
74 + if n <= 0 {
75 + n = 1
76 + }
77 + if d < 0 {
78 + d = 0
79 + }
80 + return LimitedLogger{
81 + l: l,
82 + mode: modeLimit,
83 + key: key,
84 + limit: n,
85 + window: d,
86 + }
87 +}
88 +
89 +func (l *Logger) ResetAllOnce() {
90 + if l == nil || l.rl == nil {
91 + return
92 + }
93 + l.rl.resetAllMode(modeOnce)
94 +}
95 +
96 +func (l LimitedLogger) Error(a ...any) {
97 + l.logArgs(slog.LevelError, a...)
98 +}
99 +
100 +func (l LimitedLogger) Warning(a ...any) {
101 + l.logArgs(slog.LevelWarn, a...)
102 +}
103 +
104 +func (l LimitedLogger) Notice(a ...any) {
105 + l.logArgs(levelNotice, a...)
106 +}
107 +
108 +func (l LimitedLogger) Info(a ...any) {
109 + l.logArgs(slog.LevelInfo, a...)
110 +}
111 +
112 +func (l LimitedLogger) Debug(a ...any) {
113 + l.logArgs(slog.LevelDebug, a...)
114 +}
115 +
116 +func (l LimitedLogger) Errorf(format string, a ...any) {
117 + l.logf(slog.LevelError, format, a...)
118 +}
119 +
120 +func (l LimitedLogger) Warningf(format string, a ...any) {
121 + l.logf(slog.LevelWarn, format, a...)
122 +}
123 +
124 +func (l LimitedLogger) Noticef(format string, a ...any) {
125 + l.logf(levelNotice, format, a...)
126 +}
127 +
128 +func (l LimitedLogger) Infof(format string, a ...any) {
129 + l.logf(slog.LevelInfo, format, a...)
130 +}
131 +
132 +func (l LimitedLogger) Debugf(format string, a ...any) {
133 + l.logf(slog.LevelDebug, format, a...)
134 +}
135 +
136 +func (l LimitedLogger) logArgs(level slog.Level, a ...any) {
137 + if !l.l.canLog(level) || !l.allow() {
138 + return
139 + }
140 + l.l.log(level, fmt.Sprint(a...))
141 +}
142 +
143 +func (l LimitedLogger) logf(level slog.Level, format string, a ...any) {
144 + if !l.l.canLog(level) || !l.allow() {
145 + return
146 + }
147 + l.l.log(level, fmt.Sprintf(format, a...))
148 +}
149 +
150 +func (l LimitedLogger) allow() bool {
151 + if l.l == nil || l.l.rl == nil {
152 + // Preserve nil logger behavior: no panic and no additional suppression.
153 + return true
154 + }
155 + return l.l.rl.allow(l.mode, l.key, l.limit, l.window)
156 +}
157 +
158 +func (r *rateLimiter) allow(mode limitMode, key string, limit int, window time.Duration) bool {
159 + now := r.now()
160 + allow := false
161 +
162 + r.mu.Lock()
163 + defer r.mu.Unlock()
164 +
165 + r.calls++
166 + if r.sweepEvery > 0 && r.calls%r.sweepEvery == 0 {
167 + r.sweep(now)
168 + }
169 +
170 + k := limitKey{mode: mode, key: key}
171 + e, ok := r.entries[k]
172 + if !ok {
173 + e = &limitEntry{
174 + limit: limit,
175 + window: window,
176 + windowStart: now,
177 + }
178 + r.entries[k] = e
179 + }
180 +
181 + e.lastSeen = now
182 +
183 + // d==0 means infinite window, so rollover must be disabled.
184 + if e.window > 0 && now.Sub(e.windowStart) >= e.window {
185 + e.windowStart = now
186 + e.count = 0
187 + }
188 +
189 + if e.count < e.limit {
190 + e.count++
191 + allow = true
192 + }
193 +
194 + return allow
195 +}
196 +
197 +func (r *rateLimiter) resetAllMode(mode limitMode) {
198 + r.mu.Lock()
199 + defer r.mu.Unlock()
200 +
201 + for k := range r.entries {
202 + if k.mode == mode {
203 + delete(r.entries, k)
204 + }
205 + }
206 +}
207 +
208 +func (r *rateLimiter) sweep(now time.Time) {
209 + cutoff := now.Add(-r.ttl)
210 + for k, e := range r.entries {
211 + if e.lastSeen.Before(cutoff) {
212 + delete(r.entries, k)
213 + }
214 + }
215 +}
src/go/logger/ratelimit_test.go new
+183
@@ -0,0 +1,183 @@
1 +package logger
2 +
3 +import (
4 + "log/slog"
5 + "sync"
6 + "sync/atomic"
7 + "testing"
8 + "time"
9 +
10 + "github.com/stretchr/testify/assert"
11 +)
12 +
13 +func TestLimitUsesFixedWindow(t *testing.T) {
14 + setTestLevel(t, slog.LevelDebug)
15 +
16 + l, h := newTestLogger(slog.LevelDebug)
17 + now := time.Unix(100, 0)
18 + l.rl.now = func() time.Time { return now }
19 +
20 + for i := 0; i < 3; i++ {
21 + l.Limit("k", 2, 5*time.Second).Info("msg")
22 + }
23 + assert.Equal(t, 2, h.count())
24 +
25 + now = now.Add(6 * time.Second)
26 + l.Limit("k", 2, 5*time.Second).Info("msg")
27 + assert.Equal(t, 3, h.count())
28 +}
29 +
30 +func TestLimitDZeroIsInfiniteWindow(t *testing.T) {
31 + setTestLevel(t, slog.LevelDebug)
32 +
33 + l, h := newTestLogger(slog.LevelDebug)
34 + now := time.Unix(200, 0)
35 + l.rl.now = func() time.Time { return now }
36 +
37 + l.Limit("k", 2, 0).Info("1")
38 + now = now.Add(time.Hour)
39 + l.Limit("k", 2, 0).Info("2")
40 + now = now.Add(time.Hour)
41 + l.Limit("k", 2, 0).Info("3")
42 +
43 + assert.Equal(t, 2, h.count())
44 +}
45 +
46 +func TestLimitClampsInvalidInputs(t *testing.T) {
47 + setTestLevel(t, slog.LevelDebug)
48 +
49 + l, h := newTestLogger(slog.LevelDebug)
50 + l.Limit("k", 0, -time.Second).Info("a")
51 + l.Limit("k", 0, -time.Second).Info("b")
52 +
53 + assert.Equal(t, 1, h.count())
54 +}
55 +
56 +func TestOnceIsWrapperAndSkipsFormattingWhenSuppressed(t *testing.T) {
57 + setTestLevel(t, slog.LevelDebug)
58 +
59 + l, h := newTestLogger(slog.LevelDebug)
60 + var hits atomic.Int32
61 + p := formatProbe{hits: &hits}
62 +
63 + l.Once("k").Infof("probe %s", p)
64 + l.Once("k").Infof("probe %s", p)
65 +
66 + assert.Equal(t, int32(1), hits.Load())
67 + assert.Equal(t, 1, h.count())
68 +}
69 +
70 +func TestResetAllOnceDoesNotResetLimitState(t *testing.T) {
71 + setTestLevel(t, slog.LevelDebug)
72 +
73 + l, h := newTestLogger(slog.LevelDebug)
74 + now := time.Unix(300, 0)
75 + l.rl.now = func() time.Time { return now }
76 +
77 + l.Once("k").Info("once-1")
78 + l.Limit("k", 1, time.Hour).Info("limit-1")
79 + l.Once("k").Info("once-2-suppressed")
80 + l.Limit("k", 1, time.Hour).Info("limit-2-suppressed")
81 +
82 + assert.Equal(t, 2, h.count())
83 +
84 + l.ResetAllOnce()
85 +
86 + l.Once("k").Info("once-3")
87 + l.Limit("k", 1, time.Hour).Info("limit-3-still-suppressed")
88 +
89 + assert.Equal(t, 3, h.count())
90 + assert.Equal(t, "once-3", h.last().msg)
91 +}
92 +
93 +func TestModeNamespacesAndFirstWriterWinsParams(t *testing.T) {
94 + setTestLevel(t, slog.LevelDebug)
95 +
96 + l, h := newTestLogger(slog.LevelDebug)
97 + now := time.Unix(400, 0)
98 + l.rl.now = func() time.Time { return now }
99 +
100 + // Same key in different modes should be independent.
101 + l.Once("shared").Info("once")
102 + l.Limit("shared", 1, time.Hour).Info("limit")
103 + assert.Equal(t, 2, h.count())
104 +
105 + // First writer wins params for same mode+key.
106 + l.Limit("k", 1, time.Hour).Info("first")
107 + l.Limit("k", 5, time.Millisecond).Info("second-suppressed")
108 + assert.Equal(t, 3, h.count())
109 +
110 + now = now.Add(2 * time.Hour)
111 + l.Limit("k", 5, time.Millisecond).Info("third-after-first-window")
112 + assert.Equal(t, 4, h.count())
113 +
114 + l.rl.mu.Lock()
115 + entry := l.rl.entries[limitKey{mode: modeLimit, key: "k"}]
116 + l.rl.mu.Unlock()
117 + assert.NotNil(t, entry)
118 + assert.Equal(t, 1, entry.limit)
119 + assert.Equal(t, time.Hour, entry.window)
120 +}
121 +
122 +func TestRateLimiterSweepRemovesStaleEntries(t *testing.T) {
123 + setTestLevel(t, slog.LevelDebug)
124 +
125 + l, _ := newTestLogger(slog.LevelDebug)
126 + now := time.Unix(500, 0)
127 + l.rl.now = func() time.Time { return now }
128 + l.rl.ttl = time.Second
129 + l.rl.sweepEvery = 1
130 +
131 + l.Once("stale").Info("s")
132 + now = now.Add(2 * time.Second)
133 + l.Once("fresh").Info("f")
134 +
135 + l.rl.mu.Lock()
136 + _, staleOK := l.rl.entries[limitKey{mode: modeOnce, key: "stale"}]
137 + _, freshOK := l.rl.entries[limitKey{mode: modeOnce, key: "fresh"}]
138 + l.rl.mu.Unlock()
139 +
140 + assert.False(t, staleOK)
141 + assert.True(t, freshOK)
142 +}
143 +
144 +func TestWithSharesRateLimiterState(t *testing.T) {
145 + setTestLevel(t, slog.LevelDebug)
146 +
147 + parent, h := newTestLogger(slog.LevelDebug)
148 + child := parent.With("k", "v")
149 +
150 + parent.Once("k").Info("first")
151 + child.Once("k").Info("second-suppressed")
152 +
153 + assert.Same(t, parent.rl, child.rl)
154 + assert.Equal(t, 1, h.count())
155 +}
156 +
157 +func TestOnceConcurrentLogsOnlyOnce(t *testing.T) {
158 + setTestLevel(t, slog.LevelDebug)
159 +
160 + l, h := newTestLogger(slog.LevelDebug)
161 + var wg sync.WaitGroup
162 + for i := 0; i < 100; i++ {
163 + wg.Add(1)
164 + go func() {
165 + defer wg.Done()
166 + l.Once("concurrent").Info("x")
167 + }()
168 + }
169 + wg.Wait()
170 +
171 + assert.Equal(t, 1, h.count())
172 +}
173 +
174 +func TestRateLimitNilLoggerDoesNotPanic(t *testing.T) {
175 + setTestLevel(t, slog.LevelDebug)
176 +
177 + var l *Logger
178 + assert.NotPanics(t, func() {
179 + l.Once("k").Infof("x=%d", 1)
180 + l.Limit("k", 2, time.Second).Warning("y")
181 + l.ResetAllOnce()
182 + })
183 +}
src/go/logger/testutils_test.go new
+74
@@ -0,0 +1,74 @@
1 +package logger
2 +
3 +import (
4 + "context"
5 + "log/slog"
6 + "sync"
7 + "testing"
8 +)
9 +
10 +type capturedRecord struct {
11 + level slog.Level
12 + msg string
13 +}
14 +
15 +type captureHandler struct {
16 + mu sync.Mutex
17 + minLvl slog.Level
18 + records []capturedRecord
19 +}
20 +
21 +func newCaptureHandler(minLvl slog.Level) *captureHandler {
22 + return &captureHandler{minLvl: minLvl}
23 +}
24 +
25 +func (h *captureHandler) Enabled(_ context.Context, lvl slog.Level) bool {
26 + return lvl >= h.minLvl
27 +}
28 +
29 +func (h *captureHandler) Handle(_ context.Context, r slog.Record) error {
30 + h.mu.Lock()
31 + defer h.mu.Unlock()
32 + h.records = append(h.records, capturedRecord{
33 + level: r.Level,
34 + msg: r.Message,
35 + })
36 + return nil
37 +}
38 +
39 +func (h *captureHandler) WithAttrs(_ []slog.Attr) slog.Handler {
40 + return h
41 +}
42 +
43 +func (h *captureHandler) WithGroup(_ string) slog.Handler {
44 + return h
45 +}
46 +
47 +func (h *captureHandler) count() int {
48 + h.mu.Lock()
49 + defer h.mu.Unlock()
50 + return len(h.records)
51 +}
52 +
53 +func (h *captureHandler) last() capturedRecord {
54 + h.mu.Lock()
55 + defer h.mu.Unlock()
56 + if len(h.records) == 0 {
57 + return capturedRecord{}
58 + }
59 + return h.records[len(h.records)-1]
60 +}
61 +
62 +func setTestLevel(t *testing.T, level slog.Level) {
63 + t.Helper()
64 + prev := Level.lvl.Level()
65 + Level.Set(level)
66 + t.Cleanup(func() {
67 + Level.Set(prev)
68 + })
69 +}
70 +
71 +func newTestLogger(level slog.Level) (*Logger, *captureHandler) {
72 + handler := newCaptureHandler(level)
73 + return &Logger{sl: slog.New(handler), rl: newRateLimiter()}, handler
74 +}
src/go/plugin/framework/jobruntime/job_v1.go
+2
@@ -440,6 +440,8 @@ func (j *Job) postCheck() error {
440 }
441
442 func (j *Job) runOnce() {
443 + defer j.ResetAllOnce()
444 +
445 curTime := time.Now()
446 sinceLastRun := calcSinceLastRun(curTime, j.prevRun)
447 j.prevRun = curTime
src/go/plugin/framework/jobruntime/job_v2.go
+2
@@ -329,6 +329,8 @@ func (j *JobV2) postCheck() error {
329 }
330
331 func (j *JobV2) runOnce() {
332 + defer j.ResetAllOnce()
333 +
334 j.applyPendingVnodeUpdate()
335
336 curTime := time.Now()