kvserver/rangefeed: add kv.rangefeed.output_loop_nanos

This patch introduces a new metric, kv.rangefeed.output_loop_nanos, which tracks
the duration of the per-range, per-registration runOutputLoop goroutine. This
goroutine is expected to be short-lived for unbuffered registrations, while it
remains long-lived for buffered registrations. Note that this metric is only
non-zero for unbuffered registrations.

Part of: #129816
Release note: none

docs/generated/metrics/metrics.html

@@ -248,6 +248,7 @@
 <tr><td>STORAGE</td><td>kv.rangefeed.closed_timestamp_max_behind_nanos</td><td>Largest latency between realtime and replica max closed timestamp for replicas that have active rangeeds on them</td><td>Nanoseconds</td><td>GAUGE</td><td>NANOSECONDS</td><td>AVG</td><td>NONE</td></tr>
 <tr><td>STORAGE</td><td>kv.rangefeed.mem_shared</td><td>Memory usage by rangefeeds</td><td>Memory</td><td>GAUGE</td><td>BYTES</td><td>AVG</td><td>NONE</td></tr>
 <tr><td>STORAGE</td><td>kv.rangefeed.mem_system</td><td>Memory usage by rangefeeds on system ranges</td><td>Memory</td><td>GAUGE</td><td>BYTES</td><td>AVG</td><td>NONE</td></tr>
+<tr><td>STORAGE</td><td>kv.rangefeed.output_loop_nanos</td><td>Duration of the Rangefeed O(range) output loop goroutine</td><td>Nanoseconds</td><td>COUNTER</td><td>NANOSECONDS</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>STORAGE</td><td>kv.rangefeed.processors_goroutine</td><td>Number of active RangeFeed processors using goroutines</td><td>Processors</td><td>GAUGE</td><td>COUNT</td><td>AVG</td><td>NONE</td></tr>
 <tr><td>STORAGE</td><td>kv.rangefeed.processors_scheduler</td><td>Number of active RangeFeed processors using scheduler</td><td>Processors</td><td>GAUGE</td><td>COUNT</td><td>AVG</td><td>NONE</td></tr>
 <tr><td>STORAGE</td><td>kv.rangefeed.registrations</td><td>Number of active RangeFeed registrations</td><td>Registrations</td><td>GAUGE</td><td>COUNT</td><td>AVG</td><td>NONE</td></tr>

pkg/kv/kvserver/rangefeed/metrics.go

@@ -14,6 +14,12 @@ import (
 )
 
 var (
+	metaRangeFeedOutputLoopNanos = metric.Metadata{
+		Name:        "kv.rangefeed.output_loop_nanos",
+		Help:        "Duration of the Rangefeed O(range) output loop goroutine",
+		Measurement: "Nanoseconds",
+		Unit:        metric.Unit_NANOSECONDS,
+	}
 	metaRangeFeedCatchUpScanNanos = metric.Metadata{
 		Name:        "kv.rangefeed.catchup_scan_nanos",
 		Help:        "Time spent in RangeFeed catchup scan",
@@ -98,6 +104,7 @@ var (
 
 // Metrics are for production monitoring of RangeFeeds.
 type Metrics struct {
+	RangefeedOutputLoopNanos               *metric.Counter
 	RangeFeedCatchUpScanNanos              *metric.Counter
 	RangeFeedBudgetExhausted               *metric.Counter
 	RangefeedProcessorQueueTimeout         *metric.Counter
@@ -126,6 +133,7 @@ func (*Metrics) MetricStruct() {}
 // NewMetrics makes the metrics for RangeFeeds monitoring.
 func NewMetrics() *Metrics {
 	return &Metrics{
+		RangefeedOutputLoopNanos:               metric.NewCounter(metaRangeFeedOutputLoopNanos),
 		RangeFeedCatchUpScanNanos:              metric.NewCounter(metaRangeFeedCatchUpScanNanos),
 		RangefeedProcessorQueueTimeout:         metric.NewCounter(metaQueueTimeout),
 		RangeFeedBudgetExhausted:               metric.NewCounter(metaRangeFeedExhausted),

pkg/kv/kvserver/rangefeed/unbuffered_registration.go

@@ -223,9 +223,13 @@ func (ubr *unbufferedRegistration) IsDisconnected() bool {
 //
 // nolint:deferunlockcheck
 func (ubr *unbufferedRegistration) runOutputLoop(ctx context.Context, forStacks roachpb.RangeID) {
+	start := timeutil.Now()
 	ubr.mu.Lock()
-	// Noop if publishCatchUpBuffer below returns no error.
-	defer ubr.drainAllocations(ctx)
+	defer func() {
+		// Noop if publishCatchUpBuffer below returns no error.
+		ubr.drainAllocations(ctx)
+		ubr.metrics.RangefeedOutputLoopNanos.Inc(timeutil.Since(start).Nanoseconds())
+	}()
 
 	if ubr.mu.disconnected {
 		ubr.mu.Unlock()