diff --git a/CHANGELOG.md b/CHANGELOG.md index 75220a91e1..dd32c8108f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,6 +23,7 @@ * [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] Ring: Fix an instance becoming permanently unable to acquire tokens. #7901 +* [BUGFIX] HATracker: Do not report a sample rejected by the HA tracker from a non-elected replica as a failed KV store CAS operation. It was marking the `CAS` span as failed and increasing `memberlist_client_cas_failure_total`, although the sample was being deduplicated as expected. #7914 ## 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/ha/ha_tracker.go b/pkg/ha/ha_tracker.go index 649b433f8b..7761122b21 100644 --- a/pkg/ha/ha_tracker.go +++ b/pkg/ha/ha_tracker.go @@ -614,7 +614,7 @@ func (c *HATracker) CheckReplica(ctx context.Context, userID, replicaGroup, repl err := c.checkKVStore(ctx, key, replica, userID, now) c.kvCASCalls.WithLabelValues(userID, replicaGroup).Inc() if err != nil { - // The callback within checkKVStore will return a ReplicasNotMatchError if the sample is being deduped, + // checkKVStore will return a ReplicasNotMatchError if the sample is being deduped, // otherwise there may have been an actual error CAS'ing that we should log. if !errors.Is(err, ReplicasNotMatchError{}) { level.Error(c.logger).Log("msg", "rejecting sample", "err", err) @@ -624,7 +624,10 @@ func (c *HATracker) CheckReplica(ctx context.Context, userID, replicaGroup, repl } func (c *HATracker) checkKVStore(ctx context.Context, key, replica, userID string, now time.Time) error { - return c.client.CAS(ctx, key, func(in any) (out any, retry bool, err error) { + // A replica mismatch is a dedup decision, not a CAS failure: stop the CAS without an + // error, so it isn't tracked as failed, and report the decision to the caller afterwards. + var notMatchErr error + err := c.client.CAS(ctx, key, func(in any) (out any, retry bool, err error) { if desc, ok := in.(*ReplicaDesc); ok && desc.DeletedAt == 0 { // We don't need to CAS and update the timestamp in the KV store if the timestamp we've received // this sample at is less than updateTimeout amount of time since the timestamp in the KV store. @@ -635,7 +638,8 @@ func (c *HATracker) checkKVStore(ctx context.Context, key, replica, userID strin // We shouldn't failover to accepting a new replica if the timestamp we've received this sample at // is less than failover timeout amount of time since the timestamp in the KV store. if desc.Replica != replica && now.Sub(timestamp.Time(desc.ReceivedAt)) < c.limits.HATrackerFailoverTimeout(userID) { - return nil, false, ReplicasNotMatchError{replica: replica, elected: desc.Replica} + notMatchErr = ReplicasNotMatchError{replica: replica, elected: desc.Replica} + return nil, false, nil } } @@ -648,6 +652,10 @@ func (c *HATracker) checkKVStore(ctx context.Context, key, replica, userID strin DeletedAt: 0, }, true, nil }) + if err != nil { + return err + } + return notMatchErr } func (c *HATracker) Cfg() HATrackerConfig { diff --git a/pkg/ha/ha_tracker_test.go b/pkg/ha/ha_tracker_test.go index ad239df369..31a6bbb6df 100644 --- a/pkg/ha/ha_tracker_test.go +++ b/pkg/ha/ha_tracker_test.go @@ -199,6 +199,61 @@ func TestHATracker_CleanupDeletesArePropagatedWithMemberlist(t *testing.T) { require.NotEmpty(t, broadcasts, "Cleanup Delete should generate a broadcast for tombstone propagation") } +// TestCheckReplica_ReplicaMismatchIsNotCASFailure checks that a non-elected replica rejected +// during the CAS is reported to the caller, but isn't tracked as a failed CAS operation. +func TestCheckReplica_ReplicaMismatchIsNotCASFailure(t *testing.T) { + ctx := t.Context() + logger := log.NewNopLogger() + reg := prometheus.NewPedanticRegistry() + + var kvCfg memberlist.KVConfig + flagext.DefaultValues(&kvCfg) + replicaDescCodec := GetReplicaDescCodec() + kvCfg.Codecs = []codec.Codec{replicaDescCodec} + + mkv := memberlist.NewKV(kvCfg, logger, &dnsProviderMock{}, reg) + require.NoError(t, services.StartAndAwaitRunning(ctx, mkv)) + defer services.StopAndAwaitTerminated(ctx, mkv) //nolint:errcheck + + client, err := memberlist.NewClient(mkv, replicaDescCodec) + require.NoError(t, err) + + trackerCfg := HATrackerConfig{ + EnableHATracker: false, // to inject our client + UpdateTimeout: time.Second, + KVStore: kv.Config{Store: "memberlist"}, + } + failoverTimeout := 10 * time.Second + + tracker, err := NewHATracker(trackerCfg, trackerLimits{maxReplicaGroups: 100, failoverTimeout: failoverTimeout}, HATrackerStatusConfig{}, reg, "test", logger) + require.NoError(t, err) + tracker.cfg.EnableHATracker = true + tracker.client = client + + userID := "user1" + cluster := "cluster1" + + // The tracker isn't started, so its local cache stays empty and every check goes through the CAS. + now := time.Now() + require.NoError(t, tracker.CheckReplica(ctx, userID, cluster, "replica1", now)) + + err = tracker.CheckReplica(ctx, userID, cluster, "replica2", now.Add(time.Second)) + require.ErrorIs(t, err, ReplicasNotMatchError{}) + require.EqualError(t, err, ReplicasNotMatchError{replica: "replica2", elected: "replica1"}.Error()) + + require.NoError(t, testutil.GatherAndCompare(reg, strings.NewReader(` + # HELP memberlist_client_cas_failure_total Failed CAS operations + # TYPE memberlist_client_cas_failure_total counter + memberlist_client_cas_failure_total 0 + `), "memberlist_client_cas_failure_total")) + + // The rejected replica is still elected once the failover timeout has passed. + require.NoError(t, tracker.CheckReplica(ctx, userID, cluster, "replica2", now.Add(failoverTimeout+time.Second))) + val, err := client.Get(ctx, userID+"/"+cluster) + require.NoError(t, err) + require.Equal(t, "replica2", val.(*ReplicaDesc).GetReplica()) +} + // TestWatchPrefixNilPanicWithMemberlist reproduces the panic at ha_tracker.go:437: // With memberlist, WatchPrefix delivers a nil value when a key is deleted // (memberlist KV.get() returns nil for deleted/tombstone keys).