diff --git a/CHANGELOG.md b/CHANGELOG.md index 9daa87d2f2c..365337d06a7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,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 7e86bd8aef2..7a98f3131d7 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 (200ms+400ms+800ms+1.6s+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 @@ -181,9 +192,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, casBackoffConfig) + 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 +219,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 +253,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 +263,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 +272,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..60472110f79 100644 --- a/pkg/ring/kv/consul/client_test.go +++ b/pkg/ring/kv/consul/client_test.go @@ -4,9 +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" @@ -17,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" ) @@ -263,3 +269,157 @@ 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), 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) +}