rac2,kvserver: add SendQueueLogger to periodically log per-stream send queue bytes

Informs #157793

Epic: none

Release note: None

Author: sumeerbhola

pkg/kv/kvserver/kvflowcontrol/rac2/BUILD.bazel

@@ -7,6 +7,7 @@ go_library(
         "metrics.go",
         "priority.go",
         "range_controller.go",
+        "send_queue_logger.go",
         "store_stream.go",
         "token_counter.go",
         "token_tracker.go",
@@ -49,6 +50,7 @@ go_test(
         "log_tracker_test.go",
         "priority_test.go",
         "range_controller_test.go",
+        "send_queue_logger_test.go",
         "simulation_test.go",
         "store_stream_test.go",
         "token_counter_test.go",
@@ -86,6 +88,7 @@ go_test(
         "//pkg/util/timeutil",
         "@com_github_cockroachdb_datadriven//:datadriven",
         "@com_github_cockroachdb_errors//:errors",
+        "@com_github_cockroachdb_redact//:redact",
         "@com_github_dustin_go_humanize//:go-humanize",
         "@com_github_gogo_protobuf//jsonpb",
         "@com_github_guptarohit_asciigraph//:asciigraph",

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

@@ -0,0 +1,140 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package rac2
+
+import (
+	"cmp"
+	"context"
+	"slices"
+	"time"
+
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvflowcontrol"
+	"github.com/cockroachdb/cockroach/pkg/util/humanizeutil"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
+	"github.com/cockroachdb/redact"
+)
+
+const sendQueueLoggingInterval = 30 * time.Second
+
+// SendQueueLogger logs send queue sizes for streams.
+//
+// Usage:
+//
+//		sql := NewSendQueueLogger(numStreamsToLog)
+//		// Periodically do:
+//	 if coll, ok := sql.TryStartLog(); ok {
+//		    for each range:
+//		        coll.ObserveRangeStats(rangeStats)
+//		    sql.FinishLog(ctx, coll)
+//		}
+//
+// It is thread-safe.
+type SendQueueLogger struct {
+	numStreamsToLog int
+	limiter         log.EveryN
+
+	mu struct {
+		syncutil.Mutex
+		scratch SendQueueLoggerCollector
+	}
+
+	testingLog func(context.Context, *redact.StringBuilder)
+}
+
+type streamAndBytes struct {
+	stream kvflowcontrol.Stream
+	bytes  int64
+}
+
+type SendQueueLoggerCollector struct {
+	m  map[kvflowcontrol.Stream]int64
+	sl []streamAndBytes
+}
+
+// NewSendQueueLogger creates a new SendQueueLogger that will log up to
+// numStreamsToLog streams with the highest send queue bytes when logging is
+// triggered.
+func NewSendQueueLogger(numStreamsToLog int) *SendQueueLogger {
+	return &SendQueueLogger{
+		numStreamsToLog: numStreamsToLog,
+		limiter:         log.Every(sendQueueLoggingInterval),
+	}
+}
+
+func (s *SendQueueLogger) TryStartLog() (SendQueueLoggerCollector, bool) {
+	if !s.limiter.ShouldLog() {
+		return SendQueueLoggerCollector{}, false
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	collector := s.mu.scratch
+	s.mu.scratch = SendQueueLoggerCollector{}
+	if collector.m == nil {
+		collector.m = make(map[kvflowcontrol.Stream]int64)
+	}
+	return collector, true
+}
+
+func (s *SendQueueLoggerCollector) ObserveRangeStats(stats *RangeSendStreamStats) {
+	for _, ss := range stats.internal {
+		if ss.SendQueueBytes > 0 {
+			s.m[ss.Stream] = s.m[ss.Stream] + ss.SendQueueBytes
+		}
+	}
+}
+
+// NB: this is defined here so we can access it from tests.
+const sendQueueLogFormat = "send queues: %s"
+
+func (s *SendQueueLogger) FinishLog(ctx context.Context, c SendQueueLoggerCollector) {
+	if len(c.m) == 0 {
+		return
+	}
+	defer func() {
+		clear(c.m)
+		c.sl = c.sl[:0]
+		s.mu.Lock()
+		s.mu.scratch = c
+		s.mu.Unlock()
+	}()
+	var buf redact.StringBuilder
+	c.finishLog(s.numStreamsToLog, &buf)
+	if fn := s.testingLog; fn != nil {
+		fn(ctx, &buf)
+	} else {
+		log.KvDistribution.Infof(ctx, sendQueueLogFormat, &buf)
+	}
+}
+
+func (c *SendQueueLoggerCollector) finishLog(maxNumStreams int, buf *redact.StringBuilder) {
+	for stream, bytes := range c.m {
+		c.sl = append(c.sl, streamAndBytes{
+			stream: stream,
+			bytes:  bytes,
+		})
+	}
+	// Sort by descending send queue bytes.
+	slices.SortFunc(c.sl, func(a, b streamAndBytes) int {
+		return -cmp.Compare(a.bytes, b.bytes)
+	})
+	n := min(maxNumStreams, len(c.sl))
+	for i := 0; i < n; i++ {
+		if i > 0 {
+			buf.SafeString(", ")
+		}
+		buf.Printf("%s: %s", c.sl[i].stream,
+			humanizeutil.IBytes(c.sl[i].bytes))
+	}
+	remainingBytes := int64(0)
+	for i := n; i < len(c.sl); i++ {
+		remainingBytes += c.sl[i].bytes
+	}
+	if remainingBytes > 0 {
+		buf.Printf(", + %s more bytes across %d streams",
+			humanizeutil.IBytes(remainingBytes), len(c.sl)-n)
+	}
+}

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

@@ -0,0 +1,103 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package rac2
+
+import (
+	"context"
+	"testing"
+
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvflowcontrol"
+	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/testutils/datapathutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/echotest"
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/redact"
+	"github.com/stretchr/testify/require"
+)
+
+func TestSendQueueLogger(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	s := log.ScopeWithoutShowLogs(t)
+	defer s.Close(t)
+
+	rss11_100 := ReplicaSendStreamStats{
+		Stream: kvflowcontrol.Stream{
+			TenantID: roachpb.MustMakeTenantID(1),
+			StoreID:  1,
+		},
+		ReplicaSendQueueStats: ReplicaSendQueueStats{
+			SendQueueBytes: 100,
+		},
+	}
+	rss22_200 := ReplicaSendStreamStats{
+		Stream: kvflowcontrol.Stream{
+			TenantID: roachpb.MustMakeTenantID(2),
+			StoreID:  2,
+		},
+		ReplicaSendQueueStats: ReplicaSendQueueStats{
+			SendQueueBytes: 200,
+		},
+	}
+	rss33_50 := ReplicaSendStreamStats{
+		Stream: kvflowcontrol.Stream{
+			TenantID: roachpb.MustMakeTenantID(3),
+			StoreID:  3,
+		},
+		ReplicaSendQueueStats: ReplicaSendQueueStats{
+			SendQueueBytes: 50,
+		},
+	}
+	rss44_25 := ReplicaSendStreamStats{
+		Stream: kvflowcontrol.Stream{
+			TenantID: roachpb.MustMakeTenantID(4),
+			StoreID:  4,
+		},
+		ReplicaSendQueueStats: ReplicaSendQueueStats{
+			SendQueueBytes: 25,
+		},
+	}
+	sql := NewSendQueueLogger(2)
+	var output redact.RedactableString
+	sql.testingLog = func(ctx context.Context, buf *redact.StringBuilder) {
+		if len(output) > 0 {
+			output += "\n"
+		}
+		output += redact.Sprintf(sendQueueLogFormat, buf)
+	}
+	ctx := context.Background()
+	coll, ok := sql.TryStartLog()
+
+	require.True(t, ok)
+	// Empty stats are ok.
+	stats := RangeSendStreamStats{}
+	coll.ObserveRangeStats(&stats)
+	stats = RangeSendStreamStats{
+		internal: []ReplicaSendStreamStats{rss22_200, rss33_50},
+	}
+	coll.ObserveRangeStats(&stats)
+	stats = RangeSendStreamStats{
+		internal: []ReplicaSendStreamStats{rss11_100, rss22_200},
+	}
+	coll.ObserveRangeStats(&stats)
+	stats = RangeSendStreamStats{
+		internal: []ReplicaSendStreamStats{rss11_100},
+	}
+	coll.ObserveRangeStats(&stats)
+	stats = RangeSendStreamStats{
+		internal: []ReplicaSendStreamStats{rss33_50, rss44_25},
+	}
+	coll.ObserveRangeStats(&stats)
+	sql.FinishLog(ctx, coll)
+
+	// Cannot log again immediately.
+	_, ok = sql.TryStartLog()
+	require.False(t, ok)
+	// Call FinishLog again to ensure no logging happens.
+	sql.FinishLog(ctx, SendQueueLoggerCollector{})
+
+	echotest.Require(t, string(output), datapathutils.TestDataPath(t, t.Name()+".txt"))
+}

pkg/kv/kvserver/kvflowcontrol/rac2/testdata/TestSendQueueLogger.txt

@@ -0,0 +1,3 @@
+echo
+----
+send queues: t2/s2: 400 B, t1/s1: 200 B, + 125 B more bytes across 2 streams

pkg/kv/kvserver/stores.go

@@ -14,6 +14,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvadmission"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvflowcontrol/rac2"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/rangefeed"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/storage"
@@ -44,6 +45,7 @@ type Stores struct {
 		biLatestTS hlc.Timestamp         // Timestamp of gossip bootstrap info
 		latestBI   *gossip.BootstrapInfo // Latest cached bootstrap info
 	}
+	sendQueueLogger *rac2.SendQueueLogger
 }
 
 var _ kv.Sender = &Stores{}      // Stores implements the client.Sender interface
@@ -52,9 +54,11 @@ var _ gossip.Storage = &Stores{} // Stores implements the gossip.Storage interfa
 // NewStores returns a local-only sender which directly accesses
 // a collection of stores.
 func NewStores(ambient log.AmbientContext, clock *hlc.Clock) *Stores {
+	const numStreamsToLog = 20
 	return &Stores{
-		AmbientContext: ambient,
-		clock:          clock,
+		AmbientContext:  ambient,
+		clock:           clock,
+		sendQueueLogger: rac2.NewSendQueueLogger(numStreamsToLog),
 	}
 }
 
@@ -354,3 +358,21 @@ func (ls *Stores) GetAggregatedStoreStats(
 	}
 	return storesCPURate, numStores, nil
 }
+
+func (ls *Stores) TryLogFlowControlSendQueues(ctx context.Context) {
+	coll, ok := ls.sendQueueLogger.TryStartLog()
+	if !ok {
+		return
+	}
+	rangeSendStats := rac2.RangeSendStreamStats{}
+	_ = ls.VisitStores(func(s *Store) error {
+		newStoreReplicaVisitor(s).Visit(func(rep *Replica) bool {
+			rangeSendStats.Clear()
+			rep.SendStreamStats(&rangeSendStats)
+			coll.ObserveRangeStats(&rangeSendStats)
+			return true
+		})
+		return nil
+	})
+	ls.sendQueueLogger.FinishLog(ctx, coll)
+}

pkg/server/node.go

@@ -1239,6 +1239,7 @@ func (n *Node) computeMetricsPeriodically(
 	})
 	n.updateNodeRangeCount()
 	n.storeCfg.KVFlowStreamTokenProvider.UpdateMetricGauges()
+	n.stores.TryLogFlowControlSendQueues(ctx)
 	return err
 }