.*: introduce send queue and range controller metrics

The following metrics are introduced:

```
-- send queue token metrics
kvflowcontrol.tokens.send.regular.deducted.prevent_send_queue     (counter)
kvflowcontrol.tokens.send.elastic.deducted.prevent_send_queue     (counter)
kvflowcontrol.tokens.send.elastic.deducted.force_flush_send_queue (counter)
-- range controller metrics
kvflowcontrol.range_controller.count                              (gauge)
-- send queue metrics
kvflowcontrol.send_queue.bytes                                    (gauge)
kvflowcontrol.send_queue.count                                    (gauge)
kvflowcontrol.send_queue.prevent.count                            (counter)
kvflowcontrol.send_queue.scheduled.deducted_bytes                 (gauge)
kvflowcontrol.send_queue.scheduled.force_flush                    (gauge)
```

The follower related metric names are updated from:

```
kvflowcontrol.tokens.eval.regular.disconnected
kvflowcontrol.tokens.eval.elastic.disconnected
kvflowcontrol.tokens.send.regular.disconnected
kvflowcontrol.tokens.send.elastic.disconnected
```

To:

```
kvflowcontrol.tokens.eval.regular.returned.disconnect
kvflowcontrol.tokens.eval.elastic.returned.disconnect
kvflowcontrol.tokens.send.regular.returned.disconnect
kvflowcontrol.tokens.send.elastic.returned.disconnect
```

In order to align with the format of having a general metric
`(deducted|returned)`, accounting for all token deductions or returns, and
a series of metrics which break down the deduction or return further,
using a suffix to denote the reason.

The `SendStreamStats` are also now updated periodically every 5s to
include send queue size (bytes|count). Implying that the stats used in
the metrics may be up to 5s stale, which is assumed to be reasonable.

Resolves: #131752
Release note (ops change):
Metrics representing the state of the replication flow control send
queue's are added, which when the following settings are set to:

```
kvadmission.flow_control.mode = "apply_to_all"
kvadmission.flow_control.enabled = true
```

Will be populated using the corresponding replication admission control
send queue state. Also note, the disconnected metric name is renamed,
see the commit message and the metric name diff below for further
information.

```diff
diff --git a/pkg/kv/kvserver/kvflowcontrol/rac2/testdata/simulation/metric_names b/pkg/kv/kvserver/kvflowcontrol/rac2/testdata/simulation/metric_names
index c0b4755f1eb..96b237bb57f 100644
--- a/pkg/kv/kvserver/kvflowcontrol/rac2/testdata/simulation/metric_names
+++ b/pkg/kv/kvserver/kvflowcontrol/rac2/testdata/simulation/metric_names
@@ -20,6 +20,12 @@ kvflowcontrol.eval_wait.regular.requests.admitted
 kvflowcontrol.eval_wait.regular.requests.bypassed
 kvflowcontrol.eval_wait.regular.requests.errored
 kvflowcontrol.eval_wait.regular.requests.waiting
+kvflowcontrol.range_controller.count
+kvflowcontrol.send_queue.bytes
+kvflowcontrol.send_queue.count
+kvflowcontrol.send_queue.prevent.count
+kvflowcontrol.send_queue.scheduled.deducted_bytes
+kvflowcontrol.send_queue.scheduled.force_flush
 kvflowcontrol.streams.eval.elastic.blocked_count
 kvflowcontrol.streams.eval.elastic.total_count
 kvflowcontrol.streams.eval.regular.blocked_count
@@ -30,21 +36,24 @@ kvflowcontrol.streams.send.regular.blocked_count
 kvflowcontrol.streams.send.regular.total_count
 kvflowcontrol.tokens.eval.elastic.available
 kvflowcontrol.tokens.eval.elastic.deducted
-kvflowcontrol.tokens.eval.elastic.disconnected
 kvflowcontrol.tokens.eval.elastic.returned
+kvflowcontrol.tokens.eval.elastic.returned.disconnect
 kvflowcontrol.tokens.eval.elastic.unaccounted
 kvflowcontrol.tokens.eval.regular.available
 kvflowcontrol.tokens.eval.regular.deducted
-kvflowcontrol.tokens.eval.regular.disconnected
 kvflowcontrol.tokens.eval.regular.returned
+kvflowcontrol.tokens.eval.regular.returned.disconnect
 kvflowcontrol.tokens.eval.regular.unaccounted
 kvflowcontrol.tokens.send.elastic.available
 kvflowcontrol.tokens.send.elastic.deducted
-kvflowcontrol.tokens.send.elastic.disconnected
+kvflowcontrol.tokens.send.elastic.deducted.force_flush_send_queue
+kvflowcontrol.tokens.send.elastic.deducted.prevent_send_queue
 kvflowcontrol.tokens.send.elastic.returned
+kvflowcontrol.tokens.send.elastic.returned.disconnect
 kvflowcontrol.tokens.send.elastic.unaccounted
 kvflowcontrol.tokens.send.regular.available
 kvflowcontrol.tokens.send.regular.deducted
-kvflowcontrol.tokens.send.regular.disconnected
+kvflowcontrol.tokens.send.regular.deducted.prevent_send_queue
 kvflowcontrol.tokens.send.regular.returned
+kvflowcontrol.tokens.send.regular.returned.disconnect
 kvflowcontrol.tokens.send.regular.unaccounted
```

docs/generated/metrics/metrics.html

@@ -320,6 +320,12 @@
 <tr><td>STORAGE</td><td>kvflowcontrol.eval_wait.regular.requests.bypassed</td><td>Number of waiting regular requests that bypassed the flow controller due the evaluating replica not being the leader</td><td>Requests</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>STORAGE</td><td>kvflowcontrol.eval_wait.regular.requests.errored</td><td>Number of regular requests that errored out while waiting for flow tokens</td><td>Requests</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>STORAGE</td><td>kvflowcontrol.eval_wait.regular.requests.waiting</td><td>Number of regular requests waiting for flow tokens</td><td>Requests</td><td>GAUGE</td><td>COUNT</td><td>AVG</td><td>NONE</td></tr>
+<tr><td>STORAGE</td><td>kvflowcontrol.range_controller.count</td><td>Gauge of range flow controllers currently open, this should align with the number of leaders</td><td>Count</td><td>GAUGE</td><td>COUNT</td><td>AVG</td><td>NONE</td></tr>
+<tr><td>STORAGE</td><td>kvflowcontrol.send_queue.bytes</td><td>Byte size of all raft entries queued for sending to followers, waiting on available elastic send tokens</td><td>Bytes</td><td>GAUGE</td><td>BYTES</td><td>AVG</td><td>NONE</td></tr>
+<tr><td>STORAGE</td><td>kvflowcontrol.send_queue.count</td><td>Count of all raft entries queued for sending to followers, waiting on available elastic send tokens</td><td>Bytes</td><td>GAUGE</td><td>COUNT</td><td>AVG</td><td>NONE</td></tr>
+<tr><td>STORAGE</td><td>kvflowcontrol.send_queue.prevent.count</td><td>Counter of replication streams that were prevented from forming a send queue</td><td>Preventions</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>STORAGE</td><td>kvflowcontrol.send_queue.scheduled.deducted_bytes</td><td>Gauge of elastic send token bytes already deducted by replication streams waiting on the scheduler</td><td>Bytes</td><td>GAUGE</td><td>BYTES</td><td>AVG</td><td>NONE</td></tr>
+<tr><td>STORAGE</td><td>kvflowcontrol.send_queue.scheduled.force_flush</td><td>Gauge of replication streams scheduled to force flush their send queue</td><td>Scheduled force flushes</td><td>GAUGE</td><td>COUNT</td><td>AVG</td><td>NONE</td></tr>
 <tr><td>STORAGE</td><td>kvflowcontrol.streams.eval.elastic.blocked_count</td><td>Number of eval replication streams with no flow tokens available for elastic requests</td><td>Count</td><td>GAUGE</td><td>COUNT</td><td>AVG</td><td>NONE</td></tr>
 <tr><td>STORAGE</td><td>kvflowcontrol.streams.eval.elastic.total_count</td><td>Total number of eval replication streams for elastic requests</td><td>Count</td><td>GAUGE</td><td>COUNT</td><td>AVG</td><td>NONE</td></tr>
 <tr><td>STORAGE</td><td>kvflowcontrol.streams.eval.regular.blocked_count</td><td>Number of eval replication streams with no flow tokens available for regular requests</td><td>Count</td><td>GAUGE</td><td>COUNT</td><td>AVG</td><td>NONE</td></tr>
@@ -330,23 +336,26 @@
 <tr><td>STORAGE</td><td>kvflowcontrol.streams.send.regular.total_count</td><td>Total number of send replication streams for regular requests</td><td>Count</td><td>GAUGE</td><td>COUNT</td><td>AVG</td><td>NONE</td></tr>
 <tr><td>STORAGE</td><td>kvflowcontrol.tokens.eval.elastic.available</td><td>Flow eval tokens available for elastic requests, across all replication streams</td><td>Bytes</td><td>GAUGE</td><td>BYTES</td><td>AVG</td><td>NONE</td></tr>
 <tr><td>STORAGE</td><td>kvflowcontrol.tokens.eval.elastic.deducted</td><td>Flow eval tokens deducted by elastic requests, across all replication streams</td><td>Bytes</td><td>COUNTER</td><td>BYTES</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
-<tr><td>STORAGE</td><td>kvflowcontrol.tokens.eval.elastic.disconnected</td><td>Flow eval tokens returned early by elastic due disconnects, across all replication stream, this is a subset of returned tokens</td><td>Bytes</td><td>COUNTER</td><td>BYTES</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>STORAGE</td><td>kvflowcontrol.tokens.eval.elastic.returned</td><td>Flow eval tokens returned by elastic requests, across all replication streams</td><td>Bytes</td><td>COUNTER</td><td>BYTES</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>STORAGE</td><td>kvflowcontrol.tokens.eval.elastic.returned.disconnect</td><td>Flow eval tokens returned early by elastic due disconnects, across all replication stream, this is a subset of returned tokens</td><td>Bytes</td><td>COUNTER</td><td>BYTES</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>STORAGE</td><td>kvflowcontrol.tokens.eval.elastic.unaccounted</td><td>Flow eval tokens returned by elastic requests that were unaccounted for, across all replication streams</td><td>Bytes</td><td>COUNTER</td><td>BYTES</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>STORAGE</td><td>kvflowcontrol.tokens.eval.regular.available</td><td>Flow eval tokens available for regular requests, across all replication streams</td><td>Bytes</td><td>GAUGE</td><td>BYTES</td><td>AVG</td><td>NONE</td></tr>
 <tr><td>STORAGE</td><td>kvflowcontrol.tokens.eval.regular.deducted</td><td>Flow eval tokens deducted by regular requests, across all replication streams</td><td>Bytes</td><td>COUNTER</td><td>BYTES</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
-<tr><td>STORAGE</td><td>kvflowcontrol.tokens.eval.regular.disconnected</td><td>Flow eval tokens returned early by regular due disconnects, across all replication stream, this is a subset of returned tokens</td><td>Bytes</td><td>COUNTER</td><td>BYTES</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>STORAGE</td><td>kvflowcontrol.tokens.eval.regular.returned</td><td>Flow eval tokens returned by regular requests, across all replication streams</td><td>Bytes</td><td>COUNTER</td><td>BYTES</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>STORAGE</td><td>kvflowcontrol.tokens.eval.regular.returned.disconnect</td><td>Flow eval tokens returned early by regular due disconnects, across all replication stream, this is a subset of returned tokens</td><td>Bytes</td><td>COUNTER</td><td>BYTES</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>STORAGE</td><td>kvflowcontrol.tokens.eval.regular.unaccounted</td><td>Flow eval tokens returned by regular requests that were unaccounted for, across all replication streams</td><td>Bytes</td><td>COUNTER</td><td>BYTES</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>STORAGE</td><td>kvflowcontrol.tokens.send.elastic.available</td><td>Flow send tokens available for elastic requests, across all replication streams</td><td>Bytes</td><td>GAUGE</td><td>BYTES</td><td>AVG</td><td>NONE</td></tr>
 <tr><td>STORAGE</td><td>kvflowcontrol.tokens.send.elastic.deducted</td><td>Flow send tokens deducted by elastic requests, across all replication streams</td><td>Bytes</td><td>COUNTER</td><td>BYTES</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
-<tr><td>STORAGE</td><td>kvflowcontrol.tokens.send.elastic.disconnected</td><td>Flow send tokens returned early by elastic due disconnects, across all replication stream, this is a subset of returned tokens</td><td>Bytes</td><td>COUNTER</td><td>BYTES</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>STORAGE</td><td>kvflowcontrol.tokens.send.elastic.deducted.force_flush_send_queue</td><td>Flow send tokens deducted by elastic requests, across all replication streams due to force flushing the stream&#39;s send queue</td><td>Bytes</td><td>COUNTER</td><td>BYTES</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>STORAGE</td><td>kvflowcontrol.tokens.send.elastic.deducted.prevent_send_queue</td><td>Flow send tokens deducted by elastic requests, across all replication streams to prevent forming a send queue</td><td>Bytes</td><td>COUNTER</td><td>BYTES</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>STORAGE</td><td>kvflowcontrol.tokens.send.elastic.returned</td><td>Flow send tokens returned by elastic requests, across all replication streams</td><td>Bytes</td><td>COUNTER</td><td>BYTES</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>STORAGE</td><td>kvflowcontrol.tokens.send.elastic.returned.disconnect</td><td>Flow send tokens returned early by elastic due disconnects, across all replication stream, this is a subset of returned tokens</td><td>Bytes</td><td>COUNTER</td><td>BYTES</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>STORAGE</td><td>kvflowcontrol.tokens.send.elastic.unaccounted</td><td>Flow send tokens returned by elastic requests that were unaccounted for, across all replication streams</td><td>Bytes</td><td>COUNTER</td><td>BYTES</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>STORAGE</td><td>kvflowcontrol.tokens.send.regular.available</td><td>Flow send tokens available for regular requests, across all replication streams</td><td>Bytes</td><td>GAUGE</td><td>BYTES</td><td>AVG</td><td>NONE</td></tr>
 <tr><td>STORAGE</td><td>kvflowcontrol.tokens.send.regular.deducted</td><td>Flow send tokens deducted by regular requests, across all replication streams</td><td>Bytes</td><td>COUNTER</td><td>BYTES</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
-<tr><td>STORAGE</td><td>kvflowcontrol.tokens.send.regular.disconnected</td><td>Flow send tokens returned early by regular due disconnects, across all replication stream, this is a subset of returned tokens</td><td>Bytes</td><td>COUNTER</td><td>BYTES</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>STORAGE</td><td>kvflowcontrol.tokens.send.regular.deducted.prevent_send_queue</td><td>Flow send tokens deducted by regular requests, across all replication streams to prevent forming a send queue</td><td>Bytes</td><td>COUNTER</td><td>BYTES</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>STORAGE</td><td>kvflowcontrol.tokens.send.regular.returned</td><td>Flow send tokens returned by regular requests, across all replication streams</td><td>Bytes</td><td>COUNTER</td><td>BYTES</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>STORAGE</td><td>kvflowcontrol.tokens.send.regular.returned.disconnect</td><td>Flow send tokens returned early by regular due disconnects, across all replication stream, this is a subset of returned tokens</td><td>Bytes</td><td>COUNTER</td><td>BYTES</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>STORAGE</td><td>kvflowcontrol.tokens.send.regular.unaccounted</td><td>Flow send tokens returned by regular requests that were unaccounted for, across all replication streams</td><td>Bytes</td><td>COUNTER</td><td>BYTES</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>STORAGE</td><td>leases.epoch</td><td>Number of replica leaseholders using epoch-based leases</td><td>Replicas</td><td>GAUGE</td><td>COUNT</td><td>AVG</td><td>NONE</td></tr>
 <tr><td>STORAGE</td><td>leases.error</td><td>Number of failed lease requests</td><td>Lease Requests</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>

pkg/kv/kvserver/allocator/allocatorimpl/allocator.go

@@ -11,6 +11,7 @@ import (
 	"fmt"
 	"math"
 	"math/rand"
+	"sync"
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
@@ -2015,7 +2016,7 @@ func (a *Allocator) ValidLeaseTargets(
 		StoreID() roachpb.StoreID
 		RaftStatus() *raft.Status
 		GetFirstIndex() kvpb.RaftIndex
-		SendStreamStats() rac2.RangeSendStreamStats
+		SendStreamStats(*rac2.RangeSendStreamStats)
 	},
 	opts allocator.TransferLeaseOptions,
 ) []roachpb.ReplicaDescriptor {
@@ -2080,7 +2081,7 @@ func (a *Allocator) ValidLeaseTargets(
 		candidates = append(validSnapshotCandidates, excludeReplicasInNeedOfSnapshots(
 			ctx, status, leaseRepl.GetFirstIndex(), candidates)...)
 		candidates = excludeReplicasInNeedOfCatchup(
-			ctx, leaseRepl.SendStreamStats(), candidates)
+			ctx, leaseRepl.SendStreamStats, candidates)
 	}
 
 	// Determine which store(s) is preferred based on user-specified preferences.
@@ -2189,7 +2190,7 @@ func (a *Allocator) LeaseholderShouldMoveDueToPreferences(
 		StoreID() roachpb.StoreID
 		RaftStatus() *raft.Status
 		GetFirstIndex() kvpb.RaftIndex
-		SendStreamStats() rac2.RangeSendStreamStats
+		SendStreamStats(*rac2.RangeSendStreamStats)
 	},
 	allExistingReplicas []roachpb.ReplicaDescriptor,
 	exclReplsInNeedOfSnapshots bool,
@@ -2223,7 +2224,7 @@ func (a *Allocator) LeaseholderShouldMoveDueToPreferences(
 		preferred = excludeReplicasInNeedOfSnapshots(
 			ctx, leaseRepl.RaftStatus(), leaseRepl.GetFirstIndex(), preferred)
 		preferred = excludeReplicasInNeedOfCatchup(
-			ctx, leaseRepl.SendStreamStats(), preferred)
+			ctx, leaseRepl.SendStreamStats, preferred)
 	}
 	if len(preferred) == 0 {
 		return false
@@ -2282,7 +2283,7 @@ func (a *Allocator) TransferLeaseTarget(
 		GetRangeID() roachpb.RangeID
 		RaftStatus() *raft.Status
 		GetFirstIndex() kvpb.RaftIndex
-		SendStreamStats() rac2.RangeSendStreamStats
+		SendStreamStats(*rac2.RangeSendStreamStats)
 	},
 	usageInfo allocator.RangeUsageInfo,
 	forceDecisionWithoutStats bool,
@@ -2651,7 +2652,7 @@ func (a *Allocator) ShouldTransferLease(
 		StoreID() roachpb.StoreID
 		RaftStatus() *raft.Status
 		GetFirstIndex() kvpb.RaftIndex
-		SendStreamStats() rac2.RangeSendStreamStats
+		SendStreamStats(*rac2.RangeSendStreamStats)
 	},
 	usageInfo allocator.RangeUsageInfo,
 ) TransferLeaseDecision {
@@ -3054,12 +3055,20 @@ func excludeReplicasInNeedOfSnapshots(
 	return replicas[:filled]
 }
 
+// sendStreamStatsPool is a pool of RangeSendStreamStats objects, used to avoid
+// churning memory when computing lease transfer decisions.
+var sendStreamStatsPool = sync.Pool{
+	New: func() interface{} {
+		return &rac2.RangeSendStreamStats{}
+	},
+}
+
 // excludeReplicasInNeedOfCatchup filters out the `replicas` that may be in
 // need of a catchup messages before able to apply the lease, based on the
 // provided RangeSendStreamStats.
 func excludeReplicasInNeedOfCatchup(
 	ctx context.Context,
-	sendStreamStats rac2.RangeSendStreamStats,
+	sendStreamStats func(*rac2.RangeSendStreamStats),
 	replicas []roachpb.ReplicaDescriptor,
 ) []roachpb.ReplicaDescriptor {
 	if sendStreamStats == nil {
@@ -3069,9 +3078,13 @@ func excludeReplicasInNeedOfCatchup(
 		// recently became one (concurrent to the lease transfer decision).
 		return replicas
 	}
+	stats := sendStreamStatsPool.Get().(*rac2.RangeSendStreamStats)
+	stats.Clear()
+	defer sendStreamStatsPool.Put(stats)
+	sendStreamStats(stats)
 	filled := 0
 	for _, repl := range replicas {
-		if stats, ok := sendStreamStats[repl.ReplicaID]; ok &&
+		if stats, ok := stats.ReplicaSendStreamStats(repl.ReplicaID); ok &&
 			(!stats.IsStateReplicate || stats.HasSendQueue) {
 			log.KvDistribution.VEventf(ctx, 5,
 				"not considering %s as a potential candidate for a lease transfer "+

pkg/kv/kvserver/allocator/allocatorimpl/allocator_test.go

@@ -1943,19 +1943,22 @@ func (r *mockRepl) GetRangeID() roachpb.RangeID {
 	return roachpb.RangeID(0)
 }
 
-func (r *mockRepl) SendStreamStats() rac2.RangeSendStreamStats {
-	rangeStats := rac2.RangeSendStreamStats{}
+func (r *mockRepl) SendStreamStats(stats *rac2.RangeSendStreamStats) {
 	for i := int32(1); i <= r.replicationFactor; i++ {
-		replStats := rac2.ReplicaSendStreamStats{}
-		if _, ok := r.replsWithSendQueue[roachpb.ReplicaID(i)]; ok {
+		replicaID := roachpb.ReplicaID(i)
+		replStats := rac2.ReplicaSendStreamStats{
+			ReplicaSendQueueStats: rac2.ReplicaSendQueueStats{
+				ReplicaID: replicaID,
+			},
+		}
+		if _, ok := r.replsWithSendQueue[replicaID]; ok {
 			replStats.HasSendQueue = true
 		}
-		if _, ok := r.replsNotInStateReplicate[roachpb.ReplicaID(i)]; !ok {
+		if _, ok := r.replsNotInStateReplicate[replicaID]; !ok {
 			replStats.IsStateReplicate = true
 		}
-		rangeStats[roachpb.ReplicaID(i)] = replStats
+		stats.SetReplicaSendStreamStats(replStats)
 	}
-	return rangeStats
 }
 
 func (r *mockRepl) markReplAsNeedingSnapshot(id roachpb.ReplicaID) {

pkg/kv/kvserver/allocator/plan/replicate.go

@@ -100,7 +100,7 @@ type AllocatorReplica interface {
 	LastReplicaAdded() (roachpb.ReplicaID, time.Time)
 	StoreID() roachpb.StoreID
 	GetRangeID() roachpb.RangeID
-	SendStreamStats() rac2.RangeSendStreamStats
+	SendStreamStats(*rac2.RangeSendStreamStats)
 }
 
 // ReplicaPlanner implements the ReplicationPlanner interface.

pkg/kv/kvserver/asim/queue/allocator_replica.go

@@ -159,6 +159,5 @@ func (sr *SimulatorReplica) RangeUsageInfo() allocator.RangeUsageInfo {
 	return sr.usage
 }
 
-func (sr *SimulatorReplica) SendStreamStats() rac2.RangeSendStreamStats {
-	return nil
+func (sr *SimulatorReplica) SendStreamStats(stats *rac2.RangeSendStreamStats) {
 }

pkg/kv/kvserver/asim/storerebalancer/candidate_replica.go

@@ -110,9 +110,7 @@ func (sr *simulatorReplica) AdminTransferLease(
 	return nil
 }
 
-func (sr *simulatorReplica) SendStreamStats() rac2.RangeSendStreamStats {
-	return rac2.RangeSendStreamStats{}
-}
+func (sr *simulatorReplica) SendStreamStats(stats *rac2.RangeSendStreamStats) {}
 
 // Replica returns the underlying kvserver replica, however when called from
 // the simulator it only returns nil.