From 5626dc50f86039aeab62bae9ef764c9bcc6244df Mon Sep 17 00:00:00 2001 From: Georges Chaudy Date: Thu, 27 Nov 2025 18:42:01 +0100 Subject: [PATCH] feat(unified-storage): Add adaptive backoff to event notifier polling (#114401) * use exponential backoff in notifier * Enhance BadgerDB configuration in REST options with memory table size and number of memtables * Enhance BadgerDB configuration in REST options by adding value threshold for LSM vs value log storage --- pkg/storage/unified/apistore/restoptions.go | 3 ++ pkg/storage/unified/resource/notifier.go | 40 +++++++++++++++---- pkg/storage/unified/resource/notifier_test.go | 16 +++++--- 3 files changed, 46 insertions(+), 13 deletions(-) diff --git a/pkg/storage/unified/apistore/restoptions.go b/pkg/storage/unified/apistore/restoptions.go index b48fbc6deaf..d880c663f29 100644 --- a/pkg/storage/unified/apistore/restoptions.go +++ b/pkg/storage/unified/apistore/restoptions.go @@ -54,6 +54,9 @@ func NewRESTOptionsGetterMemory(originalStorageConfig storagebackend.Config, sec // Create BadgerDB with in-memory mode db, err := badger.Open(badger.DefaultOptions(""). WithInMemory(true). + WithMemTableSize(256 << 10). // 256KB memtable size + WithValueThreshold(16 << 10). // 16KB threshold for storing values in LSM vs value log + WithNumMemtables(2). // Keep only 2 memtables in memory WithLogger(nil)) if err != nil { return nil, err diff --git a/pkg/storage/unified/resource/notifier.go b/pkg/storage/unified/resource/notifier.go index f55db60623a..5dd6a17ad29 100644 --- a/pkg/storage/unified/resource/notifier.go +++ b/pkg/storage/unified/resource/notifier.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" + "github.com/grafana/dskit/backoff" "github.com/grafana/grafana-app-sdk/logging" gocache "github.com/patrickmn/go-cache" @@ -13,8 +14,8 @@ import ( const ( defaultLookbackPeriod = 30 * time.Second - defaultPollInterval = 100 * time.Millisecond - defaultEventCacheSize = 10000 + defaultMinBackoff = 100 * time.Millisecond + defaultMaxBackoff = 5 * time.Second defaultBufferSize = 10000 ) @@ -29,15 +30,17 @@ type notifierOptions struct { type watchOptions struct { LookbackPeriod time.Duration // How far back to look for events - PollInterval time.Duration // How often to poll for new events BufferSize int // How many events to buffer + MinBackoff time.Duration // Minimum interval between polling requests + MaxBackoff time.Duration // Maximum interval between polling requests } func defaultWatchOptions() watchOptions { return watchOptions{ LookbackPeriod: defaultLookbackPeriod, - PollInterval: defaultPollInterval, BufferSize: defaultBufferSize, + MinBackoff: defaultMinBackoff, + MaxBackoff: defaultMaxBackoff, } } @@ -62,9 +65,13 @@ func (n *notifier) cacheKey(evt Event) string { } func (n *notifier) Watch(ctx context.Context, opts watchOptions) <-chan Event { - if opts.PollInterval <= 0 { - opts.PollInterval = defaultPollInterval + if opts.MinBackoff <= 0 { + opts.MinBackoff = defaultMinBackoff } + if opts.MaxBackoff <= 0 || opts.MaxBackoff <= opts.MinBackoff { + opts.MaxBackoff = defaultMaxBackoff + } + cacheTTL := opts.LookbackPeriod cacheCleanupInterval := 2 * opts.LookbackPeriod @@ -81,11 +88,21 @@ func (n *notifier) Watch(ctx context.Context, opts watchOptions) <-chan Event { go func() { defer close(events) + // Initialize backoff with minimum backoff interval + currentInterval := opts.MinBackoff + backoffConfig := backoff.Config{ + MinBackoff: opts.MinBackoff, + MaxBackoff: opts.MaxBackoff, + MaxRetries: 0, // infinite retries + } + bo := backoff.New(ctx, backoffConfig) + for { select { case <-ctx.Done(): return - case <-time.After(opts.PollInterval): + case <-time.After(currentInterval): + foundEvents := false for evt, err := range n.eventStore.ListSince(ctx, subtractDurationFromSnowflake(lastRV, opts.LookbackPeriod)) { if err != nil { n.log.Error("Failed to list events since", "error", err) @@ -102,6 +119,7 @@ func (n *notifier) Watch(ctx context.Context, opts watchOptions) <-chan Event { continue } + foundEvents = true if evt.ResourceVersion > lastRV { lastRV = evt.ResourceVersion + 1 } @@ -113,6 +131,14 @@ func (n *notifier) Watch(ctx context.Context, opts watchOptions) <-chan Event { return } } + + // Apply backoff logic: reset to min when events are found, increase when no events + if foundEvents { + bo.Reset() + currentInterval = opts.MinBackoff + } else { + currentInterval = bo.NextDelay() + } } } }() diff --git a/pkg/storage/unified/resource/notifier_test.go b/pkg/storage/unified/resource/notifier_test.go index a4272a36698..7b201f47420 100644 --- a/pkg/storage/unified/resource/notifier_test.go +++ b/pkg/storage/unified/resource/notifier_test.go @@ -32,7 +32,6 @@ func TestDefaultWatchOptions(t *testing.T) { opts := defaultWatchOptions() assert.Equal(t, defaultLookbackPeriod, opts.LookbackPeriod) - assert.Equal(t, defaultPollInterval, opts.PollInterval) assert.Equal(t, defaultBufferSize, opts.BufferSize) } @@ -158,8 +157,9 @@ func TestNotifier_Watch_NoEvents(t *testing.T) { opts := watchOptions{ LookbackPeriod: 100 * time.Millisecond, - PollInterval: 50 * time.Millisecond, BufferSize: 10, + MinBackoff: 50 * time.Millisecond, + MaxBackoff: 500 * time.Millisecond, } events := notifier.Watch(ctx, opts) @@ -210,8 +210,9 @@ func TestNotifier_Watch_WithExistingEvents(t *testing.T) { opts := watchOptions{ LookbackPeriod: 100 * time.Millisecond, - PollInterval: 50 * time.Millisecond, BufferSize: 10, + MinBackoff: 50 * time.Millisecond, + MaxBackoff: 500 * time.Millisecond, } // Start watching @@ -265,8 +266,9 @@ func TestNotifier_Watch_EventDeduplication(t *testing.T) { opts := watchOptions{ LookbackPeriod: time.Second, - PollInterval: 20 * time.Millisecond, BufferSize: 10, + MinBackoff: 20 * time.Millisecond, + MaxBackoff: 200 * time.Millisecond, } // Start watching @@ -326,8 +328,9 @@ func TestNotifier_Watch_ContextCancellation(t *testing.T) { opts := watchOptions{ LookbackPeriod: 100 * time.Millisecond, - PollInterval: 20 * time.Millisecond, BufferSize: 10, + MinBackoff: 20 * time.Millisecond, + MaxBackoff: 200 * time.Millisecond, } events := notifier.Watch(ctx, opts) @@ -369,8 +372,9 @@ func TestNotifier_Watch_MultipleEvents(t *testing.T) { opts := watchOptions{ LookbackPeriod: time.Second, - PollInterval: 20 * time.Millisecond, BufferSize: 10, + MinBackoff: 20 * time.Millisecond, + MaxBackoff: 200 * time.Millisecond, } // Start watching