rac2: add kvflowcontrol.Stream to ReplicaSendStreamStats

Informs #157793

Epic: none

Release note: None

Author: sumeerbhola

pkg/kv/kvserver/kvflowcontrol/rac2/range_controller.go

@@ -320,6 +320,8 @@ func (q *RangeSendQueueStats) Clear() {
 // ReplicaSendStreamStats contains the stats for a replica send stream that may
 // be used to inform placement decisions pertaining to the replica.
 type ReplicaSendStreamStats struct {
+	// Stream is the flow control stream for the replica.
+	Stream kvflowcontrol.Stream
 	// IsStateReplicate is true iff the replica is being sent entries.
 	IsStateReplicate bool
 	// HasSendQueue is true when a replica has a non-zero amount of queued
@@ -1721,6 +1723,7 @@ func (rc *rangeController) SendStreamStats(statsToSet *RangeSendStreamStats) {
 		// end up overwriting the same state at most twice, not a big issue.
 		for _, vs := range vss {
 			stats := ReplicaSendStreamStats{
+				Stream:           vs.stateForWaiters.evalTokenCounter.stream,
 				IsStateReplicate: vs.isStateReplicate,
 				HasSendQueue:     vs.hasSendQ,
 			}
@@ -1731,6 +1734,7 @@ func (rc *rangeController) SendStreamStats(statsToSet *RangeSendStreamStats) {
 	// Now handle the non-voters.
 	for _, nv := range rc.mu.nonVoterSet {
 		stats := ReplicaSendStreamStats{
+			Stream:           nv.evalTokenCounter.stream,
 			IsStateReplicate: nv.isStateReplicate,
 			HasSendQueue:     nv.hasSendQ,
 		}

pkg/kv/kvserver/kvflowcontrol/rac2/range_controller_test.go

@@ -1580,8 +1580,9 @@ func TestRangeController(t *testing.T) {
 				for _, repl := range sortReplicas(r) {
 					replStats, ok := stats.ReplicaSendStreamStats(repl.ReplicaID)
 					require.True(t, ok)
-					buf.WriteString(fmt.Sprintf("%v: is_state_replicate=%-5v has_send_queue=%-5v send_queue_size=%v / %v entries\n",
+					buf.WriteString(fmt.Sprintf("%v: stream=%v is_state_replicate=%-5v has_send_queue=%-5v send_queue_size=%v / %v entries\n",
 						repl,
+						replStats.Stream,
 						replStats.IsStateReplicate,
 						replStats.HasSendQueue,
 						// Cast for formatting.
@@ -2476,6 +2477,10 @@ func TestRangeSendStreamStatsString(t *testing.T) {
 	stats := RangeSendStreamStats{
 		internal: []ReplicaSendStreamStats{
 			{
+				Stream: kvflowcontrol.Stream{
+					TenantID: roachpb.MustMakeTenantID(1),
+					StoreID:  roachpb.StoreID(1),
+				},
 				IsStateReplicate: false,
 				HasSendQueue:     true,
 				ReplicaSendQueueStats: ReplicaSendQueueStats{
@@ -2485,6 +2490,10 @@ func TestRangeSendStreamStatsString(t *testing.T) {
 				},
 			},
 			{
+				Stream: kvflowcontrol.Stream{
+					TenantID: roachpb.MustMakeTenantID(2),
+					StoreID:  roachpb.StoreID(2),
+				},
 				IsStateReplicate: true,
 				HasSendQueue:     false,
 				ReplicaSendQueueStats: ReplicaSendQueueStats{

pkg/kv/kvserver/kvflowcontrol/rac2/testdata/range_controller/send_stream_stats

@@ -62,15 +62,15 @@ NormalPri:
 # queue size for s2.
 send_stream_stats range_id=1 refresh=false
 ----
-(n1,s1):1: is_state_replicate=true  has_send_queue=false send_queue_size=+0 B / 0 entries
-(n2,s2):2: is_state_replicate=false has_send_queue=true  send_queue_size=+0 B / 0 entries
-(n3,s3):3: is_state_replicate=true  has_send_queue=true  send_queue_size=+0 B / 0 entries
+(n1,s1):1: stream=t1/s1 is_state_replicate=true  has_send_queue=false send_queue_size=+0 B / 0 entries
+(n2,s2):2: stream=t1/s2 is_state_replicate=false has_send_queue=true  send_queue_size=+0 B / 0 entries
+(n3,s3):3: stream=t1/s3 is_state_replicate=true  has_send_queue=true  send_queue_size=+0 B / 0 entries
 
 send_stream_stats range_id=1 refresh=true
 ----
-(n1,s1):1: is_state_replicate=true  has_send_queue=false send_queue_size=+0 B / 0 entries
-(n2,s2):2: is_state_replicate=false has_send_queue=true  send_queue_size=+0 B / 0 entries
-(n3,s3):3: is_state_replicate=true  has_send_queue=true  send_queue_size=+2.0 MiB / 2 entries
+(n1,s1):1: stream=t1/s1 is_state_replicate=true  has_send_queue=false send_queue_size=+0 B / 0 entries
+(n2,s2):2: stream=t1/s2 is_state_replicate=false has_send_queue=true  send_queue_size=+0 B / 0 entries
+(n3,s3):3: stream=t1/s3 is_state_replicate=true  has_send_queue=true  send_queue_size=+2.0 MiB / 2 entries
 
 # Next, add another entry, which will also be queued. We want to see the stats
 # tick over and update by themselves (refresh=false) by ticking the clock the
@@ -94,9 +94,9 @@ t1/s3: eval reg=+1.0 MiB/+16 MiB ela=-7.0 MiB/+8.0 MiB
 # The stats shouldn't have changed, as we haven't refreshed them via ticking.
 send_stream_stats range_id=1 refresh=false
 ----
-(n1,s1):1: is_state_replicate=true  has_send_queue=false send_queue_size=+0 B / 0 entries
-(n2,s2):2: is_state_replicate=false has_send_queue=true  send_queue_size=+0 B / 0 entries
-(n3,s3):3: is_state_replicate=true  has_send_queue=true  send_queue_size=+2.0 MiB / 2 entries
+(n1,s1):1: stream=t1/s1 is_state_replicate=true  has_send_queue=false send_queue_size=+0 B / 0 entries
+(n2,s2):2: stream=t1/s2 is_state_replicate=false has_send_queue=true  send_queue_size=+0 B / 0 entries
+(n3,s3):3: stream=t1/s3 is_state_replicate=true  has_send_queue=true  send_queue_size=+2.0 MiB / 2 entries
 
 tick duration=7s
 ----
@@ -118,16 +118,16 @@ t1/s3: eval reg=+1.0 MiB/+16 MiB ela=-7.0 MiB/+8.0 MiB
 # The stats now should have been populated.
 send_stream_stats range_id=1 refresh=false
 ----
-(n1,s1):1: is_state_replicate=true  has_send_queue=false send_queue_size=+0 B / 0 entries
-(n2,s2):2: is_state_replicate=false has_send_queue=true  send_queue_size=+0 B / 0 entries
-(n3,s3):3: is_state_replicate=true  has_send_queue=true  send_queue_size=+14 MiB / 3 entries
+(n1,s1):1: stream=t1/s1 is_state_replicate=true  has_send_queue=false send_queue_size=+0 B / 0 entries
+(n2,s2):2: stream=t1/s2 is_state_replicate=false has_send_queue=true  send_queue_size=+0 B / 0 entries
+(n3,s3):3: stream=t1/s3 is_state_replicate=true  has_send_queue=true  send_queue_size=+14 MiB / 3 entries
 
 # Sanity check they are the same when refreshing.
 send_stream_stats range_id=1 refresh=true
 ----
-(n1,s1):1: is_state_replicate=true  has_send_queue=false send_queue_size=+0 B / 0 entries
-(n2,s2):2: is_state_replicate=false has_send_queue=true  send_queue_size=+0 B / 0 entries
-(n3,s3):3: is_state_replicate=true  has_send_queue=true  send_queue_size=+14 MiB / 3 entries
+(n1,s1):1: stream=t1/s1 is_state_replicate=true  has_send_queue=false send_queue_size=+0 B / 0 entries
+(n2,s2):2: stream=t1/s2 is_state_replicate=false has_send_queue=true  send_queue_size=+0 B / 0 entries
+(n3,s3):3: stream=t1/s3 is_state_replicate=true  has_send_queue=true  send_queue_size=+14 MiB / 3 entries
 
 close_rcs
 ----