kvcoord: Pace rangefeed client goroutine creation

Acquire catchup scan quota prior to goroutine creation
in order to pace the goroutine creation rate.

This change results in nice and smooth growth in
goroutine count, thus reducing the pressure on goroutine
scheduler, which in turn reduces the impact on SQL
latency during changefeed startup.

This change also improves observability in rangefeed
client by introducing new counters:
 * `distsender.rangefeed.post_catchup_ranges`: gauge keeping track
    of the number of ranges which have completed their catchup scan.
 * `distsender.rangefeed.retry.<reason>`: counter keeping track
    of the number of ranges that ecountered a retryable error of
    particular type (e.g. slow counsumer, range split, etc).

Observability also enhanced by adding a column to
 `crdb_internal.active_rangefeed` virtual table augment to indicate
if the range is currently in catchup scan mode.

Fixes #98842

Release note: None

pkg/cli/zip_table_registry.go

@@ -598,6 +598,7 @@ var zipInternalTablesPerNode = DebugZipTableRegistry{
 			"range_end",
 			"resolved",
 			"last_event_utc",
+			"catchup",
 		},
 	},
 	"crdb_internal.feature_usage": {

pkg/kv/kvclient/kvcoord/dist_sender.go

@@ -206,6 +206,15 @@ This counts the number of ranges with an active rangefeed.
 		Help: `Number of ranges in catchup mode
 
 This counts the number of ranges with an active rangefeed that are performing catchup scan.
+`,
+		Measurement: "Ranges",
+		Unit:        metric.Unit_COUNT,
+	}
+	metaDistSenderRangefeedPostCatchupRanges = metric.Metadata{
+		Name: "distsender.rangefeed.post_catchup_ranges",
+		Help: `Number of post-catchup ranges
+
+This counts the number of ranges that are active, and have completed their catchup scan
 `,
 		Measurement: "Ranges",
 		Unit:        metric.Unit_COUNT,
@@ -308,11 +317,13 @@ type DistSenderMetrics struct {
 
 // DistSenderRangeFeedMetrics is a set of rangefeed specific metrics.
 type DistSenderRangeFeedMetrics struct {
-	RangefeedRanges        *metric.Gauge
-	RangefeedCatchupRanges *metric.Gauge
-	RangefeedErrorCatchup  *metric.Counter
-	RangefeedRestartRanges *metric.Counter
-	RangefeedRestartStuck  *metric.Counter
+	RangefeedRanges            *metric.Gauge
+	RangefeedCatchupRanges     *metric.Gauge
+	RangefeedPostCatchupRanges *metric.Gauge
+	RangefeedErrorCatchup      *metric.Counter
+	RangefeedRetryErrors       [kvpb.NumRangeFeedRetryErrors]*metric.Counter
+	RangefeedRestartRanges     *metric.Counter
+	RangefeedRestartStuck      *metric.Counter
 }
 
 func makeDistSenderMetrics() DistSenderMetrics {
@@ -354,12 +365,24 @@ func makeDistSenderMetrics() DistSenderMetrics {
 }
 
 func makeDistSenderRangeFeedMetrics() DistSenderRangeFeedMetrics {
+	var retryCounters [kvpb.NumRangeFeedRetryErrors]*metric.Counter
+	for idx, name := range kvpb.RangeFeedRetryError_Reason_name {
+		retryCounters[idx] = metric.NewCounter(metric.Metadata{
+			Name:        fmt.Sprintf("distsender.rangefeed.retry.%s", strings.ToLower(name)),
+			Help:        `Number of ranges in retried due to rangefeed retry error`,
+			Measurement: "Ranges",
+			Unit:        metric.Unit_COUNT,
+		})
+	}
+
 	return DistSenderRangeFeedMetrics{
-		RangefeedRanges:        metric.NewGauge(metaDistSenderRangefeedTotalRanges),
-		RangefeedCatchupRanges: metric.NewGauge(metaDistSenderRangefeedCatchupRanges),
-		RangefeedErrorCatchup:  metric.NewCounter(metaDistSenderRangefeedErrorCatchupRanges),
-		RangefeedRestartRanges: metric.NewCounter(metaDistSenderRangefeedRestartRanges),
-		RangefeedRestartStuck:  metric.NewCounter(metaDistSenderRangefeedRestartStuck),
+		RangefeedRanges:            metric.NewGauge(metaDistSenderRangefeedTotalRanges),
+		RangefeedCatchupRanges:     metric.NewGauge(metaDistSenderRangefeedCatchupRanges),
+		RangefeedPostCatchupRanges: metric.NewGauge(metaDistSenderRangefeedPostCatchupRanges),
+		RangefeedRetryErrors:       retryCounters,
+		RangefeedErrorCatchup:      metric.NewCounter(metaDistSenderRangefeedErrorCatchupRanges),
+		RangefeedRestartRanges:     metric.NewCounter(metaDistSenderRangefeedRestartRanges),
+		RangefeedRestartStuck:      metric.NewCounter(metaDistSenderRangefeedRestartStuck),
 	}
 }
 

pkg/kv/kvclient/kvcoord/dist_sender_mux_rangefeed.go

@@ -95,7 +95,10 @@ func muxRangeFeed(
 		m.metrics = cfg.knobs.metrics
 	}
 
-	divideAllSpansOnRangeBoundaries(spans, m.startSingleRangeFeed, ds, &m.g)
+	m.g.GoCtx(func(ctx context.Context) error {
+		return divideAllSpansOnRangeBoundaries(ctx, spans, m.startSingleRangeFeed, ds, &m.g)
+	})
+
 	return errors.CombineErrors(m.g.Wait(), ctx.Err())
 }
 
@@ -165,12 +168,6 @@ type activeMuxRangeFeed struct {
 	roachpb.ReplicaDescriptor
 	startAfter hlc.Timestamp
 
-	// catchupRes is the catchup scan quota acquired upon the
-	// start of rangefeed.
-	// It is released when this stream receives first non-empty checkpoint
-	// (meaning: catchup scan completes).
-	catchupRes catchupAlloc
-
 	// State pertaining to execution of rangefeed call.
 	token     rangecache.EvictionToken
 	transport Transport
@@ -218,7 +215,7 @@ func (m *rangefeedMuxer) startSingleRangeFeed(
 
 	// Register active mux range feed.
 	stream := &activeMuxRangeFeed{
-		activeRangeFeed: newActiveRangeFeed(span, startAfter, m.registry, m.metrics.RangefeedRanges),
+		activeRangeFeed: newActiveRangeFeed(span, startAfter, m.registry, m.metrics),
 		rSpan:           rs,
 		startAfter:      startAfter,
 		token:           token,
@@ -409,7 +406,7 @@ func (m *rangefeedMuxer) startNodeMuxRangeFeed(
 
 		// make sure that the underlying error is not fatal. If it is, there is no
 		// reason to restart each rangefeed, so just bail out.
-		if _, err := handleRangefeedError(ctx, recvErr); err != nil {
+		if _, err := handleRangefeedError(ctx, m.metrics, recvErr); err != nil {
 			// Regardless of an error, release any resources (i.e. metrics) still
 			// being held by active stream.
 			for _, s := range toRestart {
@@ -468,8 +465,8 @@ func (m *rangefeedMuxer) receiveEventsFromNode(
 			if t.Span.Contains(active.Span) {
 				// If we see the first non-empty checkpoint, we know we're done with the catchup scan.
 				if !t.ResolvedTS.IsEmpty() && active.catchupRes != nil {
-					active.catchupRes.Release()
-					active.catchupRes = nil
+					active.releaseCatchupScan()
+					m.metrics.RangefeedPostCatchupRanges.Inc(1)
 				}
 				// Note that this timestamp means that all rows in the span with
 				// writes at or before the timestamp have now been seen. The
@@ -524,10 +521,7 @@ func (m *rangefeedMuxer) restartActiveRangeFeed(
 
 	// Release catchup scan reservation if any -- we will acquire another
 	// one when we restart.
-	if active.catchupRes != nil {
-		active.catchupRes.Release()
-		active.catchupRes = nil
-	}
+	active.releaseCatchupScan()
 
 	doRelease := true
 	defer func() {
@@ -536,7 +530,7 @@ func (m *rangefeedMuxer) restartActiveRangeFeed(
 		}
 	}()
 
-	errInfo, err := handleRangefeedError(ctx, reason)
+	errInfo, err := handleRangefeedError(ctx, m.metrics, reason)
 	if err != nil {
 		// If this is an error we cannot recover from, terminate the rangefeed.
 		return err