From efaa617e6dfa29f0bda658a7f31ff84f404e665f Mon Sep 17 00:00:00 2001 From: mathieuHa Date: Sun, 11 Oct 2026 13:57:12 +0200 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B=20keep=20the=20stream=20and=20metr?= =?UTF-8?q?ics=20tickers=20on=20their=20own=20channels=20(#399)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Traefik runs the plugin under yaegi, which shares the case list of a select statement between every goroutine that runs it. Both ticker loops ran the same select, so when the tickers fired together one loop could end up waiting on the other's channel, dropping stream ticks for up to two metrics intervals. Range over the ticker channel instead, which yaegi evaluates per execution. The stop channel was never used. Fixes #377 Co-authored-by: Claude Opus 5.5 (1M context) --- bouncer.go | 31 +++++++++++++------------------ ticker_test.go | 37 +++++++++++++++++++++++++++++++++++++ 2 files changed, 50 insertions(+), 18 deletions(-) create mode 100644 ticker_test.go diff --git a/bouncer.go b/bouncer.go index c547242..901e702 100644 --- a/bouncer.go +++ b/bouncer.go @@ -70,8 +70,8 @@ var ( isCrowdsecStreamStartup = true isCrowdsecStreamHealthy = true updateFailure int64 - streamTicker chan bool - metricsTicker chan bool + streamTicker *time.Ticker + metricsTicker *time.Ticker lastMetricsPush time.Time blockedRequests int64 ) @@ -324,7 +324,7 @@ func New(_ context.Context, next http.Handler, config *configuration.Config, nam } else { go handleStreamTicker(bouncer) } - streamTicker = startTicker("stream", config.UpdateIntervalSeconds, log, func() { + streamTicker = startTicker(config.UpdateIntervalSeconds, func() { handleStreamTicker(bouncer) }) } @@ -333,7 +333,7 @@ func New(_ context.Context, next http.Handler, config *configuration.Config, nam if metricsTicker == nil && config.MetricsUpdateIntervalSeconds > 0 { lastMetricsPush = time.Now() // Initialize lastMetricsPush when starting the metrics ticker go handleMetricsTicker(bouncer) - metricsTicker = startTicker("metrics", config.MetricsUpdateIntervalSeconds, log, func() { + metricsTicker = startTicker(config.MetricsUpdateIntervalSeconds, func() { handleMetricsTicker(bouncer) }) } @@ -564,21 +564,16 @@ func handleMetricsTicker(bouncer *Bouncer) { } } -func startTicker(name string, updateInterval int64, log *slog.Logger, work func()) chan bool { +func startTicker(updateInterval int64, work func()) *time.Ticker { ticker := time.NewTicker(time.Duration(updateInterval) * time.Second) - stop := make(chan bool, 1) - go func() { - defer log.Debug(name + "_ticker:stopped") - for { - select { - case <-ticker.C: - go work() - case <-stop: - return - } - } - }() - return stop + go runTicker(ticker.C, work) + return ticker +} + +func runTicker(ticks <-chan time.Time, work func()) { + for range ticks { + go work() + } } // We are now in none or live mode. diff --git a/ticker_test.go b/ticker_test.go new file mode 100644 index 0000000..6443060 --- /dev/null +++ b/ticker_test.go @@ -0,0 +1,37 @@ +package crowdsec_bouncer_traefik_plugin //nolint:revive,stylecheck + +import ( + "sync" + "sync/atomic" + "testing" + "time" +) + +func sendTicks(ticks chan time.Time, count int) { + for i := 0; i < count; i++ { //nolint:intrange + ticks <- time.Time{} + } +} + +func Test_runTicker_keepsEachTickerOnItsOwnChannel(t *testing.T) { + const count = 20000 + var streamRuns, metricsRuns int64 + var runs sync.WaitGroup + runs.Add(2 * count) + streamTicks := make(chan time.Time) + metricsTicks := make(chan time.Time) + go runTicker(streamTicks, func() { + atomic.AddInt64(&streamRuns, 1) + runs.Done() + }) + go runTicker(metricsTicks, func() { + atomic.AddInt64(&metricsRuns, 1) + runs.Done() + }) + go sendTicks(streamTicks, count) + go sendTicks(metricsTicks, count) + runs.Wait() + if streamRuns != count || metricsRuns != count { + t.Errorf("runTicker ran stream %d and metrics %d times, want %d each", streamRuns, metricsRuns, count) + } +}