kvserver: add subsume.locks_written metric

The new subsume.locks_written metric tracks how many locks are moved
from the in-memory lock table to the replicated lock table during a
SubsumeRequest.

While here, I updated some log lines and added some assertions to
existing tests to make sure we were seeing non-zero values for both the
new metric and lease.transfer.locks_written.

Epic: none
Release note: None

Author: stevendanna

docs/generated/metrics/metrics.yaml

@@ -17498,6 +17498,14 @@ layers:
       unit: COUNT
       aggregation: AVG
       derivative: NON_NEGATIVE_DERIVATIVE
+    - name: subsume.locks_written
+      exported_name: subsume_locks_written
+      description: Number of locks written to storage during subsume (range merge)
+      y_axis_label: Locks Written
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
     - name: sysbytes
       exported_name: sysbytes
       description: Number of bytes in system KV pairs

pkg/kv/kvserver/batcheval/cmd_lease.go

@@ -74,14 +74,16 @@ func evalNewLease(
 			// If this is a lease transfer, we write out all unreplicated leases to
 			// storage, so that any waiters will discover them on the new leaseholder.
 			acquisitions, approxSize := rec.GetConcurrencyManager().OnRangeLeaseTransferEval()
-			log.VEventf(ctx, 2, "upgrading durability of %d locks", len(acquisitions))
 			if approxSize > durabilityUpgradeLimit {
 				log.KvExec.Warningf(ctx,
 					"refusing to upgrade lock durability of %d locks since approximate lock size of %d byte exceeds %d bytes",
 					len(acquisitions),
 					approxSize,
 					durabilityUpgradeLimit)
 			} else {
+				if len(acquisitions) > 0 {
+					log.KvExec.Infof(ctx, "upgrading durability of %d locks during lease transfer", len(acquisitions))
+				}
 				for _, acq := range acquisitions {
 					if err := storage.MVCCAcquireLock(ctx, readWriter,
 						&acq.Txn, acq.IgnoredSeqNums, acq.Strength, acq.Key, ms, 0, 0, true /* allowSequenceNumberRegression */); err != nil {

pkg/kv/kvserver/batcheval/cmd_subsume.go

@@ -184,6 +184,7 @@ func Subsume(
 	// will be cleared later when OnRangeMerge is called the replica in
 	// (*Store).MergeRange.
 	stats := cArgs.EvalCtx.GetMVCCStats()
+	var locksWritten int
 	if args.PreserveUnreplicatedLocks {
 		durabilityUpgradeLimit := concurrency.GetMaxLockFlushSize(&cArgs.EvalCtx.ClusterSettings().SV)
 		acquisitions, approxSize := cArgs.EvalCtx.GetConcurrencyManager().OnRangeSubsumeEval()
@@ -194,7 +195,9 @@ func Subsume(
 				approxSize,
 				durabilityUpgradeLimit)
 		} else {
-			log.VEventf(ctx, 2, "upgrading durability of %d locks", len(acquisitions))
+			if len(acquisitions) > 0 {
+				log.KvExec.Infof(ctx, "upgrading durability of %d lock during subsume", len(acquisitions))
+			}
 			statsDelta := enginepb.MVCCStats{}
 			for _, acq := range acquisitions {
 				if err := storage.MVCCAcquireLock(ctx, readWriter,
@@ -206,6 +209,7 @@ func Subsume(
 			// response and to the stats update we expect as part of this proposal.
 			stats.Add(statsDelta)
 			cArgs.Stats.Add(statsDelta)
+			locksWritten = len(acquisitions)
 		}
 	}
 
@@ -246,5 +250,10 @@ func Subsume(
 	// waitForApplication when sending a kvpb.SubsumeRequest.
 	pd.Replicated.DoTimelyApplicationToAllReplicas = true
 	pd.Local.RepopulateSubsumeResponseLAI = args.PreserveUnreplicatedLocks
+	if locksWritten > 0 {
+		pd.Local.Metrics = &result.Metrics{
+			SubsumeLocksWritten: locksWritten,
+		}
+	}
 	return pd, nil
 }

pkg/kv/kvserver/batcheval/result/metrics.go

@@ -19,6 +19,7 @@ type Metrics struct {
 	AddSSTableAsWrites           int // AddSSTable requests with IngestAsWrites set
 	SplitsWithEstimatedStats     int // Splits that computed stats estimates
 	SplitEstimatedTotalBytesDiff int // Difference between pre- and post-split total bytes.
+	SubsumeLocksWritten          int // number of locks written during a subsume
 }
 
 // Add absorbs the supplied Metrics into the receiver.
@@ -34,4 +35,5 @@ func (mt *Metrics) Add(o Metrics) {
 	mt.AddSSTableAsWrites += o.AddSSTableAsWrites
 	mt.SplitsWithEstimatedStats += o.SplitsWithEstimatedStats
 	mt.SplitEstimatedTotalBytesDiff += o.SplitEstimatedTotalBytesDiff
+	mt.SubsumeLocksWritten += o.SubsumeLocksWritten
 }

pkg/kv/kvserver/client_metrics_test.go

@@ -159,7 +159,7 @@ func TestStoreResolveMetrics(t *testing.T) {
 	// them everywhere.
 	{
 		act := fmt.Sprintf("%+v", result.Metrics{})
-		exp := "{LeaseRequestSuccess:0 LeaseRequestError:0 LeaseTransferSuccess:0 LeaseTransferError:0 LeaseTransferLocksWritten:0 ResolveCommit:0 ResolveAbort:0 ResolvePoison:0 AddSSTableAsWrites:0 SplitsWithEstimatedStats:0 SplitEstimatedTotalBytesDiff:0}"
+		exp := "{LeaseRequestSuccess:0 LeaseRequestError:0 LeaseTransferSuccess:0 LeaseTransferError:0 LeaseTransferLocksWritten:0 ResolveCommit:0 ResolveAbort:0 ResolvePoison:0 AddSSTableAsWrites:0 SplitsWithEstimatedStats:0 SplitEstimatedTotalBytesDiff:0 SubsumeLocksWritten:0}"
 		if act != exp {
 			t.Errorf("need to update this test due to added fields: %v", act)
 		}

pkg/kv/kvserver/client_replica_test.go

@@ -6171,6 +6171,11 @@ func TestLeaseTransferReplicatesLocks(t *testing.T) {
 	t.Log("cancelling txn2")
 	txn2Cancel()
 	require.NoError(t, g.Wait())
+
+	// Check metrics
+	locksWritten, err := tc.GetFirstStoreFromServer(t, 0).Metrics().GetStoreMetric("leases.transfers.locks_written")
+	require.NoError(t, err)
+	require.GreaterOrEqual(t, locksWritten, int64(1))
 }
 
 func TestLeaseTransferDropsLocksIfLargerThanCommandSize(t *testing.T) {
@@ -6312,15 +6317,16 @@ func TestMergeReplicatesLocks(t *testing.T) {
 	)
 	concurrency.UnreplicatedLockReliabilityMerge.Override(ctx, &st.SV, true)
 
-	for _, b := range []bool{true, false} {
+	for _, rhsLock := range []bool{true, false} {
 		name := "lhs-lock"
 		lockKeySuffix := lhsKey
-		if b {
+		if rhsLock {
 			name = "rhs-lock"
 			lockKeySuffix = rhsKey
 		}
 		t.Run(name, func(t *testing.T) {
-			tc := testcluster.StartTestCluster(t, 3, base.TestClusterArgs{
+			nodeCount := 3
+			tc := testcluster.StartTestCluster(t, nodeCount, base.TestClusterArgs{
 				ServerArgs: base.TestServerArgs{
 					Settings: st,
 				},
@@ -6417,6 +6423,17 @@ func TestMergeReplicatesLocks(t *testing.T) {
 			for _, err := range failures {
 				t.Errorf("consistency failure: %s", err.Error())
 			}
+			if rhsLock {
+				// The range could have been on any node. We just care that this metric
+				// is written somewhere.
+				var locksWritten int64
+				for i := range nodeCount {
+					l, err := tc.GetFirstStoreFromServer(t, i).Metrics().GetStoreMetric("subsume.locks_written")
+					require.NoError(t, err)
+					locksWritten += l
+				}
+				require.GreaterOrEqual(t, locksWritten, int64(1))
+			}
 		})
 	}
 }

pkg/kv/kvserver/metrics.go

@@ -295,6 +295,13 @@ var (
 		Unit:        metric.Unit_COUNT,
 	}
 
+	metaSubsumeLocksWrittenCount = metric.Metadata{
+		Name:        "subsume.locks_written",
+		Help:        "Number of locks written to storage during subsume (range merge)",
+		Measurement: "Locks Written",
+		Unit:        metric.Unit_COUNT,
+	}
+
 	metaReqCPUNanos = metric.Metadata{
 		Name:        "replicas.cpunanospersecond",
 		Help:        "Nanoseconds of CPU time in Replica request processing including evaluation but not replication",
@@ -3363,6 +3370,8 @@ type StoreMetrics struct {
 	SplitsWithEstimatedStats     *metric.Counter
 	SplitEstimatedTotalBytesDiff *metric.Counter
 
+	SubsumeLocksWritten *metric.Counter
+
 	FlushUtilization *metric.GaugeFloat64
 	FsyncLatency     *metric.ManualWindowHistogram
 
@@ -3718,6 +3727,9 @@ func newStoreMetrics(histogramWindow time.Duration) *StoreMetrics {
 		ResolveAbortCount:  metric.NewCounter(metaResolveAbort),
 		ResolvePoisonCount: metric.NewCounter(metaResolvePoison),
 
+		// Subsume metrics
+		SubsumeLocksWritten: metric.NewCounter(metaSubsumeLocksWrittenCount),
+
 		Capacity:  metric.NewGauge(metaCapacity),
 		Available: metric.NewGauge(metaAvailable),
 		Used:      metric.NewGauge(metaUsed),
@@ -4516,6 +4528,9 @@ func (sm *StoreMetrics) handleMetricsResult(ctx context.Context, metric result.M
 	sm.SplitEstimatedTotalBytesDiff.Inc(int64(metric.SplitEstimatedTotalBytesDiff))
 	metric.SplitEstimatedTotalBytesDiff = 0
 
+	sm.SubsumeLocksWritten.Inc(int64(metric.SubsumeLocksWritten))
+	metric.SubsumeLocksWritten = 0
+
 	if metric != (result.Metrics{}) {
 		log.KvExec.Fatalf(ctx, "unhandled fields in metrics result: %+v", metric)
 	}