Memberlist: support for debouncing notifications

CHANGELOG.md

@@ -80,9 +80,11 @@
 * [CHANGE] Changed `ShouldLog()` function signature in `middleware.OptionalLogging` interface to `ShouldLog(context.Context) (bool, string)`: the returned `string` contains an optional reason. When reason is valued, `GRPCServerLog` adds `(<reason>)` suffix to the error. #514
 * [CHANGE] Cache: Remove superfluous `cache.RemoteCacheClient` interface and unify all caches using the `cache.Cache` interface. #520
 * [CHANGE] Updated the minimum required Go version to 1.21. #540
+* [CHANGE] Backoff: added `Backoff.ErrCause()` which is like `Backoff.Err()` but returns the context cause if backoff is terminated because the context has been canceled. #538
 * [CHANGE] memberlist: Metric `memberlist_client_messages_in_broadcast_queue` is now split into `queue="local"` and `queue="gossip"` values. #539
 * [CHANGE] memberlist: Failure to fast-join a cluster via contacting a node is now logged at `info` instead of `debug`. #585
 * [CHANGE] `Service.AddListener` and `Manager.AddListener` now return function for stopping the listener. #564
+* [CHANGE] ring: KVStrore updates now processed at most once per `-ring.update-interval`. #592
 * [FEATURE] Cache: Add support for configuring a Redis cache backend. #268 #271 #276
 * [FEATURE] Add support for waiting on the rate limiter using the new `WaitN` method. #279
 * [FEATURE] Add `log.BufferedLogger` type. #338
@@ -230,7 +232,6 @@
 * [ENHANCEMENT] Runtimeconfig: support gzip-compressed files with `.gz` extension. #571
 * [ENHANCEMENT] grpcclient: Support custom gRPC compressors. #583
 * [ENHANCEMENT] Adapt `metrics.SendSumOfGaugesPerTenant` to use `metrics.MetricOption`. #584
-* [CHANGE] Backoff: added `Backoff.ErrCause()` which is like `Backoff.Err()` but returns the context cause if backoff is terminated because the context has been canceled. #538
 * [BUGFIX] spanlogger: Support multiple tenant IDs. #59
 * [BUGFIX] Memberlist: fixed corrupted packets when sending compound messages with more than 255 messages or messages bigger than 64KB. #85
 * [BUGFIX] Ring: `ring_member_ownership_percent` and `ring_tokens_owned` metrics are not updated on scale down. #109

ring/ring.go

@@ -18,6 +18,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
+	"go.uber.org/atomic"
 
 	"github.com/grafana/dskit/flagext"
 	"github.com/grafana/dskit/internal/slices"
@@ -143,6 +144,7 @@ var (
 type Config struct {
 	KVStore              kv.Config              `yaml:"kvstore"`
 	HeartbeatTimeout     time.Duration          `yaml:"heartbeat_timeout" category:"advanced"`
+	UpdateInterval       time.Duration          `yaml:"update_interval" category:"advanced"`
 	ReplicationFactor    int                    `yaml:"replication_factor"`
 	ZoneAwarenessEnabled bool                   `yaml:"zone_awareness_enabled"`
 	ExcludedZones        flagext.StringSliceCSV `yaml:"excluded_zones" category:"advanced"`
@@ -162,6 +164,7 @@ func (cfg *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
 	cfg.KVStore.RegisterFlagsWithPrefix(prefix, "collectors/", f)
 
 	f.DurationVar(&cfg.HeartbeatTimeout, prefix+"ring.heartbeat-timeout", time.Minute, "The heartbeat timeout after which ingesters are skipped for reads/writes. 0 = never (timeout disabled).")
+	f.DurationVar(&cfg.UpdateInterval, prefix+"ring.update-interval", 250*time.Millisecond, "How often to recompute ring state when a change is detected from the KVStore.")
 	f.IntVar(&cfg.ReplicationFactor, prefix+"distributor.replication-factor", 3, "The number of ingesters to write to and read from.")
 	f.BoolVar(&cfg.ZoneAwarenessEnabled, prefix+"distributor.zone-awareness-enabled", false, "True to enable the zone-awareness and replicate ingested samples across different availability zones.")
 	f.Var(&cfg.ExcludedZones, prefix+"distributor.excluded-zones", "Comma-separated list of zones to exclude from the ring. Instances in excluded zones will be filtered out from the ring.")
@@ -215,20 +218,23 @@ type Ring struct {
 	// Number of registered instances per zone.
 	instancesCountPerZone map[string]int
 
-	// Nubmber of registered instances with tokens per zone.
+	// Number of registered instances with tokens per zone.
 	instancesWithTokensCountPerZone map[string]int
 
 	// Number of registered instances are writable and have tokens.
 	writableInstancesWithTokensCount int
 
-	// Nubmber of registered instances with tokens per zone that are writable.
+	// Number of registered instances with tokens per zone that are writable.
 	writableInstancesWithTokensCountPerZone map[string]int
 
 	// Cache of shuffle-sharded subrings per identifier. Invalidated when topology changes.
 	// If set to nil, no caching is done (used by tests, and subrings).
 	shuffledSubringCache             map[subringCacheKey]*Ring
 	shuffledSubringWithLookbackCache map[subringCacheKey]cachedSubringWithLookback[*Ring]
 
+	// The last observed update from the KV store.
+	watchKeyUpdate atomic.Pointer[Desc]
+
 	numMembersGaugeVec      *prometheus.GaugeVec
 	totalTokensGauge        prometheus.Gauge
 	oldestTimestampGaugeVec *prometheus.GaugeVec
@@ -324,18 +330,46 @@ func (r *Ring) loop(ctx context.Context) error {
 	r.updateRingMetrics()
 	r.mtx.Unlock()
 
+	// Debounce WatchKey updates, as they can be frequent enough to cause lock
+	// contention. The most recent update is the one we'll use when we
+	// periodically update the ring.
+
+	go func() {
+		t := time.NewTicker(r.cfg.UpdateInterval)
+		defer t.Stop()
+		for {
+			select {
+			case <-t.C:
+				r.observeKeyUpdate()
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
 	r.KVClient.WatchKey(ctx, r.key, func(value interface{}) bool {
 		if value == nil {
 			level.Info(r.logger).Log("msg", "ring doesn't exist in KV store yet")
 			return true
 		}
-
-		r.updateRingState(value.(*Desc))
+		r.storeKeyUpdate(value.(*Desc))
 		return true
 	})
 	return nil
 }
 
+// storeKeyUpdate stores a new watch key update for later storage in the ring.
+func (r *Ring) storeKeyUpdate(value *Desc) {
+	r.watchKeyUpdate.Store(value)
+}
+
+// observeKeyUpdate is called periodically to update the ring state if a new watch key update is available.
+func (r *Ring) observeKeyUpdate() {
+	if value := r.watchKeyUpdate.Swap(nil); value != nil {
+		r.updateRingState(value)
+	}
+}
+
 func (r *Ring) updateRingState(ringDesc *Desc) {
 	r.mtx.RLock()
 	prevRing := r.ringDesc

ring/ring_test.go

@@ -139,6 +139,7 @@ func benchmarkUpdateRingState(b *testing.B, numInstances, numTokens int, updateT
 	cfg := Config{
 		KVStore:              kv.Config{},
 		HeartbeatTimeout:     0, // get healthy stats
+		UpdateInterval:       250 * time.Millisecond,
 		ReplicationFactor:    3,
 		ZoneAwarenessEnabled: true,
 	}
@@ -3570,6 +3571,46 @@ func compareReplicationSets(first, second ReplicationSet) (added, removed []stri
 	return
 }
 
+// Verify that WatchKey updates make it to the ring, and the latest update is observed.
+func TestRingWatchKeyUpdates(t *testing.T) {
+	cfg := Config{ReplicationFactor: 3}
+	ring, err := NewWithStoreClientAndStrategy(cfg, testRingName, testRingKey, nil, NewDefaultReplicationStrategy(), prometheus.NewRegistry(), log.NewNopLogger())
+	require.NoError(t, err)
+
+	desc1 := NewDesc()
+	desc1.Ingesters = map[string]InstanceDesc{
+		"instance-1": generateRingInstanceWithInfo("instance-1", "zone-a", []uint32{1}, time.Now()),
+	}
+	desc2 := NewDesc()
+	desc2.Ingesters = map[string]InstanceDesc{
+		"instance-1": generateRingInstanceWithInfo("instance-1", "zone-a", []uint32{1}, time.Now()),
+		"instance-2": generateRingInstanceWithInfo("instance-2", "zone-a", []uint32{2}, time.Now()),
+	}
+
+	// processUpdate is a stand-in for the ring's timer goroutine that normally calls observeKeyUpdate.
+	processUpdate := func() *Desc {
+		ring.observeKeyUpdate()
+		ring.mtx.RLock()
+		defer ring.mtx.RUnlock()
+		return ring.ringDesc
+	}
+
+	assert.Empty(t, processUpdate().Ingesters, "no desc initially")
+	assert.Empty(t, processUpdate().Ingesters, "can update multiple times without any key update stored")
+
+	ring.storeKeyUpdate(desc1)
+	assert.Same(t, desc1, processUpdate())
+	assert.Same(t, desc1, processUpdate(), "no change if no new update")
+
+	ring.storeKeyUpdate(desc2)
+	assert.Same(t, desc2, processUpdate())
+	assert.Same(t, desc2, processUpdate(), "no change if no new update")
+
+	ring.storeKeyUpdate(desc1)
+	ring.storeKeyUpdate(desc2)
+	assert.Same(t, desc2, processUpdate(), "should observe last update")
+}
+
 // This test verifies that ring is getting updates, even after extending check in the loop method.
 func TestRingUpdates(t *testing.T) {
 	const (
@@ -3608,6 +3649,7 @@ func TestRingUpdates(t *testing.T) {
 			cfg := Config{
 				KVStore:           kv.Config{Mock: inmem},
 				HeartbeatTimeout:  1 * time.Minute,
+				UpdateInterval:    20 * time.Millisecond,
 				ReplicationFactor: 3,
 				ExcludedZones:     flagext.StringSliceCSV(testData.excludedZones),
 			}
@@ -3716,6 +3758,7 @@ func TestRing_ShuffleShard_Caching(t *testing.T) {
 	cfg := Config{
 		KVStore:              kv.Config{Mock: inmem},
 		HeartbeatTimeout:     1 * time.Minute,
+		UpdateInterval:       20 * time.Millisecond,
 		ReplicationFactor:    3,
 		ZoneAwarenessEnabled: true,
 	}