feat(util): add key based debouncer (#102073)
This commit is contained in:
Jean-Philippe Quéméner
@@ -0,0 +1,310 @@
|
||||
package debouncer
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/grafana/dskit/instrument"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"github.com/prometheus/client_golang/prometheus/promauto"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrBufferFull = errors.New("debouncer buffer full")
|
||||
)
|
||||
|
||||
type ProcessFunc[T comparable] func(context.Context, T) error
|
||||
type ErrorFunc[T comparable] func(T, error)
|
||||
|
||||
type metrics struct {
|
||||
itemsAddedCounter prometheus.Counter
|
||||
itemsDroppedCounter prometheus.Counter
|
||||
itemsProcessedCounter prometheus.Counter
|
||||
processingErrorsCounter prometheus.Counter
|
||||
processingDurationHistogram prometheus.Histogram
|
||||
}
|
||||
|
||||
func newMetrics(reg prometheus.Registerer, name string) *metrics {
|
||||
return &metrics{
|
||||
itemsAddedCounter: promauto.With(reg).NewCounter(prometheus.CounterOpts{
|
||||
Name: "debouncer_items_added_total",
|
||||
Help: "Total number of items added to the debouncer",
|
||||
ConstLabels: prometheus.Labels{
|
||||
"name": name,
|
||||
},
|
||||
}),
|
||||
itemsDroppedCounter: promauto.With(reg).NewCounter(prometheus.CounterOpts{
|
||||
Name: "debouncer_items_dropped_total",
|
||||
Help: "Total number of items dropped due to a full buffer",
|
||||
ConstLabels: prometheus.Labels{
|
||||
"name": name,
|
||||
},
|
||||
}),
|
||||
itemsProcessedCounter: promauto.With(reg).NewCounter(prometheus.CounterOpts{
|
||||
Name: "debouncer_items_processed_total",
|
||||
Help: "Total number of items processed by the debouncer",
|
||||
ConstLabels: prometheus.Labels{
|
||||
"name": name,
|
||||
},
|
||||
}),
|
||||
processingErrorsCounter: promauto.With(reg).NewCounter(prometheus.CounterOpts{
|
||||
Name: "debouncer_processing_errors_total",
|
||||
Help: "Total number of errors during processing",
|
||||
ConstLabels: prometheus.Labels{
|
||||
"name": name,
|
||||
},
|
||||
}),
|
||||
processingDurationHistogram: promauto.With(reg).NewHistogram(prometheus.HistogramOpts{
|
||||
Name: "debouncer_processing_duration_seconds",
|
||||
Help: "Time taken to process items",
|
||||
Buckets: instrument.DefBuckets,
|
||||
NativeHistogramBucketFactor: 1.1,
|
||||
NativeHistogramMaxBucketNumber: 160,
|
||||
NativeHistogramMinResetDuration: time.Hour,
|
||||
ConstLabels: prometheus.Labels{
|
||||
"name": name,
|
||||
},
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
// DebouncerOpts hold all the options to create a debouncer group.
|
||||
type DebouncerOpts[T comparable] struct {
|
||||
// Name should be a unique name for this debouncer group. It is
|
||||
// also used a name label value for the metrics.
|
||||
Name string
|
||||
// BufferSize is the maximum number of pending events to buffer.
|
||||
BufferSize int
|
||||
|
||||
// ErrorHandler is the function that is called when a process for a given
|
||||
// key returns an error while running.
|
||||
ErrorHandler ErrorFunc[T]
|
||||
// ProcessHandler is the function that is called once a process for a given
|
||||
// key should be run.
|
||||
ProcessHandler ProcessFunc[T]
|
||||
// MinWait is the cooldown period after receiving an event. If another event with the
|
||||
// same key arrives during this period, the timer resets and we wait another MinWait duration.
|
||||
MinWait time.Duration
|
||||
// MaxWait is the maximum time any event will wait before processing. Even if new events
|
||||
// for the same key keep arriving, we guarantee processing after MaxWait from the first event.
|
||||
MaxWait time.Duration
|
||||
Reg prometheus.Registerer
|
||||
}
|
||||
|
||||
type Group[T comparable] struct {
|
||||
buffer chan T
|
||||
|
||||
// mutex protecting the debouncers map.
|
||||
debouncersMu sync.Mutex
|
||||
debouncers map[T]*debouncer[T]
|
||||
|
||||
wg sync.WaitGroup
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
errorHandler ErrorFunc[T]
|
||||
processHandler ProcessFunc[T]
|
||||
minWait time.Duration
|
||||
maxWait time.Duration
|
||||
metrics *metrics
|
||||
}
|
||||
|
||||
// NewGroup creates a new debouncer group for processing events with unique keys.
|
||||
//
|
||||
// A debouncer group helps optimize expensive operations by:
|
||||
// 1. Grouping identical events that occur in rapid succession
|
||||
// 2. Processing each unique key only once after waiting periods expire
|
||||
//
|
||||
// Example usage:
|
||||
//
|
||||
// group := debouncer.NewGroup(DebouncerOpts[string]{
|
||||
// BufferSize: 1000,
|
||||
// ProcessHandler: func(ctx context.Context, key string) error {
|
||||
// // This is where you perform the expensive operation
|
||||
// return doSuperExpensiveCommand(key)
|
||||
// }
|
||||
// MinWait: time.Second * 10,
|
||||
// MaxWait: time.Minute,
|
||||
// })
|
||||
//
|
||||
// // Start the debouncer group.
|
||||
// group.Start(ctx)
|
||||
//
|
||||
// // Queue events
|
||||
// if err := group.Add("user-1"); err != nil {
|
||||
// // Do something with the error.
|
||||
// }
|
||||
// // Adding the same key resets MinWait but not MaxWait
|
||||
// if err := group.Add("user-1"); err != nil {
|
||||
// // Do something with the error.
|
||||
// }
|
||||
//
|
||||
// The event will be processed when either MinWait expires (after the most recent add)
|
||||
// or MaxWait expires (after the first add), whichever comes first.
|
||||
func NewGroup[T comparable](opts DebouncerOpts[T]) (*Group[T], error) {
|
||||
if opts.BufferSize <= 0 {
|
||||
opts.BufferSize = 100
|
||||
}
|
||||
|
||||
if opts.MinWait <= 0 {
|
||||
opts.MinWait = time.Minute
|
||||
}
|
||||
if opts.MaxWait <= 0 {
|
||||
opts.MaxWait = 5 * time.Minute
|
||||
}
|
||||
if opts.MinWait > opts.MaxWait {
|
||||
return nil, errors.New("minWait is bigger than maxWait")
|
||||
}
|
||||
|
||||
if opts.ProcessHandler == nil {
|
||||
return nil, errors.New("processHandler is required")
|
||||
}
|
||||
|
||||
if opts.ErrorHandler == nil {
|
||||
opts.ErrorHandler = func(_ T, _ error) {}
|
||||
}
|
||||
|
||||
return &Group[T]{
|
||||
buffer: make(chan T, opts.BufferSize),
|
||||
debouncers: make(map[T]*debouncer[T]),
|
||||
processHandler: opts.ProcessHandler,
|
||||
errorHandler: opts.ErrorHandler,
|
||||
minWait: opts.MinWait,
|
||||
maxWait: opts.MaxWait,
|
||||
metrics: newMetrics(opts.Reg, opts.Name),
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Add will create a new debouncer for the given Key if it doesn't exist yet.
|
||||
// If a key has already a debouncer it will either reset the MinWait timer for
|
||||
// this key, or if they key is already running its process be no-op.
|
||||
func (g *Group[T]) Add(value T) error {
|
||||
select {
|
||||
case g.buffer <- value:
|
||||
g.metrics.itemsAddedCounter.Inc()
|
||||
return nil
|
||||
default:
|
||||
g.metrics.itemsDroppedCounter.Inc()
|
||||
return ErrBufferFull
|
||||
}
|
||||
}
|
||||
|
||||
func (g *Group[T]) Start(ctx context.Context) {
|
||||
g.ctx, g.cancel = context.WithCancel(ctx)
|
||||
g.wg.Add(1)
|
||||
go func() {
|
||||
defer g.wg.Done()
|
||||
for {
|
||||
select {
|
||||
case <-g.ctx.Done():
|
||||
return
|
||||
case value := <-g.buffer:
|
||||
g.processValue(value)
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
func (g *Group[T]) Stop() {
|
||||
if g.cancel != nil {
|
||||
g.cancel()
|
||||
g.wg.Wait()
|
||||
}
|
||||
}
|
||||
|
||||
func (g *Group[T]) processValue(key T) {
|
||||
g.debouncersMu.Lock()
|
||||
deb, ok := g.debouncers[key]
|
||||
if !ok {
|
||||
deb = newDebouncer[T](g.minWait, g.maxWait, key, func(v T) {
|
||||
g.processWithMetrics(g.ctx, v, g.processHandler)
|
||||
|
||||
g.debouncersMu.Lock()
|
||||
defer g.debouncersMu.Unlock()
|
||||
if current, exists := g.debouncers[key]; exists && current == deb {
|
||||
delete(g.debouncers, key)
|
||||
}
|
||||
})
|
||||
g.wg.Add(1)
|
||||
go func() {
|
||||
defer g.wg.Done()
|
||||
deb.run(g.ctx)
|
||||
}()
|
||||
g.debouncers[key] = deb
|
||||
}
|
||||
g.debouncersMu.Unlock()
|
||||
|
||||
deb.reset()
|
||||
}
|
||||
|
||||
func (g *Group[T]) processWithMetrics(ctx context.Context, value T, processFunc ProcessFunc[T]) {
|
||||
timer := prometheus.NewTimer(g.metrics.processingDurationHistogram)
|
||||
defer timer.ObserveDuration()
|
||||
g.metrics.itemsProcessedCounter.Inc()
|
||||
|
||||
if err := processFunc(ctx, value); err != nil {
|
||||
g.errorHandler(value, err)
|
||||
g.metrics.processingErrorsCounter.Inc()
|
||||
}
|
||||
}
|
||||
|
||||
// debouncer handles debouncing for a specific key.
|
||||
type debouncer[T comparable] struct {
|
||||
key T
|
||||
resetChan chan struct{}
|
||||
minWait time.Duration
|
||||
maxWait time.Duration
|
||||
processFunc func(T)
|
||||
}
|
||||
|
||||
// newDebouncer creates a new key debouncer.
|
||||
func newDebouncer[T comparable](minWait, maxWait time.Duration, key T, processFunc func(T)) *debouncer[T] {
|
||||
deb := &debouncer[T]{
|
||||
key: key,
|
||||
resetChan: make(chan struct{}, 1),
|
||||
minWait: minWait,
|
||||
maxWait: maxWait,
|
||||
processFunc: processFunc,
|
||||
}
|
||||
return deb
|
||||
}
|
||||
|
||||
// reset triggers a timer reset for the minWait.
|
||||
func (d *debouncer[T]) reset() {
|
||||
select {
|
||||
case d.resetChan <- struct{}{}:
|
||||
// Value sent successfully.
|
||||
default:
|
||||
// Value was dropped. Is not an issue as
|
||||
// a reset is already about to being processed
|
||||
// or the process is being run.
|
||||
}
|
||||
}
|
||||
|
||||
// run manages the debouncing process for a specific key.
|
||||
func (d *debouncer[T]) run(ctx context.Context) {
|
||||
// Create timers after getting the first updateChan.
|
||||
minTimer := time.NewTimer(d.minWait)
|
||||
maxTimer := time.NewTimer(d.maxWait)
|
||||
defer func() {
|
||||
minTimer.Stop()
|
||||
maxTimer.Stop()
|
||||
}()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-d.resetChan:
|
||||
minTimer.Stop()
|
||||
minTimer.Reset(d.minWait)
|
||||
case <-minTimer.C:
|
||||
d.processFunc(d.key)
|
||||
return
|
||||
case <-maxTimer.C:
|
||||
d.processFunc(d.key)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,217 @@
|
||||
package debouncer
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"github.com/prometheus/client_golang/prometheus/testutil"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestDebouncer(t *testing.T) {
|
||||
t.Run("should process values after min wait", func(t *testing.T) {
|
||||
var processedMu sync.Mutex
|
||||
processedValues := make(map[string]int)
|
||||
|
||||
group, err := NewGroup(DebouncerOpts[string]{
|
||||
BufferSize: 10,
|
||||
ProcessHandler: func(ctx context.Context, value string) error {
|
||||
processedMu.Lock()
|
||||
processedValues[value]++
|
||||
processedMu.Unlock()
|
||||
return nil
|
||||
},
|
||||
MinWait: 10 * time.Millisecond,
|
||||
MaxWait: 500 * time.Millisecond,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
group.Start(ctx)
|
||||
|
||||
require.NoError(t, group.Add("key1"))
|
||||
require.NoError(t, group.Add("key2"))
|
||||
// Should be deduplicated.
|
||||
require.NoError(t, group.Add("key1"))
|
||||
|
||||
require.Eventually(t, func() bool {
|
||||
// We should have processed key1 and key2 exactly once.
|
||||
processedMu.Lock()
|
||||
if processedValues["key1"] == 1 && processedValues["key2"] == 1 {
|
||||
return true
|
||||
}
|
||||
processedMu.Unlock()
|
||||
return false
|
||||
}, time.Millisecond*200, time.Millisecond*20)
|
||||
})
|
||||
|
||||
t.Run("should process values after max wait", func(t *testing.T) {
|
||||
processed := make(map[string]int, 1)
|
||||
|
||||
group, err := NewGroup(DebouncerOpts[string]{
|
||||
BufferSize: 10,
|
||||
ProcessHandler: func(ctx context.Context, value string) error {
|
||||
processed[value]++
|
||||
return nil
|
||||
},
|
||||
MinWait: 50 * time.Millisecond,
|
||||
MaxWait: 500 * time.Millisecond,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
group.Start(ctx)
|
||||
|
||||
ticker := time.NewTicker(time.Millisecond * 40)
|
||||
defer ticker.Stop()
|
||||
|
||||
start := time.Now()
|
||||
|
||||
for counter := 0; counter < 25; counter++ {
|
||||
<-ticker.C
|
||||
_ = group.Add("key1")
|
||||
if processed["key1"] == 1 {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
require.WithinDuration(t, start.Add(time.Millisecond*500), time.Now(), time.Millisecond*100)
|
||||
})
|
||||
|
||||
t.Run("should handle buffer full", func(t *testing.T) {
|
||||
group, err := NewGroup(DebouncerOpts[string]{
|
||||
BufferSize: 1,
|
||||
ProcessHandler: func(ctx context.Context, value string) error { return nil },
|
||||
MinWait: 10 * time.Millisecond,
|
||||
MaxWait: 100 * time.Millisecond,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
require.NoError(t, group.Add("key1"))
|
||||
// Buffer should be full by now as we are not reading from it yet.
|
||||
require.ErrorIs(t, group.Add("key2"), ErrBufferFull)
|
||||
})
|
||||
|
||||
t.Run("should track metrics", func(t *testing.T) {
|
||||
var wg sync.WaitGroup
|
||||
|
||||
group, err := NewGroup(DebouncerOpts[string]{
|
||||
BufferSize: 10,
|
||||
ProcessHandler: func(ctx context.Context, value string) error {
|
||||
wg.Done()
|
||||
return nil
|
||||
},
|
||||
MinWait: 10 * time.Millisecond,
|
||||
MaxWait: 100 * time.Millisecond,
|
||||
|
||||
Name: "test",
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
group.Start(ctx)
|
||||
|
||||
wg.Add(1)
|
||||
require.NoError(t, group.Add("key1"))
|
||||
require.NoError(t, group.Add("key1"))
|
||||
|
||||
wg.Wait()
|
||||
|
||||
require.Equal(t, float64(2), testutil.ToFloat64(group.metrics.itemsAddedCounter))
|
||||
require.Equal(t, float64(1), testutil.ToFloat64(group.metrics.itemsProcessedCounter))
|
||||
})
|
||||
|
||||
t.Run("should handle errors", func(t *testing.T) {
|
||||
var (
|
||||
wg sync.WaitGroup
|
||||
errs = make(chan error, 10)
|
||||
expectedErr = errors.New("test error")
|
||||
)
|
||||
|
||||
group, err := NewGroup(DebouncerOpts[string]{
|
||||
BufferSize: 10,
|
||||
ProcessHandler: func(ctx context.Context, value string) error {
|
||||
wg.Done()
|
||||
return expectedErr
|
||||
},
|
||||
MinWait: 10 * time.Millisecond,
|
||||
MaxWait: 100 * time.Millisecond,
|
||||
Reg: prometheus.NewPedanticRegistry(),
|
||||
Name: "test_errors",
|
||||
ErrorHandler: func(_ string, err error) { errs <- err },
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
group.Start(ctx)
|
||||
|
||||
wg.Add(1)
|
||||
require.NoError(t, group.Add("key1"))
|
||||
|
||||
wg.Wait()
|
||||
|
||||
select {
|
||||
case err := <-errs:
|
||||
require.Equal(t, expectedErr, err)
|
||||
default:
|
||||
t.Fatal("expected error")
|
||||
}
|
||||
|
||||
require.Equal(t, float64(1), testutil.ToFloat64(group.metrics.processingErrorsCounter))
|
||||
})
|
||||
|
||||
t.Run("should gracefully handle stops", func(t *testing.T) {
|
||||
// Create a channel to signal when processing is done.
|
||||
done := make(chan struct{})
|
||||
|
||||
group, err := NewGroup(DebouncerOpts[string]{
|
||||
BufferSize: 10,
|
||||
ProcessHandler: func(ctx context.Context, item string) error {
|
||||
// Start a goroutine to wait for context cancellation.
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
close(done)
|
||||
}()
|
||||
return nil
|
||||
},
|
||||
MinWait: 50 * time.Millisecond,
|
||||
MaxWait: 500 * time.Millisecond,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
// Start the group with a context
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
group.Start(ctx)
|
||||
|
||||
// Send an item to trigger processing.
|
||||
require.NoError(t, group.Add("key-1"))
|
||||
|
||||
// Give the group a moment to process the item.
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
// Stop the group, which should cancel the context.
|
||||
group.Stop()
|
||||
|
||||
// Wait for the done signal or timeout.
|
||||
select {
|
||||
case <-done:
|
||||
// Success - the group was stopped and the context was canceled
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("Timed out waiting for group to stop")
|
||||
}
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user