Repository: grafana/dskit (go)

Issue:
Potential memberlist metrics issues.
Hey all,

We are using Tempo, and trying to make use of the Tempo mixin Grafana dashboards... https://github.com/grafana/tempo/tree/main/operations/tempo-mixin

However I've noticed that some of the metrics that the dashboard expects that come from dskit/kv/memberlist/metrics might not be coming out correctly.

1. `tempo_memberlist_client_kv_store_count`
This chart https://github.com/grafana/tempo/blob/main/operations/tempo-mixin/dashboards/tempo-operational.json#L4585
is looking for the metric `tempo_memberlist_client_kv_store_count`. I couldn't find it exposed on any of the `/metrics` endpoints I looked at.

I went looking for where this metric is configured and found this: https://github.com/grafana/dskit/blob/main/kv/memberlist/metrics.go#L123

```
	m.storeValuesDesc = prometheus.NewDesc(
		prometheus.BuildFQName(m.cfg.MetricsNamespace, subsystem, "kv_store_count"), // gauge
		"Number of values in KV Store",
		nil, nil)
```

I'm not super familiar with Prometheus, but the rest of the metric definitions in that file have `NewGaugeVec` or `NewCounterVec`. From what I can tell `NewDesc` is used for metric metadata?
But the metric name "kv_store_count" sounds like it should be a numerical value?

2. `gauge_memberlist_health_score`
This chart https://github.com/grafana/tempo/blob/main/operations/tempo-mixin/dashboards/tempo-operational.json#LL4319C1-L4319C1 is looking for a metric `gauge_memberlist_health_score`. Again I couldn't find it exposed in the `/metrics` endpoint.

I went looking for where this metric is configured and I couldn't find it!

But I did find this: https://github.com/grafana/dskit/blob/main/kv/memberlist/metrics.go#L154

```
	m.memberlistHealthScore = promauto.With(m.registerer).NewGaugeFunc(prometheus.GaugeOpts{
		Namespace: m.cfg.MetricsNamespace,
		Subsystem: subsystem,
		Name:      "cluster_node_health_score",
		Help:      "Health score of this cluster. Lower value is better. 0 = healthy",
	}, func() float64 {
		// m.memberlist is not set before Starting state
		if m.State() == services.Running || m.State() == services.Stopping {
			return float64(m.memberlist.GetHealthScore())
		}
		return 0
	})
```

The Go struct member is `memberlistHealthScore` similar to the metric I'm looking for, but the actual metric name is `cluster_node_health_score`.

Could this be a metric that has been changed in code, but the metric name hasn't been updated yet?

Also I think the Tempo mixin might be wrong... `gauge_memberlist_health_score` seems like it should be `tempo_memberlist_client_cluster_node_health_score` or something?

Pull request:
memberlist metrics
**What this PR does**:

This PR fixes registration of `memberlist_client_kv_store_count` metric (this broke in #22) and also disables expiration of internal memberlist-library metrics.

This PR also removes `MetricsRegisterer` field from `memberlist.KVConfig` struct in favor of registrerer passed to `memberlist.NewKV` function.

**Which issue(s) this PR fixes**:

Fixes #305

**Checklist**
- [x] Tests updated
- [x] `CHANGELOG.md` updated - the order of entries should be `[CHANGE]`, `[FEATURE]`, `[ENHANCEMENT]`, `[BUGFIX]`


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 42feec3..73f59f0 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -46,6 +46,7 @@
 * [CHANGE] Cache: Remove the `context.Context` argument from the `Cache.Store` method and rename the method to `Cache.StoreAsync`. #273
 * [CHANGE] ring: make it harder to leak contexts when using `DoUntilQuorum`. #319
 * [CHANGE] ring: `CountTokens()`, used for the calculation of ownership in the ring page has been changed in such a way that when zone-awareness is enabled, it calculates the ownership per-zone. Previously zones were not taken into account. #325
+* [CHANGE] memberlist: `MetricsRegisterer` field has been removed from `memberlist.KVConfig` in favor of registrerer passed via into `NewKV` function. #327
 * [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
 * [ENHANCEMENT] Add configuration to customize backoff for the gRPC clients.
@@ -121,6 +122,7 @@
 * [ENHANCEMENT] Added `TokenGenerator` interface with `RandomTokenGenerator` (generating random tokens), and `SpreadMinimizingTokenGenerator` (generating tokens with almost even distribution) implementation. By default `RandomTokenGenerator` is used. #321
 * [ENHANCEMENT] Lifecycler: Added `RingTokenGenerator` configuration that specifies the `TokenGenerator` implementation that is used for token generation. Default value is nil, meaning that `RandomTokenGenerator` is used. #323
 * [ENHANCEMENT] BasicLifecycler: Added `RingTokenGenerator` configuration that specifies the `TokenGenerator` implementation that is used for token generation. Default value is nil, meaning that `RandomTokenGenerator` is used. #323
+* [ENHANCEMENT] Ring: add support for hedging to `DoUntilQuorum` when request minimization is enabled. #330
 * [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
@@ -139,3 +141,4 @@
 * [BUGFIX] Ring: prevent iterating the whole ring when using `ExcludedZones`. #285
 * [BUGFIX] grpcclient: fix missing `.` in flag name for initial connection window size flag. #314
 * [BUGFIX] ring.Lifecycler: Handle when previous ring state is leaving and the number of tokens has changed. #79
+* [BUGFIX] memberlist metrics: fix registration of `memberlist_client_kv_store_count` metric and disable expiration of metrics exposed by memberlist library. #327
\ No newline at end of file
diff --git a/kv/memberlist/memberlist_client.go b/kv/memberlist/memberlist_client.go
index bcfc6ac..30a2753 100644
--- a/kv/memberlist/memberlist_client.go
+++ b/kv/memberlist/memberlist_client.go
@@ -164,9 +164,7 @@ type KVConfig struct {
 
 	TCPTransport TCPTransportConfig `yaml:",inline"`
 
-	// Where to put custom metrics. Metrics are not registered, if this is nil.
-	MetricsRegisterer prometheus.Registerer `yaml:"-"`
-	MetricsNamespace  string                `yaml:"-"`
+	MetricsNamespace string `yaml:"-"`
 
 	// Codecs to register. Codecs need to be registered before joining other members.
 	Codecs []codec.Codec `yaml:"-"`
@@ -354,7 +352,6 @@ var (
 // trigger connecting to the existing memberlist cluster. If that fails and AbortIfJoinFails is true, error is returned
 // and service enters Failed state.
 func NewKV(cfg KVConfig, logger log.Logger, dnsProvider DNSProvider, registerer prometheus.Registerer) *KV {
-	cfg.TCPTransport.MetricsRegisterer = cfg.MetricsRegisterer
 	cfg.TCPTransport.MetricsNamespace = cfg.MetricsNamespace
 
 	mlkv := &KV{
@@ -386,7 +383,7 @@ func defaultMemberlistConfig() *memberlist.Config {
 }
 
 func (m *KV) buildMemberlistConfig() (*memberlist.Config, error) {
-	tr, err := NewTCPTransport(m.cfg.TCPTransport, m.logger)
+	tr, err := NewTCPTransport(m.cfg.TCPTransport, m.logger, m.registerer)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create transport: %v", err)
 	}
diff --git a/kv/memberlist/metrics.go b/kv/memberlist/metrics.go
index 9ab56a6..75a6b23 100644
--- a/kv/memberlist/metrics.go
+++ b/kv/memberlist/metrics.go
@@ -171,17 +171,25 @@ func (m *KV) createAndRegisterMetrics() {
 		Help:      "Number of dropped notifications in WatchPrefix function",
 	}, []string{"prefix"})
 
-	if m.cfg.MetricsRegisterer == nil {
+	if m.registerer == nil {
 		return
 	}
 
+	m.registerer.MustRegister(m)
+
 	// memberlist uses armonmetrics package for internal usage
-	// here we configure armonmetrics to use prometheus
-	sink, err := armonprometheus.NewPrometheusSink() // there is no option to pass registrerer, this uses default
+	// here we configure armonmetrics to use prometheus.
+	opts := armonprometheus.PrometheusOpts{
+		Expiration: 0, // Don't expire metrics.
+		Registerer: m.registerer,
+	}
+
+	sink, err := armonprometheus.NewPrometheusSinkFrom(opts)
 	if err == nil {
 		cfg := armonmetrics.DefaultConfig("")
 		cfg.EnableHostname = false         // no need to put hostname into metric
 		cfg.EnableHostnameLabel = false    // no need to put hostname into labels
+		cfg.EnableServiceLabel = false     // Don't put service name into a label. (We don't set service name anyway).
 		cfg.EnableRuntimeMetrics = false   // metrics about Go runtime already provided by prometheus
 		cfg.EnableTypePrefix = true        // to make better sense of internal memberlist metrics
 		cfg.TimerGranularity = time.Second // timers are in seconds in prometheus world
diff --git a/kv/memberlist/tcp_transport.go b/kv/memberlist/tcp_transport.go
index 78fe196..bd82c58 100644
--- a/kv/memberlist/tcp_transport.go
+++ b/kv/memberlist/tcp_transport.go
@@ -56,8 +56,7 @@ type TCPTransportConfig struct {
 	TransportDebug bool `yaml:"-" category:"advanced"`
 
 	// Where to put custom metrics. nil = don't register.
-	MetricsRegisterer prometheus.Registerer `yaml:"-"`
-	MetricsNamespace  string                `yaml:"-"`
+	MetricsNamespace string `yaml:"-"`
 
 	TLSEnabled bool               `yaml:"tls_enabled" category:"advanced"`
 	TLS        dstls.ClientConfig `yaml:",inline"`
@@ -113,7 +112,7 @@ type TCPTransport struct {
 
 // NewTCPTransport returns a new tcp-based transport with the given configuration. On
 // success all the network listeners will be created and listening.
-func NewTCPTransport(config TCPTransportConfig, logger log.Logger) (*TCPTransport, error) {
+func NewTCPTransport(config TCPTransportConfig, logger log.Logger, registerer prometheus.Registerer) (*TCPTransport, error) {
 	if len(config.BindAddrs) == 0 {
 		config.BindAddrs = []string{zeroZeroZeroZero}
 	}
@@ -135,7 +134,7 @@ func NewTCPTransport(config TCPTransportConfig, logger log.Logger) (*TCPTranspor
 		}
 	}
 
-	t.registerMetrics(config.MetricsRegisterer)
+	t.registerMetrics(registerer)
 
 	// Clean up listeners if there's an error.
 	defer func() {
diff --git a/ring/replication_set.go b/ring/replication_set.go
index fcaae9a..d0cba7f 100644
--- a/ring/replication_set.go
+++ b/ring/replication_set.go
@@ -2,6 +2,7 @@ package ring
 
 import (
 	"context"
+	"errors"
 	"sort"
 	"time"
 )
@@ -89,37 +90,69 @@ func (r ReplicationSet) Do(ctx context.Context, delay time.Duration, f func(cont
 	return results, nil
 }
 
+type DoUntilQuorumConfig struct {
+	// If true, enable request minimization.
+	// See docs for DoUntilQuorum for more information.
+	MinimizeRequests bool
+
+	// If non-zero and MinimizeRequests is true, enables hedging.
+	// See docs for DoUntilQuorum for more information.
+	HedgingDelay time.Duration
+}
+
+func (c DoUntilQuorumConfig) Validate() error {
+	if c.HedgingDelay < 0 {
+		return errors.New("invalid DoUntilQuorumConfig: HedgingDelay must be non-negative")
+	}
+
+	return nil
+}
+
 // DoUntilQuorum runs function f in parallel for all replicas in r.
 //
 // # Result selection
 //
-// If r.MaxUnavailableZones is greater than zero:
+// If r.MaxUnavailableZones is greater than zero, DoUntilQuorum operates in zone-aware mode:
 //   - DoUntilQuorum returns an error if calls to f for instances in more than r.MaxUnavailableZones zones return errors
 //   - Otherwise, DoUntilQuorum returns all results from all replicas in the first zones for which f succeeds
 //     for every instance in that zone (eg. if there are 3 zones and r.MaxUnavailableZones is 1, DoUntilQuorum will
 //     return the results from all instances in 2 zones, even if all calls to f succeed).
 //
-// Otherwise:
+// Otherwise, DoUntilQuorum operates in non-zone-aware mode:
 //   - DoUntilQuorum returns an error if more than r.MaxErrors calls to f return errors
 //   - Otherwise, DoUntilQuorum returns all results from the first len(r.Instances) - r.MaxErrors instances
 //     (eg. if there are 6 replicas and r.MaxErrors is 2, DoUntilQuorum will return the results from the first 4
 //     successful calls to f, even if all 6 calls to f succeed).
 //
-// # Request minimisation
+// # Request minimization
+//
+// cfg.MinimizeRequests enables or disables request minimization.
+//
+// Regardless of the value of cfg.MinimizeRequests, if one of the termination conditions above is satisfied or ctx is
+// cancelled before f is called for an instance, f may not be called for that instance at all.
+//
+// ## When disabled
+//
+// If request minimization is disabled, DoUntilQuorum will call f for each instance in r. The value of cfg.HedgingDelay
+// is ignored.
 //
-// If minimizeRequests is false, DoUntilQuorum will call f for each instance in r.
+// ## When enabled
 //
-// If minimizeRequests is true, DoUntilQuorum will initially call f for the minimum number of instances needed to reach
-// the termination conditions above, and later call f for further instances if required. For example, if
+// If request minimization is enabled, DoUntilQuorum will initially call f for the minimum number of instances needed to
+// reach the termination conditions above, and later call f for further instances if required. For example, if
 // r.MaxUnavailableZones is 1 and there are three zones, DoUntilQuorum will initially only call f for instances in two
 // zones, and only call f for instances in the remaining zone if a request in the initial two zones fails.
 //
-// If minimizeRequests is true, DoUntilQuorum will randomly select available zones / instances such that calling
-// DoUntilQuorum multiple times with the same ReplicationSet should evenly distribute requests across all zones /
-// instances.
+// DoUntilQuorum will randomly select available zones / instances such that calling DoUntilQuorum multiple times with
+// the same ReplicationSet should evenly distribute requests across all zones / instances.
 //
-// Regardless of the value of minimizeRequests, if one of the termination conditions above is satisfied or ctx is
-// cancelled before f is called for an instance, f may not be called for that instance at all.
+// If cfg.HedgingDelay is non-zero, DoUntilQuorum will call f for an additional zone's instances (if zone-aware) / an
+// additional instance (if not zone-aware) every cfg.HedgingDelay until one of the termination conditions above is
+// reached. For example, if r.MaxUnavailableZones is 2, cfg.HedgingDelay is 4 seconds and there are fives zones,
+// DoUntilQuorum will initially only call f for instances in three zones, and unless one of the termination conditions
+// is reached earlier, will then call f for instances in a fourth zone approximately 4 seconds later, and then call f
+// for instances in the final zone approximately 4 seconds after that (ie. roughly 8 seconds since the call to
+// DoUntilQuorum began).
 //
 // # Cleanup
 //
@@ -135,7 +168,7 @@ func (r ReplicationSet) Do(ctx context.Context, delay time.Duration, f func(cont
 // f will not be used.
 //
 // DoUntilQuorum cancels the context.Context passed to each invocation of f before DoUntilQuorum returns.
-func DoUntilQuorum[T any](ctx context.Context, r ReplicationSet, minimizeRequests bool, f func(context.Context, *InstanceDesc) (T, error), cleanupFunc func(T)) ([]T, error) {
+func DoUntilQuorum[T any](ctx context.Context, r ReplicationSet, cfg DoUntilQuorumConfig, f func(context.Context, *InstanceDesc) (T, error), cleanupFunc func(T)) ([]T, error) {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
@@ -143,7 +176,7 @@ func DoUntilQuorum[T any](ctx context.Context, r ReplicationSet, minimizeRequest
 		return f(ctx, desc)
 	}
 
-	return DoUntilQuorumWithoutSuccessfulContextCancellation(ctx, r, minimizeRequests, wrappedF, cleanupFunc)
+	return DoUntilQuorumWithoutSuccessfulContextCancellation(ctx, r, cfg, wrappedF, cleanupFunc)
 }
 
 // DoUntilQuorumWithoutSuccessfulContextCancellation behaves the same as DoUntilQuorum, except it does not cancel
@@ -158,7 +191,11 @@ func DoUntilQuorum[T any](ctx context.Context, r ReplicationSet, minimizeRequest
 //     DoUntilQuorumWithoutSuccessfulContextCancellation
 //
 // Failing to do this may result in a memory leak.
-func DoUntilQuorumWithoutSuccessfulContextCancellation[T any](ctx context.Context, r ReplicationSet, minimizeRequests bool, f func(context.Context, *InstanceDesc, context.CancelFunc) (T, error), cleanupFunc func(T)) ([]T, error) {
+func DoUntilQuorumWithoutSuccessfulContextCancellation[T any](ctx context.Context, r ReplicationSet, cfg DoUntilQuorumConfig, f func(context.Context, *InstanceDesc, context.CancelFunc) (T, error), cleanupFunc func(T)) ([]T, error) {
+	if err := cfg.Validate(); err != nil {
+		return nil, err
+	}
+
 	resultsChan := make(chan instanceResult[T], len(r.Instances))
 	resultsRemaining := len(r.Instances)
 
@@ -185,7 +222,7 @@ func DoUntilQuorumWithoutSuccessfulContextCancellation[T any](ctx context.Contex
 		contextTracker = newDefaultContextTracker(ctx, r.Instances)
 	}
 
-	if minimizeRequests {
+	if cfg.MinimizeRequests {
 		resultTracker.startMinimumRequests()
 	} else {
 		resultTracker.startAllRequests()
@@ -222,6 +259,14 @@ func DoUntilQuorumWithoutSuccessfulContextCancellation[T any](ctx context.Contex
 		}
 	}
 
+	var hedgingTrigger <-chan time.Time
+
+	if cfg.HedgingDelay > 0 {
+		ticker := time.NewTicker(cfg.HedgingDelay)
+		defer ticker.Stop()
+		hedgingTrigger = ticker.C
+	}
+
 	for !resultTracker.succeeded() {
 		select {
 		case <-ctx.Done():
@@ -229,6 +274,8 @@ func DoUntilQuorumWithoutSuccessfulContextCancellation[T any](ctx context.Contex
 			cleanupResultsAlreadyReceived()
 
 			return nil, ctx.Err()
+		case <-hedgingTrigger:
+			resultTracker.startAdditionalRequests()
 		case result := <-resultsChan:
 			resultsRemaining--
 			resultTracker.done(result.instance, result.err)
diff --git a/ring/replication_set_tracker.go b/ring/replication_set_tracker.go
index 2bec036..6dd1ccf 100644
--- a/ring/replication_set_tracker.go
+++ b/ring/replication_set_tracker.go
@@ -34,6 +34,14 @@ type replicationSetResultTracker interface {
 	// This method must only be called before calling done.
 	startMinimumRequests()
 
+	// Starts additional request(s) as defined by the quorum requirements of this tracker.
+	// For example, a zone-aware tracker would start requests for another zone, whereas a
+	// non-zone-aware tracker would start a request for another instance.
+	// This method must only be called after calling startMinimumRequests or startAllRequests.
+	// If requests for all instances have already been started, this method does nothing.
+	// This method must only be called before calling done.
+	startAdditionalRequests()
+
 	// Starts requests for all instances.
 	// Calling this method multiple times may lead to unpredictable behaviour.
 	// Calling both this method and releaseMinimumRequests may lead to unpredictable behaviour.
@@ -95,13 +103,7 @@ func (t *defaultResultTracker) done(_ *InstanceDesc, err error) {
 		}
 	} else {
 		t.numErrors++
-
-		if len(t.pendingInstances) > 0 {
-			// There are some outstanding requests we could make before we reach maxErrors. Release the next one.
-			i := t.pendingInstances[0]
-			t.instanceRelease[i] <- struct{}{}
-			t.pendingInstances = t.pendingInstances[1:]
-		}
+		t.startAdditionalRequests()
 	}
 }
 
@@ -154,6 +156,15 @@ func (t *defaultResultTracker) startMinimumRequests() {
 	}
 }
 
+func (t *defaultResultTracker) startAdditionalRequests() {
+	if len(t.pendingInstances) > 0 {
+		// There are some outstanding requests we could make before we reach maxErrors. Release the next one.
+		i := t.pendingInstances[0]
+		t.instanceRelease[i] <- struct{}{}
+		t.pendingInstances = t.pendingInstances[1:]
+	}
+}
+
 func (t *defaultResultTracker) startAllRequests() {
 	t.instanceRelease = make(map[*InstanceDesc]chan struct{}, len(t.instances))
 
@@ -251,11 +262,9 @@ func (t *zoneAwareResultTracker) done(instance *InstanceDesc, err error) {
 	} else {
 		t.failuresByZone[instance.Zone]++
 
-		if len(t.pendingZones) > 0 && t.failuresByZone[instance.Zone] == 1 {
-			// If there are more zones we could try before reaching maxUnavailableZones and this was the first
-			// failure for this zone, release another zone's requests and signal they should start.
-			t.releaseZone(t.pendingZones[0], true)
-			t.pendingZones = t.pendingZones[1:]
+		if t.failuresByZone[instance.Zone] == 1 {
+			// If this was the first failure for this zone, release another zone's requests and signal they should start.
+			t.startAdditionalRequests()
 		}
 	}
 }
@@ -318,6 +327,14 @@ func (t *zoneAwareResultTracker) startMinimumRequests() {
 	}
 }
 
+func (t *zoneAwareResultTracker) startAdditionalRequests() {
+	if len(t.pendingZones) > 0 {
+		// If there are more zones we could try before reaching maxUnavailableZones, release another zone's requests and signal they should start.
+		t.releaseZone(t.pendingZones[0], true)
+		t.pendingZones = t.pendingZones[1:]
+	}
+}
+
 func (t *zoneAwareResultTracker) startAllRequests() {
 	t.createReleaseChannels()
 


Test patch:
diff --git a/kv/memberlist/memberlist_client_test.go b/kv/memberlist/memberlist_client_test.go
index 084e5e5..e3e2450 100644
--- a/kv/memberlist/memberlist_client_test.go
+++ b/kv/memberlist/memberlist_client_test.go
@@ -18,6 +18,7 @@ import (
 
 	"github.com/go-kit/log"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -1478,6 +1479,29 @@ func TestDelegateMethodsDontCrashBeforeKVStarts(t *testing.T) {
 	assert.Equal(t, msg, val)
 }
 
+func TestMetricsRegistration(t *testing.T) {
+	c := dataCodec{}
+
+	cfg := KVConfig{}
+	cfg.Codecs = append(cfg.Codecs, c)
+
+	reg := prometheus.NewPedanticRegistry()
+	kv := NewKV(cfg, log.NewNopLogger(), &dnsProviderMock{}, reg)
+	err := kv.CAS(context.Background(), "test", c, func(in interface{}) (out interface{}, retry bool, err error) {
+		return &data{Members: map[string]member{
+			"member": {},
+		}}, true, nil
+	})
+	require.NoError(t, err)
+
+	assert.NoError(t, testutil.GatherAndCompare(reg, strings.NewReader(`
+			# HELP memberlist_client_kv_store_count Number of values in KV Store
+			# TYPE memberlist_client_kv_store_count gauge
+			memberlist_client_kv_store_count 1
+	`), "memberlist_client_kv_store_count"))
+
+}
+
 func decodeDataFromMarshalledKeyValuePair(t *testing.T, marshalledKVP []byte, key string, codec dataCodec) *data {
 	kvp := KeyValuePair{}
 	require.NoError(t, kvp.Unmarshal(marshalledKVP))
diff --git a/kv/memberlist/tcp_transport_test.go b/kv/memberlist/tcp_transport_test.go
index a9da508..9abad52 100644
--- a/kv/memberlist/tcp_transport_test.go
+++ b/kv/memberlist/tcp_transport_test.go
@@ -4,6 +4,7 @@ import (
 	"testing"
 
 	"github.com/go-kit/log"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -44,7 +45,7 @@ func TestTCPTransport_WriteTo_ShouldNotLogAsWarningExpectedFailures(t *testing.T
 				testData.setup(t, &cfg)
 			}
 
-			transport, err := NewTCPTransport(cfg, logger)
+			transport, err := NewTCPTransport(cfg, logger, nil)
 			require.NoError(t, err)
 
 			_, err = transport.WriteTo([]byte("test"), testData.remoteAddr)
@@ -83,7 +84,7 @@ func TestFinalAdvertiseAddr(t *testing.T) {
 			cfg.BindAddrs = testData.bindAddrs
 			cfg.BindPort = testData.bindPort
 
-			transport, err := NewTCPTransport(cfg, logger)
+			transport, err := NewTCPTransport(cfg, logger, prometheus.NewPedanticRegistry())
 			require.NoError(t, err)
 
 			ip, port, err := transport.FinalAdvertiseAddr(testData.advertiseAddr, testData.bindPort)
diff --git a/ring/replication_set_test.go b/ring/replication_set_test.go
index a9a6288..dcfda8d 100644
--- a/ring/replication_set_test.go
+++ b/ring/replication_set_test.go
@@ -411,7 +411,8 @@ func TestDoUntilQuorumWithoutSuccessfulContextCancellation(t *testing.T) {
 						return res, err
 					}
 
-					actualResults, actualError := DoUntilQuorumWithoutSuccessfulContextCancellation(ctx, testCase.replicationSet, minimizeRequests, wrappedF, cleanupTracker.cleanup)
+					cfg := DoUntilQuorumConfig{MinimizeRequests: minimizeRequests}
+					actualResults, actualError := DoUntilQuorumWithoutSuccessfulContextCancellation(ctx, testCase.replicationSet, cfg, wrappedF, cleanupTracker.cleanup)
 					require.ElementsMatch(t, testCase.expectedResults, actualResults)
 					require.Equal(t, testCase.expectedError, actualError)
 
@@ -472,7 +473,8 @@ func TestDoUntilQuorumWithoutSuccessfulContextCancellation_MultipleUnavailableZo
 				return desc.Addr, nil
 			}
 
-			actualResults, actualError := DoUntilQuorumWithoutSuccessfulContextCancellation(ctx, replicationSet, minimizeRequests, wrappedF, cleanupTracker.cleanup)
+			cfg := DoUntilQuorumConfig{MinimizeRequests: minimizeRequests}
+			actualResults, actualError := DoUntilQuorumWithoutSuccessfulContextCancellation(ctx, replicationSet, cfg, wrappedF, cleanupTracker.cleanup)
 			require.Empty(t, actualResults)
 			require.EqualError(t, actualError, "error from a replica-1 instance")
 
@@ -521,7 +523,8 @@ func TestDoUntilQuorumWithoutSuccessfulContextCancellation_PartialZoneFailure(t
 				return desc.Addr, nil
 			}
 
-			actualResults, err := DoUntilQuorumWithoutSuccessfulContextCancellation(ctx, replicationSet, minimizeRequests, f, cleanupTracker.cleanup)
+			cfg := DoUntilQuorumConfig{MinimizeRequests: minimizeRequests}
+			actualResults, err := DoUntilQuorumWithoutSuccessfulContextCancellation(ctx, replicationSet, cfg, f, cleanupTracker.cleanup)
 			require.ElementsMatch(t, expectedResults, actualResults)
 			require.NoError(t, err)
 
@@ -565,8 +568,9 @@ func TestDoUntilQuorumWithoutSuccessfulContextCancellation_RunsCallsInParallel(t
 				return desc.Addr, nil
 			}
 
+			cfg := DoUntilQuorumConfig{MinimizeRequests: minimizeRequests}
 			cleanupFunc := func(_ string) {}
-			results, err := DoUntilQuorumWithoutSuccessfulContextCancellation(ctx, replicationSet, false, f, cleanupFunc)
+			results, err := DoUntilQuorumWithoutSuccessfulContextCancellation(ctx, replicationSet, cfg, f, cleanupFunc)
 			require.NoError(t, err)
 			require.ElementsMatch(t, []string{"replica-1", "replica-2"}, results)
 		})
@@ -607,7 +611,8 @@ func TestDoUntilQuorumWithoutSuccessfulContextCancellation_ReturnsMinimumResultS
 				return desc.Addr, nil
 			}
 
-			results, err := DoUntilQuorumWithoutSuccessfulContextCancellation(ctx, replicationSet, minimizeRequests, f, cleanupTracker.cleanup)
+			cfg := DoUntilQuorumConfig{MinimizeRequests: minimizeRequests}
+			results, err := DoUntilQuorumWithoutSuccessfulContextCancellation(ctx, replicationSet, cfg, f, cleanupTracker.cleanup)
 			require.NoError(t, err)
 			require.Len(t, results, 4)
 
@@ -698,7 +703,8 @@ func TestDoUntilQuorumWithoutSuccessfulContextCancellation_ReturnsMinimumResultS
 				return desc.Addr, nil
 			}
 
-			results, err := DoUntilQuorumWithoutSuccessfulContextCancellation(ctx, replicationSet, minimizeRequests, f, cleanupTracker.cleanup)
+			cfg := DoUntilQuorumConfig{MinimizeRequests: minimizeRequests}
+			results, err := DoUntilQuorumWithoutSuccessfulContextCancellation(ctx, replicationSet, cfg, f, cleanupTracker.cleanup)
 			require.NoError(t, err)
 			require.Len(t, results, 5, "should only have results from instances required to meet quorum requirement")
 
@@ -791,7 +797,8 @@ func TestDoUntilQuorumWithoutSuccessfulContextCancellation_DoesNotWaitForUnneces
 				return desc.Addr, nil
 			}
 
-			actualResults, err := DoUntilQuorumWithoutSuccessfulContextCancellation(ctx, testCase.replicationSet, false, f, cleanupTracker.cleanup)
+			cfg := DoUntilQuorumConfig{MinimizeRequests: false}
+			actualResults, err := DoUntilQuorumWithoutSuccessfulContextCancellation(ctx, testCase.replicationSet, cfg, f, cleanupTracker.cleanup)
 			require.ElementsMatch(t, testCase.expectedResults, actualResults)
 			require.NoError(t, err)
 
@@ -853,7 +860,8 @@ func TestDoUntilQuorumWithoutSuccessfulContextCancellation_ParentContextHandling
 		return desc.Addr, nil
 	}
 
-	results, err := DoUntilQuorumWithoutSuccessfulContextCancellation(parentCtx, replicationSet, false, f, cleanupTracker.cleanup)
+	cfg := DoUntilQuorumConfig{MinimizeRequests: false}
+	results, err := DoUntilQuorumWithoutSuccessfulContextCancellation(parentCtx, replicationSet, cfg, f, cleanupTracker.cleanup)
 	require.Empty(t, results)
 	require.Equal(t, context.Canceled, err)
 
@@ -914,7 +922,8 @@ func TestDoUntilQuorumWithoutSuccessfulContextCancellation_ParentContextHandling
 				cancel()
 			}()
 
-			results, err := DoUntilQuorumWithoutSuccessfulContextCancellation(ctx, replicationSet, true, f, cleanupTracker.cleanup)
+			cfg := DoUntilQuorumConfig{MinimizeRequests: true}
+			results, err := DoUntilQuorumWithoutSuccessfulContextCancellation(ctx, replicationSet, cfg, f, cleanupTracker.cleanup)
 			require.Empty(t, results)
 			require.Equal(t, context.Canceled, err)
 
@@ -971,7 +980,8 @@ func TestDoUntilQuorumWithoutSuccessfulContextCancellation_InstanceContextHandli
 				return desc.Addr, nil
 			}
 
-			instancesCalled, err := DoUntilQuorumWithoutSuccessfulContextCancellation(ctx, testCase.replicationSet, true, f, cleanupTracker.cleanup)
+			cfg := DoUntilQuorumConfig{MinimizeRequests: true}
+			instancesCalled, err := DoUntilQuorumWithoutSuccessfulContextCancellation(ctx, testCase.replicationSet, cfg, f, cleanupTracker.cleanup)
 			require.NoError(t, err)
 			require.Len(t, instancesCalled, testCase.expectedResultCount)
 
@@ -999,6 +1009,172 @@ func TestDoUntilQuorumWithoutSuccessfulContextCancellation_InstanceContextHandli
 	}
 }
 
+func TestDoUntilQuorumWithoutSuccessfulContextCancellation_InvalidHedgingConfig(t *testing.T) {
+	replicationSet := ReplicationSet{}
+	cfg := DoUntilQuorumConfig{
+		MinimizeRequests: true,
+		HedgingDelay:     -1 * time.Second,
+	}
+
+	f := func(ctx context.Context, desc *InstanceDesc, cancelFunc context.CancelFunc) (string, error) {
+		require.FailNow(t, "should never call f")
+		return "", nil
+	}
+
+	cleanup := func(_ string) {
+		require.FailNow(t, "should never call cleanup")
+	}
+
+	_, err := DoUntilQuorumWithoutSuccessfulContextCancellation(context.Background(), replicationSet, cfg, f, cleanup)
+	require.EqualError(t, err, "invalid DoUntilQuorumConfig: HedgingDelay must be non-negative")
+}
+
+type hedgingTestInvocation struct {
+	instance     *InstanceDesc
+	invokedAfter time.Duration
+}
+
+func instancesFor(results []hedgingTestInvocation) []*InstanceDesc {
+	instances := make([]*InstanceDesc, 0, len(results))
+
+	for _, result := range results {
+		instances = append(instances, result.instance)
+	}
+
+	return instances
+}
+
+func TestDoUntilQuorumWithoutSuccessfulContextCancellation_Hedging_ZoneAware(t *testing.T) {
+	replicationSet := ReplicationSet{
+		Instances: []InstanceDesc{
+			{Addr: "instance-1", Zone: "zone-a"},
+			{Addr: "instance-2", Zone: "zone-a"},
+			{Addr: "instance-3", Zone: "zone-b"},
+			{Addr: "instance-4", Zone: "zone-b"},
+			{Addr: "instance-5", Zone: "zone-c"},
+			{Addr: "instance-6", Zone: "zone-c"},
+			{Addr: "instance-7", Zone: "zone-d"},
+			{Addr: "instance-8", Zone: "zone-d"},
+		},
+		MaxUnavailableZones: 2,
+	}
+
+	defer goleak.VerifyNone(t)
+
+	ctx := context.Background()
+	cleanupTracker := newCleanupTracker(t, 4)
+
+	wg := sync.WaitGroup{}
+	wg.Add(len(replicationSet.Instances))
+	mtx := sync.RWMutex{}
+	invocations := make([]hedgingTestInvocation, 0, len(replicationSet.Instances))
+	startedAt := time.Now()
+
+	f := func(ctx context.Context, desc *InstanceDesc, cancel context.CancelFunc) (string, error) {
+		result := hedgingTestInvocation{
+			instance:     desc,
+			invokedAfter: time.Since(startedAt),
+		}
+
+		cleanupTracker.trackCall(ctx, desc, cancel)
+
+		mtx.Lock()
+		invocations = append(invocations, result)
+		mtx.Unlock()
+
+		wg.Done()
+		wg.Wait()
+
+		return desc.Addr, nil
+	}
+
+	cfg := DoUntilQuorumConfig{MinimizeRequests: true, HedgingDelay: time.Second}
+	successfulInstances, err := DoUntilQuorumWithoutSuccessfulContextCancellation(ctx, replicationSet, cfg, f, cleanupTracker.cleanup)
+	require.NoError(t, err)
+	require.Len(t, successfulInstances, 4)
+
+	cleanupTracker.collectCleanedUpInstances()
+	cleanupTracker.assertCorrectCleanup(successfulInstances, cleanupTracker.cleanedUpInstances)
+
+	require.ElementsMatch(t, append(successfulInstances, cleanupTracker.cleanedUpInstances...), []string{"instance-1", "instance-2", "instance-3", "instance-4", "instance-5", "instance-6", "instance-7", "instance-8"}, "all instances should be called")
+
+	require.Equal(t, 2, uniqueZoneCount(instancesFor(invocations[0:4])), "should initially release instances from two zones")
+	require.Equal(t, 1, uniqueZoneCount(instancesFor(invocations[4:6])), "should release one zone after first delay")
+	require.Equal(t, 1, uniqueZoneCount(instancesFor(invocations[6:8])), "should release one zone after first delay")
+
+	tolerance := float64((100 * time.Millisecond).Nanoseconds())
+
+	for _, result := range invocations[0:4] {
+		require.InDelta(t, 0, result.invokedAfter, tolerance, "expected first four requests to be released immediately")
+	}
+
+	for _, result := range invocations[4:6] {
+		require.InDelta(t, cfg.HedgingDelay, result.invokedAfter, tolerance, "expected next zone to be released after hedging delay")
+	}
+
+	for _, result := range invocations[6:8] {
+		require.InDelta(t, 2*cfg.HedgingDelay, result.invokedAfter, tolerance, "expected final zone to be released after another hedging delay")
+	}
+}
+
+func TestDoUntilQuorumWithoutSuccessfulContextCancellation_Hedging_NonZoneAware(t *testing.T) {
+	replicationSet := ReplicationSet{
+		Instances: []InstanceDesc{
+			{Addr: "instance-1"},
+			{Addr: "instance-2"},
+			{Addr: "instance-3"},
+			{Addr: "instance-4"},
+		},
+		MaxErrors: 2,
+	}
+
+	defer goleak.VerifyNone(t)
+
+	ctx := context.Background()
+	cleanupTracker := newCleanupTracker(t, 2)
+
+	wg := sync.WaitGroup{}
+	wg.Add(len(replicationSet.Instances))
+	mtx := sync.RWMutex{}
+	invocations := make([]hedgingTestInvocation, 0, len(replicationSet.Instances))
+	startedAt := time.Now()
+
+	f := func(ctx context.Context, desc *InstanceDesc, cancel context.CancelFunc) (string, error) {
+		result := hedgingTestInvocation{
+			instance:     desc,
+			invokedAfter: time.Since(startedAt),
+		}
+
+		cleanupTracker.trackCall(ctx, desc, cancel)
+
+		mtx.Lock()
+		invocations = append(invocations, result)
+		mtx.Unlock()
+
+		wg.Done()
+		wg.Wait()
+
+		return desc.Addr, nil
+	}
+
+	cfg := DoUntilQuorumConfig{MinimizeRequests: true, HedgingDelay: time.Second}
+	successfulInstances, err := DoUntilQuorumWithoutSuccessfulContextCancellation(ctx, replicationSet, cfg, f, cleanupTracker.cleanup)
+	require.NoError(t, err)
+	require.Len(t, successfulInstances, 2)
+
+	cleanupTracker.collectCleanedUpInstances()
+	cleanupTracker.assertCorrectCleanup(successfulInstances, cleanupTracker.cleanedUpInstances)
+
+	require.ElementsMatch(t, append(successfulInstances, cleanupTracker.cleanedUpInstances...), []string{"instance-1", "instance-2", "instance-3", "instance-4"}, "all instances should be called")
+
+	tolerance := float64((100 * time.Millisecond).Nanoseconds())
+
+	require.InDelta(t, 0, invocations[0].invokedAfter, tolerance, "expected first request to be released immediately")
+	require.InDelta(t, 0, invocations[1].invokedAfter, tolerance, "expected second request to be released immediately")
+	require.InDelta(t, cfg.HedgingDelay, invocations[2].invokedAfter, tolerance, "expected next request to be released after hedging delay")
+	require.InDelta(t, 2*cfg.HedgingDelay, invocations[3].invokedAfter, tolerance, "expected final request to be released after another hedging delay")
+}
+
 func TestDoUntilQuorum_InstanceContextHandling(t *testing.T) {
 	testCases := map[string]struct {
 		replicationSet      ReplicationSet
@@ -1046,7 +1222,8 @@ func TestDoUntilQuorum_InstanceContextHandling(t *testing.T) {
 				return desc.Addr, nil
 			}
 
-			instancesCalled, err := DoUntilQuorum(ctx, testCase.replicationSet, true, f, cleanupTracker.cleanup)
+			cfg := DoUntilQuorumConfig{MinimizeRequests: true}
+			instancesCalled, err := DoUntilQuorum(ctx, testCase.replicationSet, cfg, f, cleanupTracker.cleanup)
 			require.NoError(t, err)
 			require.Len(t, instancesCalled, testCase.expectedResultCount)
 
diff --git a/ring/replication_set_tracker_test.go b/ring/replication_set_tracker_test.go
index d0214e3..6ef3588 100644
--- a/ring/replication_set_tracker_test.go
+++ b/ring/replication_set_tracker_test.go
@@ -444,6 +444,52 @@ func TestDefaultResultTracker_StartMinimumRequests_MaxErrorsIsNumberOfInstances(
 	}
 }
 
+func TestDefaultResultTracker_StartAdditionalRequests(t *testing.T) {
+	instance1 := InstanceDesc{Addr: "127.0.0.1"}
+	instance2 := InstanceDesc{Addr: "127.0.0.2"}
+	instance3 := InstanceDesc{Addr: "127.0.0.3"}
+	instance4 := InstanceDesc{Addr: "127.0.0.4"}
+	instances := []InstanceDesc{instance1, instance2, instance3, instance4}
+
+	tracker := newDefaultResultTracker(instances, 2)
+	tracker.startMinimumRequests()
+
+	mtx := sync.RWMutex{}
+	countInstancesReleased := 0
+	countInstancesSignalledToStart := 0
+
+	for instanceIdx := range instances {
+		instanceIdx := instanceIdx
+		instance := &instances[instanceIdx]
+		go func() {
+			err := tracker.awaitStart(context.Background(), instance)
+
+			mtx.Lock()
+			defer mtx.Unlock()
+
+			countInstancesReleased++
+			if err == nil {
+				countInstancesSignalledToStart++
+			}
+		}()
+	}
+
+	waitForInstancesReleased := func(expected int, msg string) {
+		require.Eventually(t, func() bool {
+			mtx.RLock()
+			defer mtx.RUnlock()
+
+			return countInstancesReleased == expected && countInstancesSignalledToStart == expected
+		}, time.Second, 10*time.Millisecond, msg)
+	}
+
+	waitForInstancesReleased(2, "should initially release two requests")
+	tracker.startAdditionalRequests()
+	waitForInstancesReleased(3, "should release a third request after startAdditionalRequests()")
+	tracker.startAdditionalRequests()
+	waitForInstancesReleased(4, "should release remaining request after startAdditionalRequests()")
+}
+
 func TestDefaultContextTracker(t *testing.T) {
 	instance1 := InstanceDesc{Addr: "127.0.0.1"}
 	instance2 := InstanceDesc{Addr: "127.0.0.2"}
@@ -1120,6 +1166,56 @@ func TestZoneAwareResultTracker_StartMinimumRequests_MaxUnavailableZonesIsNumber
 	}
 }
 
+func TestZoneAwareResultTracker_StartAdditionalRequests(t *testing.T) {
+	instance1 := InstanceDesc{Addr: "127.0.0.1", Zone: "zone-a"}
+	instance2 := InstanceDesc{Addr: "127.0.0.2", Zone: "zone-a"}
+	instance3 := InstanceDesc{Addr: "127.0.0.3", Zone: "zone-b"}
+	instance4 := InstanceDesc{Addr: "127.0.0.4", Zone: "zone-b"}
+	instance5 := InstanceDesc{Addr: "127.0.0.5", Zone: "zone-c"}
+	instance6 := InstanceDesc{Addr: "127.0.0.6", Zone: "zone-c"}
+	instance7 := InstanceDesc{Addr: "127.0.0.7", Zone: "zone-d"}
+	instance8 := InstanceDesc{Addr: "127.0.0.8", Zone: "zone-d"}
+	instances := []InstanceDesc{instance1, instance2, instance3, instance4, instance5, instance6, instance7, instance8}
+
+	tracker := newZoneAwareResultTracker(instances, 2)
+	tracker.startMinimumRequests()
+
+	mtx := sync.RWMutex{}
+	instancesAwaitReleaseResults := make(map[*InstanceDesc]error, len(instances))
+	instancesReleased := make([]*InstanceDesc, 0, len(instances))
+	for instanceIdx := range instances {
+		instanceIdx := instanceIdx
+		instance := &instances[instanceIdx]
+		go func() {
+			released := tracker.awaitStart(context.Background(), instance)
+
+			mtx.Lock()
+			defer mtx.Unlock()
+			instancesAwaitReleaseResults[instance] = released
+			instancesReleased = append(instancesReleased, instance)
+		}()
+	}
+
+	waitForZonesReleased := func(expectedZones int, msg string) {
+		expectedInstances := 2 * expectedZones
+		require.Eventually(t, func() bool {
+			mtx.RLock()
+			defer mtx.RUnlock()
+
+			return len(instancesReleased) == expectedInstances
+		}, time.Second, 10*time.Millisecond, msg)
+
+		require.Equal(t, expectedInstances, nilErrorCount(instancesAwaitReleaseResults), "expected all released instances to be signalled to start")
+		require.Equal(t, expectedZones, uniqueZoneCount(instancesReleased), msg)
+	}
+
+	waitForZonesReleased(2, "should initially release two zones")
+	tracker.startAdditionalRequests()
+	waitForZonesReleased(3, "should release a third zone after startAdditionalRequests()")
+	tracker.startAdditionalRequests()
+	waitForZonesReleased(4, "should release remaining zone after startAdditionalRequests()")
+}
+
 func TestZoneAwareContextTracker(t *testing.T) {
 	instance1 := InstanceDesc{Addr: "127.0.0.1", Zone: "zone-a"}
 	instance2 := InstanceDesc{Addr: "127.0.0.2", Zone: "zone-a"}