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) + } +}