kvserver: add kv.closed_timestamp.policy_change

Previously, it was difficult to measure how often policies changed for ranges,
which is important because such changes can trigger additional range updates
sent in side transport.

This commit adds a metric to track the number of policy changes on replicas.

Part of: #143890
Release note: none

Author: wenyihu6

docs/generated/metrics/metrics.html

@@ -193,6 +193,7 @@
 <tr><td>STORAGE</td><td>kv.allocator.load_based_replica_rebalancing.missing_stats_for_existing_store</td><td>The number times the allocator was missing the qps stats for the existing store</td><td>Attempts</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>STORAGE</td><td>kv.allocator.load_based_replica_rebalancing.should_transfer</td><td>The number times the allocator determined that the replica should be rebalanced to another store for better load distribution</td><td>Attempts</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>STORAGE</td><td>kv.closed_timestamp.max_behind_nanos</td><td>Largest latency between realtime and replica max closed timestamp</td><td>Nanoseconds</td><td>GAUGE</td><td>NANOSECONDS</td><td>AVG</td><td>NONE</td></tr>
+<tr><td>STORAGE</td><td>kv.closed_timestamp.policy_change</td><td>Number of times closed timestamp policy change occurred on ranges</td><td>Events</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>STORAGE</td><td>kv.concurrency.avg_lock_hold_duration_nanos</td><td>Average lock hold duration across locks currently held in lock tables. Does not include replicated locks (intents) that are not held in memory</td><td>Nanoseconds</td><td>GAUGE</td><td>NANOSECONDS</td><td>AVG</td><td>NONE</td></tr>
 <tr><td>STORAGE</td><td>kv.concurrency.avg_lock_wait_duration_nanos</td><td>Average lock wait duration across requests currently waiting in lock wait-queues</td><td>Nanoseconds</td><td>GAUGE</td><td>NANOSECONDS</td><td>AVG</td><td>NONE</td></tr>
 <tr><td>STORAGE</td><td>kv.concurrency.latch_conflict_wait_durations</td><td>Durations in nanoseconds spent on latch acquisition waiting for conflicts with other latches</td><td>Nanoseconds</td><td>HISTOGRAM</td><td>NANOSECONDS</td><td>AVG</td><td>NONE</td></tr>

pkg/kv/kvserver/metrics.go

@@ -2432,6 +2432,14 @@ throttled they do count towards 'delay.total' and 'delay.enginebackpressure'.
 		Unit:        metric.Unit_NANOSECONDS,
 	}
 
+	// Closed timestamp policy change metrics.
+	metaClosedTimestampPolicyChange = metric.Metadata{
+		Name:        "kv.closed_timestamp.policy_change",
+		Help:        "Number of times closed timestamp policy change occurred on ranges",
+		Measurement: "Events",
+		Unit:        metric.Unit_COUNT,
+	}
+
 	// Replica circuit breaker.
 	metaReplicaCircuitBreakerCurTripped = metric.Metadata{
 		Name: "kv.replica_circuit_breaker.num_tripped_replicas",
@@ -3033,6 +3041,9 @@ type StoreMetrics struct {
 	// Closed timestamp metrics.
 	ClosedTimestampMaxBehindNanos *metric.Gauge
 
+	// Closed timestamp policy change on ranges metrics.
+	ClosedTimestampPolicyChange *metric.Counter
+
 	// Replica circuit breaker.
 	ReplicaCircuitBreakerCurTripped *metric.Gauge
 	ReplicaCircuitBreakerCumTripped *metric.Counter
@@ -3849,6 +3860,8 @@ func newStoreMetrics(histogramWindow time.Duration) *StoreMetrics {
 		// Estimated MVCC stats in split.
 		SplitsWithEstimatedStats:     metric.NewCounter(metaSplitEstimatedStats),
 		SplitEstimatedTotalBytesDiff: metric.NewCounter(metaSplitEstimatedTotalBytesDiff),
+
+		ClosedTimestampPolicyChange: metric.NewCounter(metaClosedTimestampPolicyChange),
 	}
 	sm.categoryIterMetrics.init(storeRegistry)
 

pkg/kv/kvserver/replica.go

@@ -1405,7 +1405,7 @@ func (r *Replica) closedTimestampPolicyRLocked() ctpb.RangeClosedTimestampPolicy
 // RefreshPolicy updates the replica's cached closed timestamp policy based on
 // span configurations and provided node round-trip latencies.
 func (r *Replica) RefreshPolicy(latencies map[roachpb.NodeID]time.Duration) {
-	policy := func() ctpb.RangeClosedTimestampPolicy {
+	computeNewPolicy := func(oldPolicy ctpb.RangeClosedTimestampPolicy) ctpb.RangeClosedTimestampPolicy {
 		desc, conf := r.DescAndSpanConfig()
 		// The node liveness range ignores zone configs and always uses a
 		// LAG_BY_CLUSTER_SETTING closed timestamp policy. If it was to begin
@@ -1438,12 +1438,17 @@ func (r *Replica) RefreshPolicy(latencies map[roachpb.NodeID]time.Duration) {
 			maxLatency = max(maxLatency, peerLatency)
 		}
 		return closedts.FindBucketBasedOnNetworkRTTWithDampening(
-			ctpb.RangeClosedTimestampPolicy(r.cachedClosedTimestampPolicy.Load()),
+			oldPolicy,
 			maxLatency,
 			closedts.PolicySwitchWhenLatencyExceedsBucketFraction.Get(&r.store.GetStoreConfig().Settings.SV),
 		)
 	}
-	r.cachedClosedTimestampPolicy.Store(int32(policy()))
+	oldPolicy := ctpb.RangeClosedTimestampPolicy(r.cachedClosedTimestampPolicy.Load())
+	newPolicy := computeNewPolicy(oldPolicy)
+	if newPolicy != oldPolicy {
+		r.store.metrics.ClosedTimestampPolicyChange.Inc(1)
+		r.cachedClosedTimestampPolicy.Store(int32(newPolicy))
+	}
 }
 
 // NodeID returns the ID of the node this replica belongs to.