kvserver/rangefeed: add metrics for lockedMuxStream.Send calls

Adds observability into the latency of `lockedMuxStream.Send` with a histogram
metric that uses the `IOLatencyBuckets` bucket configuration (see #147440 (comment)).

Also adds a counter for the number of slow sends, using the same criteria as the
log line added in #146765.

Resolves: #147435
Epic: CRDB-51061
Release note: None

Author: Michael Xu

docs/generated/metrics/metrics.yaml

@@ -12370,6 +12370,22 @@ layers:
       unit: BYTES
       aggregation: AVG
       derivative: NONE
+    - name: kv.rangefeed.mux_stream_send.latency
+      exported_name: kv_rangefeed_mux_stream_send_latency
+      description: Latency of sending RangeFeed events to the client
+      y_axis_label: Latency
+      type: HISTOGRAM
+      unit: NANOSECONDS
+      aggregation: AVG
+      derivative: NONE
+    - name: kv.rangefeed.mux_stream_send.slow_events
+      exported_name: kv_rangefeed_mux_stream_send_slow_events
+      description: Number of RangeFeed events that took longer than 10s to send to the client
+      y_axis_label: Events
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
     - name: kv.rangefeed.output_loop_unbuffered_registration_nanos
       exported_name: kv_rangefeed_output_loop_unbuffered_registration_nanos
       description: Duration of the Rangefeed O(range) output loop goroutine. This is only applicable for unbuffered registrations since buffered registrations spawns long-living goroutines.

pkg/kv/kvserver/rangefeed/metrics.go

@@ -289,3 +289,38 @@ func NewBufferedSenderMetrics() *BufferedSenderMetrics {
 		BufferedSenderQueueSize: metric.NewGauge(metaBufferedSenderQueueSize),
 	}
 }
+
+var (
+	metaRangeFeedMuxStreamSendLatencyNanos = metric.Metadata{
+		Name:        "kv.rangefeed.mux_stream_send.latency",
+		Help:        "Latency of sending RangeFeed events to the client",
+		Measurement: "Latency",
+		Unit:        metric.Unit_NANOSECONDS,
+	}
+	metaRangeFeedMuxStreamSlowSends = metric.Metadata{
+		Name:        "kv.rangefeed.mux_stream_send.slow_events",
+		Help:        "Number of RangeFeed events that took longer than 10s to send to the client",
+		Measurement: "Events",
+		Unit:        metric.Unit_COUNT,
+	}
+)
+
+type LockedMuxStreamMetrics struct {
+	SendLatencyNanos metric.IHistogram
+	SlowSends        *metric.Counter
+}
+
+// MetricStruct implements metrics.Struct interface.
+func (*LockedMuxStreamMetrics) MetricStruct() {}
+
+func NewLockedMuxStreamMetrics(histogramWindow time.Duration) *LockedMuxStreamMetrics {
+	return &LockedMuxStreamMetrics{
+		SendLatencyNanos: metric.NewHistogram(metric.HistogramOptions{
+			Mode:         metric.HistogramModePrometheus,
+			Metadata:     metaRangeFeedMuxStreamSendLatencyNanos,
+			Duration:     histogramWindow,
+			BucketConfig: metric.IOLatencyBuckets,
+		}),
+		SlowSends: metric.NewCounter(metaRangeFeedMuxStreamSlowSends),
+	}
+}

pkg/server/node.go

@@ -260,7 +260,8 @@ type nodeMetrics struct {
 	StreamManagerMetrics *rangefeed.StreamManagerMetrics
 	// BufferedSenderMetrics is for monitoring of BufferedSenders for rangefeed.
 	// Note that there could be multiple buffered senders in a node.
-	BufferedSenderMetrics *rangefeed.BufferedSenderMetrics
+	BufferedSenderMetrics  *rangefeed.BufferedSenderMetrics
+	LockedMuxStreamMetrics *rangefeed.LockedMuxStreamMetrics
 }
 
 func makeNodeMetrics(reg *metric.Registry, histogramWindow time.Duration) *nodeMetrics {
@@ -282,6 +283,7 @@ func makeNodeMetrics(reg *metric.Registry, histogramWindow time.Duration) *nodeM
 		CrossZoneBatchResponseBytes:   metric.NewCounter(metaCrossZoneBatchResponse),
 		StreamManagerMetrics:          rangefeed.NewStreamManagerMetrics(),
 		BufferedSenderMetrics:         rangefeed.NewBufferedSenderMetrics(),
+		LockedMuxStreamMetrics:        rangefeed.NewLockedMuxStreamMetrics(histogramWindow),
 	}
 
 	for i := range nm.MethodCounts {
@@ -2104,6 +2106,7 @@ func (n *Node) RangeLookup(
 type lockedMuxStream struct {
 	wrapped kvpb.Internal_MuxRangeFeedServer
 	sendMu  syncutil.Mutex
+	metrics *rangefeed.LockedMuxStreamMetrics
 }
 
 func (s *lockedMuxStream) SendIsThreadSafe() {}
@@ -2122,10 +2125,11 @@ func (s *lockedMuxStream) Send(e *kvpb.MuxRangeFeedEvent) error {
 	// factors, e.g., the number of rangefeeds contending for the lockedMuxStream.
 	start := crtime.NowMono()
 	defer func() {
-		if dur := start.Elapsed(); dur > slowMuxStreamSendThreshold {
-			log.Infof(s.wrapped.Context(),
-				"slow send on stream %d for r%d took %s",
-				e.StreamID, e.RangeID, dur)
+		dur := start.Elapsed()
+		s.metrics.SendLatencyNanos.RecordValue(dur.Nanoseconds())
+		if dur > slowMuxStreamSendThreshold {
+			s.metrics.SlowSends.Inc(1)
+			log.Infof(s.wrapped.Context(), "slow send on stream %d for r%d took %s", e.StreamID, e.RangeID, dur)
 		}
 	}()
 
@@ -2196,7 +2200,10 @@ func (n *Node) defaultRangefeedConsumerID() int64 {
 
 // MuxRangeFeed implements the roachpb.InternalServer interface.
 func (n *Node) MuxRangeFeed(muxStream kvpb.Internal_MuxRangeFeedServer) error {
-	lockedMuxStream := &lockedMuxStream{wrapped: muxStream}
+	lockedMuxStream := &lockedMuxStream{
+		wrapped: muxStream,
+		metrics: n.metrics.LockedMuxStreamMetrics,
+	}
 
 	// All context created below should derive from this context, which is
 	// cancelled once MuxRangeFeed exits.