From 560f198ce9227738893f11cfcbf21873610ced77 Mon Sep 17 00:00:00 2001 From: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Date: Sun, 25 Oct 2026 00:47:08 -0700 Subject: [PATCH 1/3] fix: back off Consul CAS retries during KV outages CAS now waits with the existing backoff configuration before retrying failed reads or writes, preserving the backoff across successful reads when writes continue to fail. Conflicts and callback retries do not trigger error backoff. Retry waits honor context cancellation, with additional cancellation checks before reads, writes, and the final error return. A regression test checks that two reads receiving HTTP 429 are separated by at least one second and never invoke the callback. Fixes #4697 Signed-off-by: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> --- CHANGELOG.md | 2 ++ pkg/ring/kv/consul/client.go | 27 ++++++++++++++++++++++++++- pkg/ring/kv/consul/client_test.go | 25 +++++++++++++++++++++++++ 3 files changed, 53 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 969e6c8f3b5..2c262b6daf9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## master / unreleased +* [BUGFIX] Consul: Back off CAS retries after failed reads or writes to reduce request amplification during outages, and honor context cancellation while waiting to retry. + * [FEATURE] Ruler: Add experimental support for federated rule groups. A rule group listing tenants in its `source_tenants` field is evaluated against those tenants while the resulting series and alerts are written to the tenant owning the rule group. Enabled with `-ruler.enable-federated-rules` (requires `-tenant-federation.enabled`), and restricted to selected tenants with `-ruler.allowed-federated-tenants` and `-ruler.disallowed-federated-tenants`. #7828 * [FEATURE] Distributor: Add experimental OTLP metrics ingestion over gRPC (`opentelemetry.proto.collector.metrics.v1.MetricsService/Export`) on the gRPC server port. Enable it with `-distributor.otlp.grpc-enabled`. The tenant is read from the `X-Scope-OrgID` gRPC metadata, and the request size is limited by `-server.grpc-max-recv-msg-size-bytes`. #7873 * [ENHANCEMENT] Ruler: Add rule information to query parameters sent to the Query Frontend when `-ruler.frontend-address` is configured, so the Query Frontend query stats logs (`-frontend.enabled-ruler-query-stats`) can identify the evaluated rule. In this mode, the Ruler query stats log (`-ruler.query-stats-enabled`) omits the fetched statistics fields, which are always zero. #6539 diff --git a/pkg/ring/kv/consul/client.go b/pkg/ring/kv/consul/client.go index 7e86bd8aef2..3cd282c9547 100644 --- a/pkg/ring/kv/consul/client.go +++ b/pkg/ring/kv/consul/client.go @@ -181,9 +181,26 @@ func (c *Client) cas(ctx context.Context, key string, f func(in any) (out any, r } index := uint64(0) + // Keep error backoff across attempts: a successful read doesn't imply + // that writes have recovered. Conflicts and callback retries don't wait. + errorBackoff := backoff.New(ctx, backoffConfig) + retryAfterError := false for i := 0; i < retries; i++ { + if retryAfterError { + errorBackoff.Wait() + retryAfterError = false + } if i > 0 && sleepBeforeRetry > 0 { - time.Sleep(sleepBeforeRetry) + timer := time.NewTimer(sleepBeforeRetry) + select { + case <-ctx.Done(): + timer.Stop() + return ctx.Err() + case <-timer.C: + } + } + if err := ctx.Err(); err != nil { + return err } // Get with default options - don't want stale data to compare with @@ -191,6 +208,7 @@ func (c *Client) cas(ctx context.Context, key string, f func(in any) (out any, r kvp, _, err := c.kv.Get(key, options.WithContext(ctx)) if err != nil { level.Error(c.logger).Log("msg", "error getting key", "key", key, "err", err) + retryAfterError = true continue } var intermediate any @@ -224,6 +242,9 @@ func (c *Client) cas(ctx context.Context, key string, f func(in any) (out any, r level.Error(c.logger).Log("msg", "error serialising value", "key", key, "err", err) continue } + if err := ctx.Err(); err != nil { + return err + } ok, _, err := c.kv.CAS(&consul.KVPair{ Key: key, Value: bytes, @@ -231,6 +252,7 @@ func (c *Client) cas(ctx context.Context, key string, f func(in any) (out any, r }, writeOptions.WithContext(ctx)) if err != nil { level.Error(c.logger).Log("msg", "error CASing", "key", key, "err", err) + retryAfterError = true continue } if !ok { @@ -239,6 +261,9 @@ func (c *Client) cas(ctx context.Context, key string, f func(in any) (out any, r } return nil } + if err := ctx.Err(); err != nil { + return err + } return fmt.Errorf("failed to CAS %s", key) } diff --git a/pkg/ring/kv/consul/client_test.go b/pkg/ring/kv/consul/client_test.go index 35c75b58ed8..1a4775adf14 100644 --- a/pkg/ring/kv/consul/client_test.go +++ b/pkg/ring/kv/consul/client_test.go @@ -5,6 +5,8 @@ import ( "crypto/x509" "crypto/x509/pkix" "fmt" + "net/http" + "net/http/httptest" "path/filepath" "strconv" "testing" @@ -263,3 +265,26 @@ type testLogger struct { func (l testLogger) Log(keyvals ...any) error { return nil } + +func TestCASBackoff(t *testing.T) { + requests := make(chan time.Time, 2) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests <- time.Now() + http.Error(w, "unavailable", http.StatusTooManyRequests) + })) + defer server.Close() + + c, err := NewClient(Config{Host: server.URL, MaxCasRetries: 2}, codec.String{}, testLogger{}, prometheus.NewPedanticRegistry()) + require.NoError(t, err) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + err = c.CAS(ctx, "test", func(any) (any, bool, error) { + t.Fatal("callback called after failed read") + return nil, false, nil + }) + require.EqualError(t, err, "failed to CAS test") + require.Len(t, requests, 2) + first, second := <-requests, <-requests + require.GreaterOrEqual(t, second.Sub(first), time.Second) +} From 5996434adcbaf7f30a023d5afc44771224903b1a Mon Sep 17 00:00:00 2001 From: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Date: Wed, 28 Oct 2026 11:21:29 -0700 Subject: [PATCH 2/3] consul: use a short dedicated backoff for CAS retries The 1s to 1m backoff is meant for the endless watch loops. Inside CAS, with the default 10 attempts, it could add minutes of waiting when Consul fails fast, and several CAS callers (lifecycler shutdown, heartbeats, autoJoin, initRing) pass context.Background(), so the context does not bound it. CAS now uses its own backoff (100ms min, 2s max). With the default 10 attempts there are at most 9 waits, totalling less than 13s, well below the default 1m ring heartbeat timeout. The watch loops keep the existing backoff. Add tests for failed writes after successful reads (delay keeps growing), context cancellation during the wait, and CAS conflicts being retried without waiting. Move the CHANGELOG entry below the existing BUGFIX entries and reference the PR. Signed-off-by: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> --- CHANGELOG.md | 3 +- pkg/ring/kv/consul/client.go | 13 ++- pkg/ring/kv/consul/client_test.go | 137 +++++++++++++++++++++++++++++- 3 files changed, 149 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2c262b6daf9..b567480ec74 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,8 +2,6 @@ ## master / unreleased -* [BUGFIX] Consul: Back off CAS retries after failed reads or writes to reduce request amplification during outages, and honor context cancellation while waiting to retry. - * [FEATURE] Ruler: Add experimental support for federated rule groups. A rule group listing tenants in its `source_tenants` field is evaluated against those tenants while the resulting series and alerts are written to the tenant owning the rule group. Enabled with `-ruler.enable-federated-rules` (requires `-tenant-federation.enabled`), and restricted to selected tenants with `-ruler.allowed-federated-tenants` and `-ruler.disallowed-federated-tenants`. #7828 * [FEATURE] Distributor: Add experimental OTLP metrics ingestion over gRPC (`opentelemetry.proto.collector.metrics.v1.MetricsService/Export`) on the gRPC server port. Enable it with `-distributor.otlp.grpc-enabled`. The tenant is read from the `X-Scope-OrgID` gRPC metadata, and the request size is limited by `-server.grpc-max-recv-msg-size-bytes`. #7873 * [ENHANCEMENT] Ruler: Add rule information to query parameters sent to the Query Frontend when `-ruler.frontend-address` is configured, so the Query Frontend query stats logs (`-frontend.enabled-ruler-query-stats`) can identify the evaluated rule. In this mode, the Ruler query stats log (`-ruler.query-stats-enabled`) omits the fetched statistics fields, which are always zero. #6539 @@ -15,6 +13,7 @@ * [BUGFIX] Ingester: Staleness markers written by the Ruler are no longer reported as discarded samples for the tenant. #7887 * [BUGFIX] Ruler: Stop retrying staleness markers that can never be accepted. A rule moved between rule groups left the original group re-appending staleness markers for its series on every evaluation until the ruler restarted. #7888 * [BUGFIX] Store Gateway: Fix the Parquet bucket store ignoring query sharding, which made every shard return all series. #7892 +* [BUGFIX] Consul: Back off CAS retries after failed reads or writes to reduce request amplification during outages, and honor context cancellation while waiting to retry. #7893 ## 1.22.0 in progress diff --git a/pkg/ring/kv/consul/client.go b/pkg/ring/kv/consul/client.go index 3cd282c9547..90f2a2d1150 100644 --- a/pkg/ring/kv/consul/client.go +++ b/pkg/ring/kv/consul/client.go @@ -30,10 +30,21 @@ const ( var ( writeOptions = &consul.WriteOptions{} + // backoffConfig is used by the long-running watch loops. backoffConfig = backoff.Config{ MinBackoff: 1 * time.Second, MaxBackoff: 1 * time.Minute, } + + // casBackoffConfig is used between CAS attempts after a failed read or + // write. It is kept short because CAS callers (e.g. ring heartbeats and + // shutdown paths) may not have a deadline: with the default 10 attempts + // the total wait is less than 13s (100ms+200ms+400ms+800ms+5*2s upper + // bounds), well below the default 1m ring heartbeat timeout. + casBackoffConfig = backoff.Config{ + MinBackoff: 100 * time.Millisecond, + MaxBackoff: 2 * time.Second, + } ) // Config to create a ConsulClient @@ -183,7 +194,7 @@ func (c *Client) cas(ctx context.Context, key string, f func(in any) (out any, r index := uint64(0) // Keep error backoff across attempts: a successful read doesn't imply // that writes have recovered. Conflicts and callback retries don't wait. - errorBackoff := backoff.New(ctx, backoffConfig) + errorBackoff := backoff.New(ctx, casBackoffConfig) retryAfterError := false for i := 0; i < retries; i++ { if retryAfterError { diff --git a/pkg/ring/kv/consul/client_test.go b/pkg/ring/kv/consul/client_test.go index 1a4775adf14..60472110f79 100644 --- a/pkg/ring/kv/consul/client_test.go +++ b/pkg/ring/kv/consul/client_test.go @@ -4,11 +4,14 @@ import ( "context" "crypto/x509" "crypto/x509/pkix" + "encoding/json" "fmt" "net/http" "net/http/httptest" "path/filepath" "strconv" + "sync" + "sync/atomic" "testing" "time" @@ -19,6 +22,7 @@ import ( "github.com/cortexproject/cortex/integration/ca" "github.com/cortexproject/cortex/pkg/ring/kv/codec" + "github.com/cortexproject/cortex/pkg/util/backoff" "github.com/cortexproject/cortex/pkg/util/tls" ) @@ -286,5 +290,136 @@ func TestCASBackoff(t *testing.T) { require.EqualError(t, err, "failed to CAS test") require.Len(t, requests, 2) first, second := <-requests, <-requests - require.GreaterOrEqual(t, second.Sub(first), time.Second) + require.GreaterOrEqual(t, second.Sub(first), casBackoffConfig.MinBackoff) +} + +// setCASBackoff overrides the CAS backoff for the duration of a test. +func setCASBackoff(t *testing.T, cfg backoff.Config) { + t.Helper() + orig := casBackoffConfig + casBackoffConfig = cfg + t.Cleanup(func() { casBackoffConfig = orig }) +} + +// casTestServer returns a fake Consul server whose reads always succeed and +// whose writes are answered by onWrite. It records the time of every write. +func casTestServer(t *testing.T, onWrite func(w http.ResponseWriter, n int)) (*httptest.Server, func() []time.Time) { + t.Helper() + var ( + mtx sync.Mutex + writes []time.Time + ) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodGet: + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode([]consul.KVPair{{Key: "test", Value: []byte("value"), ModifyIndex: 1}}) + case http.MethodPut: + mtx.Lock() + writes = append(writes, time.Now()) + n := len(writes) + mtx.Unlock() + onWrite(w, n) + default: + http.Error(w, "unexpected method", http.StatusMethodNotAllowed) + } + })) + t.Cleanup(server.Close) + return server, func() []time.Time { + mtx.Lock() + defer mtx.Unlock() + return append([]time.Time(nil), writes...) + } +} + +func TestCASBackoffAfterFailedWrites(t *testing.T) { + server, writes := casTestServer(t, func(w http.ResponseWriter, _ int) { + http.Error(w, "unavailable", http.StatusTooManyRequests) + }) + + c, err := NewClient(Config{Host: server.URL, MaxCasRetries: 4}, codec.String{}, testLogger{}, prometheus.NewPedanticRegistry()) + require.NoError(t, err) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + calls := 0 + err = c.CAS(ctx, "test", func(in any) (any, bool, error) { + calls++ + require.Equal(t, "value", in) + return "updated", false, nil + }) + require.EqualError(t, err, "failed to CAS test") + // Every read succeeds, so every attempt reaches the write. + require.Equal(t, 4, calls) + + ts := writes() + require.Len(t, ts, 4) + // The backoff is not reset by the successful reads in between, so the + // minimum delay between writes keeps doubling. + minDelay := casBackoffConfig.MinBackoff + for i := 1; i < len(ts); i++ { + require.GreaterOrEqual(t, ts[i].Sub(ts[i-1]), minDelay, "delay before write %d", i+1) + minDelay *= 2 + } +} + +func TestCASBackoffContextCanceledDuringWait(t *testing.T) { + // Make the wait long enough that returning early can only be explained + // by the context cancellation. + setCASBackoff(t, backoff.Config{MinBackoff: time.Minute, MaxBackoff: time.Minute}) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + var requests atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + http.Error(w, "unavailable", http.StatusTooManyRequests) + })) + defer server.Close() + + c, err := NewClient(Config{Host: server.URL}, codec.String{}, testLogger{}, prometheus.NewPedanticRegistry()) + require.NoError(t, err) + + // Cancel shortly after the first read has failed, while CAS is waiting. + go func() { + for requests.Load() == 0 { + time.Sleep(10 * time.Millisecond) + } + time.Sleep(200 * time.Millisecond) + cancel() + }() + + start := time.Now() + err = c.CAS(ctx, "test", func(any) (any, bool, error) { + t.Fatal("callback called after failed read") + return nil, false, nil + }) + require.ErrorIs(t, err, context.Canceled) + require.Less(t, time.Since(start), 10*time.Second) + require.Equal(t, int32(1), requests.Load()) +} + +func TestCASConflictRetriedWithoutBackoff(t *testing.T) { + // Any error backoff would make the test exceed its deadline. + setCASBackoff(t, backoff.Config{MinBackoff: time.Minute, MaxBackoff: time.Minute}) + + const conflicts = 3 + server, writes := casTestServer(t, func(w http.ResponseWriter, n int) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(n > conflicts) + }) + + c, err := NewClient(Config{Host: server.URL}, codec.String{}, testLogger{}, prometheus.NewPedanticRegistry()) + require.NoError(t, err) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + calls := 0 + err = c.CAS(ctx, "test", func(any) (any, bool, error) { + calls++ + return "updated", false, nil + }) + require.NoError(t, err) + require.Equal(t, conflicts+1, calls) + require.Len(t, writes(), conflicts+1) } From 0a53279577287ffd682fc4eebf45e407647b7cba Mon Sep 17 00:00:00 2001 From: Matt Van Horn Date: Thu, 29 Oct 2026 06:36:32 -0700 Subject: [PATCH 3/3] consul: correct the CAS backoff upper bounds in the comment The jitter range starts at [100ms, 200ms) and doubles, so the upper bounds of the first four waits are 200ms, 400ms, 800ms and 1.6s, then 2s for the remaining five, which totals 13s. Signed-off-by: Matt Van Horn --- pkg/ring/kv/consul/client.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/ring/kv/consul/client.go b/pkg/ring/kv/consul/client.go index 90f2a2d1150..7a98f3131d7 100644 --- a/pkg/ring/kv/consul/client.go +++ b/pkg/ring/kv/consul/client.go @@ -39,7 +39,7 @@ var ( // casBackoffConfig is used between CAS attempts after a failed read or // write. It is kept short because CAS callers (e.g. ring heartbeats and // shutdown paths) may not have a deadline: with the default 10 attempts - // the total wait is less than 13s (100ms+200ms+400ms+800ms+5*2s upper + // the total wait is less than 13s (200ms+400ms+800ms+1.6s+5*2s upper // bounds), well below the default 1m ring heartbeat timeout. casBackoffConfig = backoff.Config{ MinBackoff: 100 * time.Millisecond,