report non-elected replica rejection without marking CAS as failed

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

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 {

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).