kvserver: record follower write bytes during recordStatsOnCommit

Previously, we recorded write bytes for follower replicas in
runPostAddTriggersReplicaOnly, right before apply, which could be inaccurate
since stats should be recorded at commit time. This commit moves the recording
to recordStatsOnCommit by including the write bytes in the batch stats.

Epic: none
Release note: none

Author: wenyihu6

pkg/kv/kvserver/app_batch.go

@@ -33,6 +33,7 @@ type appBatchStats struct {
 	numEntriesProcessedBytes   int64
 	numEmptyEntries            int
 	numAddSST, numAddSSTCopies int
+	numWriteBytes              int64
 
 	// NB: update `merge` when adding a new field.
 }
@@ -42,6 +43,7 @@ func (s *appBatchStats) merge(ss appBatchStats) {
 	s.numEntriesProcessed += ss.numEntriesProcessed
 	s.numEntriesProcessedBytes += ss.numEntriesProcessedBytes
 	ss.numEmptyEntries += ss.numEmptyEntries
+	s.numWriteBytes += ss.numWriteBytes
 }
 
 // appBatch is the in-progress foundation for standalone log entry

pkg/kv/kvserver/replica_app_batch.go

@@ -256,7 +256,7 @@ func (b *replicaAppBatch) runPostAddTriggersReplicaOnly(
 	if !cmd.IsLocal() {
 		writeBytes, ingestedBytes := cmd.getStoreWriteByteSizes()
 		if writeBytes > 0 || ingestedBytes > 0 {
-			b.r.recordRequestWriteBytes(writeBytes, ingestedBytes)
+			b.ab.numWriteBytes += writeBytes + ingestedBytes
 		}
 		// TODO(irfansharif): This code block can be removed once below-raft
 		// admission control is the only form of IO admission control. It pre-dates
@@ -725,6 +725,7 @@ func (b *replicaAppBatch) recordStatsOnCommit() {
 	b.applyStats.appBatchStats.merge(b.ab.appBatchStats)
 	b.applyStats.numBatchesProcessed++
 	b.applyStats.followerStoreWriteBytes.Merge(b.followerStoreWriteBytes)
+	b.r.recordRequestWriteBytes(b.ab.numWriteBytes)
 
 	if n := b.ab.numAddSST; n > 0 {
 		b.r.store.metrics.AddSSTableApplications.Inc(int64(n))

pkg/kv/kvserver/replica_raft.go

@@ -823,6 +823,9 @@ func (s handleRaftReadyStats) SafeFormat(p redact.SafePrinter, _ rune) {
 			p.Printf(" (copies=%d)", c)
 		}
 	}
+	if b := s.apply.numWriteBytes; b > 0 {
+		p.Printf(", apply-write-bytes=%s", humanizeutil.IBytes(b))
+	}
 	p.SafeString("]")
 
 	if n := s.apply.assertionsRequested; n > 0 {

pkg/kv/kvserver/replica_send.go

@@ -220,7 +220,7 @@ func (r *Replica) SendWithWriteBytes(
 	// accounting.
 	r.recordBatchRequestLoad(ctx, ba)
 	if writeBytes != nil {
-		r.recordRequestWriteBytes(writeBytes.WriteBytes, writeBytes.IngestedBytes)
+		r.recordRequestWriteBytes(writeBytes.WriteBytes + writeBytes.IngestedBytes)
 	}
 	r.recordImpactOnRateLimiter(ctx, br, isReadOnly)
 	return br, writeBytes, pErr
@@ -1069,10 +1069,10 @@ func (r *Replica) getBatchRequestQPS(ctx context.Context, ba *kvpb.BatchRequest)
 
 // recordRequestWriteBytes records the write bytes from a replica batch
 // request.
-func (r *Replica) recordRequestWriteBytes(writeBytes int64, ingestedBytes int64) {
+func (r *Replica) recordRequestWriteBytes(writeBytes int64) {
 	// TODO(kvoli): Consider recording the ingested bytes (AddSST) separately
 	// to the write bytes.
-	r.loadStats.RecordWriteBytes(float64(writeBytes + ingestedBytes))
+	r.loadStats.RecordWriteBytes(float64(writeBytes))
 }
 
 // checkBatchRequest verifies BatchRequest validity requirements. In particular,