kvserver: account for replica follower write bytes

Previously, only bytes written at the leaseholder were accounted for on
a replica. Now, follower replicas will record the bytes written.

Informs: cockroachdb#91152

Release note: None

pkg/kv/kvserver/replica_app_batch.go

@@ -604,9 +604,6 @@ func (b *replicaAppBatch) ApplyToStateMachine(ctx context.Context) error {
 	deltaStats.Subtract(prevStats)
 	r.store.metrics.addMVCCStats(ctx, r.tenantMetricsRef, deltaStats)
 
-	// Record the number of keys written to the replica.
-	b.r.loadStats.RecordWriteKeys(float64(b.ab.numMutations))
-
 	now := timeutil.Now()
 	if needsSplitBySize && r.splitQueueThrottle.ShouldProcess(now) {
 		r.store.splitQueue.MaybeAddAsync(ctx, r, r.store.Clock().NowAsClockTimestamp())

pkg/kv/kvserver/replica_raft.go

@@ -993,10 +993,19 @@ func (r *Replica) handleRaftReadyRaftMuLocked(
 			return stats, err
 		}
 
-		if r.store.cfg.KVAdmissionController != nil &&
-			stats.apply.followerStoreWriteBytes.NumEntries > 0 {
-			r.store.cfg.KVAdmissionController.FollowerStoreWriteBytes(
-				r.store.StoreID(), stats.apply.followerStoreWriteBytes)
+		r.loadStats.RecordWriteKeys(float64(stats.apply.numMutations))
+		if stats.apply.followerStoreWriteBytes.NumEntries > 0 {
+			// NB: The write bytes for non-followers are recorded in replica
+			// send, the folower bytes are accounted for here on the follower
+			// replica's store.
+			r.loadStats.RecordWriteBytes(float64(
+				stats.apply.followerStoreWriteBytes.WriteBytes +
+					stats.apply.followerStoreWriteBytes.IngestedBytes))
+
+			if r.store.cfg.KVAdmissionController != nil {
+				r.store.cfg.KVAdmissionController.FollowerStoreWriteBytes(
+					r.store.StoreID(), stats.apply.followerStoreWriteBytes)
+			}
 		}
 
 		// etcd raft occasionally adds a nil entry (our own commands are never