From dd7c835c41660aa863d2d50a1558b4439bc173ea 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/4] 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 6bfc9ac8532..b03c012f8ca 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 * [FEATURE] Storage: Add support for Alibaba Cloud OSS as a backend for blocks, ruler, and alertmanager storage. Configured via `-.oss.*` flags with `backend: oss`. #7907 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 44b7dcc5dfa1c0847cc40a2c8adadaaf30749ad2 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/4] 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 b03c012f8ca..7060f640d9e 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 * [FEATURE] Storage: Add support for Alibaba Cloud OSS as a backend for blocks, ruler, and alertmanager storage. Configured via `-.oss.*` flags with `backend: oss`. #7907 @@ -20,6 +18,7 @@ * [BUGFIX] Store Gateway: Fix the Parquet bucket store ignoring query sharding, which made every shard return all series. #7892 * [BUGFIX] Querier: Fix the querier crashing instead of reporting the error when a blocks store set subservice fails. `blocksStoreReplicationSet` never initialised its `FailureWatcher`, so the failure callback dereferenced a nil pointer on the goroutine the service manager runs listeners on, and the resulting panic's stack trace buried the error logged for the actual cause. #7896 * [BUGFIX] Compactor: Delete block deletion markers left in the global `markers/` location of a tenant after the block they refer to was removed. They were skipped at every bucket index update and kept the tenant location in the bucket forever. #7890 +* [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 * [CHANGE] Ruler: Remove the deprecated `-ruler.evaluation-delay-duration` flag and its `ruler_evaluation_delay_duration` per-tenant limit. Use `-ruler.query-offset` / `ruler_query_offset`, which no longer takes the higher of the two values. Cortex decodes the runtime config strictly, so a leftover `ruler_evaluation_delay_duration` override makes the runtime config fail to load: Cortex **exits at startup** (`module failed`, `module=runtime-config`), and on an already-running process every reload fails, pinning the last good overrides and dropping `cortex_runtime_config_last_reload_successful` to 0. Run `grep -r ruler_evaluation_delay_duration` over your runtime configs before upgrading. #7792 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 b548abb6d995a019e50558552f388be783348ed6 Mon Sep 17 00:00:00 2001 From: Matt Van Horn Date: Thu, 29 Oct 2026 06:36:32 -0700 Subject: [PATCH 3/4] 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 Signed-off-by: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> --- 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, From e3c069ced6e2bf4702ea6fa1fd4e40b5f70c582e Mon Sep 17 00:00:00 2001 From: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Date: Thu, 5 Nov 2026 00:34:09 -0800 Subject: [PATCH 4/4] consul: use go.uber.org/atomic in the CAS tests faillint blocks sync/atomic in favor of go.uber.org/atomic. Signed-off-by: Matt Van Horn Signed-off-by: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> --- pkg/ring/kv/consul/client_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/ring/kv/consul/client_test.go b/pkg/ring/kv/consul/client_test.go index 60472110f79..a9d577365e4 100644 --- a/pkg/ring/kv/consul/client_test.go +++ b/pkg/ring/kv/consul/client_test.go @@ -11,7 +11,6 @@ import ( "path/filepath" "strconv" "sync" - "sync/atomic" "testing" "time" @@ -19,6 +18,7 @@ import ( "github.com/prometheus/client_golang/prometheus" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "go.uber.org/atomic" "github.com/cortexproject/cortex/integration/ca" "github.com/cortexproject/cortex/pkg/ring/kv/codec"