kvserver: push unavailability computation inside the leaderlessWatcher

No reason to do this on the Replica.

Release note: None

Author: arulajmani

pkg/kv/kvserver/replica.go

@@ -235,6 +235,49 @@ func (lw *leaderlessWatcher) IsUnavailable() bool {
 	return lw.mu.unavailable
 }
 
+// refreshUnavailableState refreshes the unavailable state on the leaderless
+// watcher. Replicas are considered unavailable if they have been leaderless for
+// a long time, where long is defined by the ReplicaUnavailableThreshold.
+func (lw *leaderlessWatcher) refreshUnavailableState(
+	ctx context.Context, postTickLead raftpb.PeerID, nowPhysicalTime time.Time, st *cluster.Settings,
+) {
+	lw.mu.Lock()
+	defer lw.mu.Unlock()
+
+	threshold := ReplicaLeaderlessUnavailableThreshold.Get(&st.SV)
+	if threshold == time.Duration(0) {
+		// The leaderless watcher is disabled. It's important to reset the
+		// leaderless watcher when it's disabled to reset any replica that was
+		// marked as unavailable before the watcher was disabled.
+		lw.resetLocked()
+		return
+	}
+
+	if postTickLead != raft.None {
+		// If we know about the leader, reset the leaderless timer, and mark the
+		// replica as available.
+		lw.resetLocked()
+	} else if lw.mu.leaderlessTimestamp.IsZero() {
+		// If we don't know about the leader, and we haven't been leaderless before,
+		// mark the time we became leaderless.
+		lw.mu.leaderlessTimestamp = nowPhysicalTime
+	} else if !lw.mu.unavailable {
+		// At this point we know that we have been leaderless for some time, and we
+		// haven't marked the replica as unavailable yet. Make sure we didn't exceed
+		// the threshold. Otherwise, mark the replica as unavailable.
+		durationSinceLeaderless := nowPhysicalTime.Sub(lw.mu.leaderlessTimestamp)
+		if durationSinceLeaderless >= threshold {
+			err := errors.Errorf("have been leaderless for %.2fs, setting the "+
+				"leaderless watcher replica's state as unavailable",
+				durationSinceLeaderless.Seconds())
+			if log.ExpensiveLogEnabled(ctx, 1) {
+				log.VEventf(ctx, 1, "%s", err)
+			}
+			lw.mu.unavailable = true
+		}
+	}
+}
+
 func (lw *leaderlessWatcher) resetLocked() {
 	lw.mu.leaderlessTimestamp = time.Time{}
 	lw.mu.unavailable = false

pkg/kv/kvserver/replica_raft.go

@@ -1510,9 +1510,10 @@ func (r *Replica) tick(
 	r.mu.internalRaftGroup.Tick()
 	postTickStatus := r.mu.internalRaftGroup.BasicStatus()
 
-	// Check if the replica has been leaderless for too long, and potentially set
-	// the leaderless watcher replica state as unavailable.
-	r.maybeMarkReplicaUnavailableInLeaderlessWatcher(ctx, postTickStatus.Lead, nowPhysicalTime)
+	// Refresh the unavailability state on the leaderlessWatcher.
+	r.LeaderlessWatcher.refreshUnavailableState(
+		ctx, postTickStatus.Lead, nowPhysicalTime, r.store.cfg.Settings,
+	)
 
 	if preTickStatus.RaftState != postTickStatus.RaftState {
 		if postTickStatus.RaftState == raftpb.StatePreCandidate {
@@ -2114,49 +2115,6 @@ func (r *Replica) reportSnapshotStatus(ctx context.Context, to roachpb.ReplicaID
 	}
 }
 
-// maybeMarkReplicaUnavailableInLeaderlessWatcher marks the replica as
-// unavailable in the leaderless watcher if the replica has been leaderless
-// for a duration of time greater than or equal to the threshold.
-func (r *Replica) maybeMarkReplicaUnavailableInLeaderlessWatcher(
-	ctx context.Context, postTickLead raftpb.PeerID, storeClockTime time.Time,
-) {
-	r.LeaderlessWatcher.mu.Lock()
-	defer r.LeaderlessWatcher.mu.Unlock()
-
-	threshold := ReplicaLeaderlessUnavailableThreshold.Get(&r.store.cfg.Settings.SV)
-	if threshold == time.Duration(0) {
-		// The leaderless watcher is disabled. It's important to reset the
-		// leaderless watcher when it's disabled to reset any replica that was
-		// marked as unavailable before the watcher was disabled.
-		r.LeaderlessWatcher.resetLocked()
-		return
-	}
-
-	if postTickLead != raft.None {
-		// If we know about the leader, reset the leaderless timer, and mark the
-		// replica as available.
-		r.LeaderlessWatcher.resetLocked()
-	} else if r.LeaderlessWatcher.mu.leaderlessTimestamp.IsZero() {
-		// If we don't know about the leader, and we haven't been leaderless before,
-		// mark the time we became leaderless.
-		r.LeaderlessWatcher.mu.leaderlessTimestamp = storeClockTime
-	} else if !r.LeaderlessWatcher.mu.unavailable {
-		// At this point we know that we have been leaderless for sometime, and
-		// we haven't marked the replica as unavailable yet. Make sure we didn't
-		// exceed the threshold. Otherwise, mark the replica as unavailable.
-		durationSinceLeaderless := storeClockTime.Sub(r.LeaderlessWatcher.mu.leaderlessTimestamp)
-		if durationSinceLeaderless >= threshold {
-			err := errors.Errorf("have been leaderless for %.2fs, setting the "+
-				"leaderless watcher replica's state as unavailable",
-				durationSinceLeaderless.Seconds())
-			if log.ExpensiveLogEnabled(ctx, 1) {
-				log.VEventf(ctx, 1, "%s", err)
-			}
-			r.LeaderlessWatcher.mu.unavailable = true
-		}
-	}
-}
-
 type snapTruncationInfo struct {
 	index          kvpb.RaftIndex
 	recipientStore roachpb.StoreID

pkg/kv/kvserver/replica_raft_test.go

@@ -481,7 +481,7 @@ func TestMaybeMarkReplicaUnavailableInLeaderlessWatcher(t *testing.T) {
 		repl := tContext.repl
 		repl.LeaderlessWatcher.mu.unavailable = tc.initReplicaUnavailable
 		repl.LeaderlessWatcher.mu.leaderlessTimestamp = tc.initLeaderlessTimestamp
-		repl.maybeMarkReplicaUnavailableInLeaderlessWatcher(ctx, tc.leader, now)
+		repl.LeaderlessWatcher.refreshUnavailableState(ctx, tc.leader, now, cfg.Settings)
 		require.Equal(t, tc.expectedUnavailable, repl.LeaderlessWatcher.IsUnavailable())
 		require.Equal(t, tc.expectedLeaderlessTime, repl.LeaderlessWatcher.mu.leaderlessTimestamp)