Skip to content

Commit

Permalink
Merge #131857
Browse files Browse the repository at this point in the history 
131857: .*: introduce send queue and range controller metrics r=sumeerbhola a=kvoli

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` method is also split into two methods:

```
SendStreamStatsRaftMuLocked
BasicSendStreamStats
```

In order to enable efficient send queue gauge update calls, and any
future allocator interaction which requires the send queue size
(bytes|count). The allocator now uses `BasicSendStreamStats`, while
metrics uses the more expensive `SendStreamStatsRaftMuLocked`

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
```


Co-authored-by: Austen McClernon <[email protected]>
  • Loading branch information
@kvoli
craig[bot] and kvoli committed Oct 9, 2024
2 parents 9999e3a + fe3b700 commit 58c475d
Show file tree
Hide file tree
Showing 56 changed files with 3,581 additions and 2,357 deletions.
17 changes: 13 additions & 4 deletions docs/generated/metrics/metrics.html
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand All @@ -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>
Expand Down
29 changes: 21 additions & 8 deletions pkg/kv/kvserver/allocator/allocatorimpl/allocator.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import (
"fmt"
"math"
"math/rand"
"sync"
"time"

"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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 {
Expand All @@ -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 "+
Expand Down
17 changes: 10 additions & 7 deletions pkg/kv/kvserver/allocator/allocatorimpl/allocator_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
2 changes: 1 addition & 1 deletion pkg/kv/kvserver/allocator/plan/replicate.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
3 changes: 1 addition & 2 deletions pkg/kv/kvserver/asim/queue/allocator_replica.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
}
4 changes: 1 addition & 3 deletions pkg/kv/kvserver/asim/storerebalancer/candidate_replica.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
Loading

0 comments on commit 58c475d

Please sign in to comment.