changefeedccl: add sink backpressure time metric

Add a new histogram metric `changefeed.sink_backpressure_nanos` that
measures time spent waiting for quota when emitting to the sink.
This provides visibility into downstream sink backpressure affecting
changefeed performance.

The metric is recorded per changefeed scope in batching_sink when
AdmitRequest returns ErrNotEnoughQuota, measuring the time from when
we start waiting for quota until it becomes available.

Includes unit test that exercises the metric under induced backpressure.

Fixes: #148417
Release note (enterprise change): Added changefeed.sink_backpressure_nanos
metric to track time spent waiting for quota when emitting to the sink.

Author: asg0451

docs/generated/metrics/metrics.yaml

@@ -1456,6 +1456,14 @@ layers:
       unit: NANOSECONDS
       aggregation: AVG
       derivative: NON_NEGATIVE_DERIVATIVE
+    - name: changefeed.sink_backpressure_nanos
+      exported_name: changefeed_sink_backpressure_nanos
+      description: Time spent waiting for quota when emitting to the sink (back-pressure). Only populated for sinks using the batching_sink wrapper. As of writing, this includes Kafka (v2), Pub/Sub (v2), and Webhook (v2).
+      y_axis_label: Nanoseconds
+      type: HISTOGRAM
+      unit: NANOSECONDS
+      aggregation: AVG
+      derivative: NONE
     - name: changefeed.sink_batch_hist_nanos
       exported_name: changefeed_sink_batch_hist_nanos
       description: Time spent batched in the sink buffer before being flushed and acknowledged

pkg/ccl/changefeedccl/batching_sink.go

@@ -409,6 +409,8 @@ func (s *batchingSink) runBatchingWorker(ctx context.Context) {
 
 		req, send, err := ioEmitter.AdmitRequest(ctx, batchBuffer)
 		if errors.Is(err, ErrNotEnoughQuota) {
+			waitStart := timeutil.Now()
+
 			// Quota can only be freed by consuming a result.
 			select {
 			case <-ctx.Done():
@@ -428,8 +430,12 @@ func (s *batchingSink) runBatchingWorker(ctx context.Context) {
 			} else if err != nil {
 				return err
 			}
+
+			s.metrics.recordSinkBackpressure(timeutil.Since(waitStart))
 		} else if err != nil {
 			return err
+		} else {
+			s.metrics.recordSinkBackpressure(0)
 		}
 
 		// The request was admitted, it must be sent. There are no concurrent requests being sent which

pkg/ccl/changefeedccl/changefeed_test.go

@@ -11424,6 +11424,59 @@ func TestParallelIOMetrics(t *testing.T) {
 	cdcTest(t, testFn, feedTestForceSink("pubsub"))
 }
 
+// TestSinkBackpressureMetric tests that the sink backpressure metric is recorded
+// when quota limits are hit.
+func TestSinkBackpressureMetric(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
+		registry := s.Server.JobRegistry().(*jobs.Registry)
+		metrics := registry.MetricsStruct().Changefeed.(*Metrics).AggMetrics
+
+		db := sqlutils.MakeSQLRunner(s.DB)
+		db.Exec(t, `SET CLUSTER SETTING changefeed.sink_io_workers = 1`)
+		db.Exec(t, `CREATE TABLE foo (a INT PRIMARY KEY)`)
+
+		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+		defer cancel()
+
+		g := ctxgroup.WithContext(ctx)
+		done := make(chan struct{})
+		g.Go(func() error {
+			for {
+				select {
+				case <-ctx.Done():
+					return nil
+				case <-done:
+					return nil
+				default:
+					_, err := s.DB.Exec(`UPSERT INTO foo (a)  SELECT * FROM generate_series(1, 10)`)
+					if err != nil {
+						return err
+					}
+				}
+			}
+		})
+
+		foo, err := f.Feed(`CREATE CHANGEFEED FOR TABLE foo WITH pubsub_sink_config='{"Flush": {"Frequency": "100ms"}}'`)
+		require.NoError(t, err)
+
+		testutils.SucceedsSoon(t, func() error {
+			numSamples, sum := metrics.SinkBackpressureNanos.WindowedSnapshot().Total()
+			if numSamples <= 0 && sum <= 0.0 {
+				return errors.Newf("waiting for backpressure nanos: %d %f", numSamples, sum)
+			}
+			return nil
+		})
+
+		close(done)
+		require.NoError(t, g.Wait())
+		require.NoError(t, foo.Close())
+	}
+	cdcTest(t, testFn, feedTestForceSink("pubsub"))
+}
+
 type changefeedLogSpy struct {
 	syncutil.Mutex
 	logs []string

pkg/ccl/changefeedccl/metrics.go

@@ -40,6 +40,7 @@ const (
 	changefeedIOQueueMaxLatency        = 5 * time.Minute
 	admitLatencyMaxValue               = 1 * time.Minute
 	commitLatencyMaxValue              = 10 * time.Minute
+	backpressureMaxValue               = 10 * time.Minute
 	kafkaThrottlingTimeMaxValue        = 5 * time.Minute
 )
 
@@ -68,6 +69,7 @@ type AggMetrics struct {
 	ParallelIOResultQueueNanos  *aggmetric.AggHistogram
 	ParallelIOInFlightKeys      *aggmetric.AggGauge
 	SinkIOInflight              *aggmetric.AggGauge
+	SinkBackpressureNanos       *aggmetric.AggHistogram
 	CommitLatency               *aggmetric.AggHistogram
 	BackfillCount               *aggmetric.AggGauge
 	BackfillPendingRanges       *aggmetric.AggGauge
@@ -123,6 +125,7 @@ type metricsRecorder interface {
 	recordSizeBasedFlush()
 	newParallelIOMetricsRecorder() parallelIOMetricsRecorder
 	recordSinkIOInflightChange(int64)
+	recordSinkBackpressure(time.Duration)
 	makeCloudstorageFileAllocCallback() func(delta int64)
 	getKafkaThrottlingMetrics(*cluster.Settings) metrics.Histogram
 	netMetrics() *cidr.NetMetrics
@@ -153,6 +156,7 @@ type sliMetrics struct {
 	ParallelIOResultQueueNanos  *aggmetric.Histogram
 	ParallelIOInFlightKeys      *aggmetric.Gauge
 	SinkIOInflight              *aggmetric.Gauge
+	SinkBackpressureNanos       *aggmetric.Histogram
 	CommitLatency               *aggmetric.Histogram
 	ErrorRetries                *aggmetric.Counter
 	AdmitLatency                *aggmetric.Histogram
@@ -599,6 +603,14 @@ func (m *sliMetrics) recordSinkIOInflightChange(delta int64) {
 	m.SinkIOInflight.Inc(delta)
 }
 
+func (m *sliMetrics) recordSinkBackpressure(duration time.Duration) {
+	if m == nil {
+		return
+	}
+
+	m.SinkBackpressureNanos.RecordValue(duration.Nanoseconds())
+}
+
 type wrappingCostController struct {
 	ctx      context.Context
 	inner    metricsRecorder
@@ -677,6 +689,10 @@ func (w *wrappingCostController) recordSinkIOInflightChange(delta int64) {
 	w.inner.recordSinkIOInflightChange(delta)
 }
 
+func (w *wrappingCostController) recordSinkBackpressure(duration time.Duration) {
+	w.inner.recordSinkBackpressure(duration)
+}
+
 func (w *wrappingCostController) newParallelIOMetricsRecorder() parallelIOMetricsRecorder {
 	return w.inner.newParallelIOMetricsRecorder()
 }
@@ -959,6 +975,14 @@ func newAggregateMetrics(histogramWindow time.Duration, lookup *cidr.Lookup) *Ag
 		Measurement: "Messages",
 		Unit:        metric.Unit_COUNT,
 	}
+	metaChangefeedSinkBackpressureNanos := metric.Metadata{
+		Name: "changefeed.sink_backpressure_nanos",
+		Help: "Time spent waiting for quota when emitting to the sink (back-pressure). " +
+			"Only populated for sinks using the batching_sink wrapper. As of writing, " +
+			"this includes Kafka (v2), Pub/Sub (v2), and Webhook (v2).",
+		Measurement: "Nanoseconds",
+		Unit:        metric.Unit_NANOSECONDS,
+	}
 	metaAggregatorProgress := metric.Metadata{
 		Name:        "changefeed.aggregator_progress",
 		Help:        "The earliest timestamp up to which any aggregator is guaranteed to have emitted all values for",
@@ -1072,6 +1096,13 @@ func newAggregateMetrics(histogramWindow time.Duration, lookup *cidr.Lookup) *Ag
 		}),
 		ParallelIOInFlightKeys: b.Gauge(metaChangefeedParallelIOInFlightKeys),
 		SinkIOInflight:         b.Gauge(metaChangefeedSinkIOInflight),
+		SinkBackpressureNanos: b.Histogram(metric.HistogramOptions{
+			Metadata:     metaChangefeedSinkBackpressureNanos,
+			Duration:     histogramWindow,
+			MaxVal:       backpressureMaxValue.Nanoseconds(),
+			SigFigs:      2,
+			BucketConfig: metric.ChangefeedBatchLatencyBuckets,
+		}),
 		BatchHistNanos: b.Histogram(metric.HistogramOptions{
 			Metadata:     metaChangefeedBatchHistNanos,
 			Duration:     histogramWindow,
@@ -1178,6 +1209,7 @@ func (a *AggMetrics) getOrCreateScope(scope string) (*sliMetrics, error) {
 		ParallelIOResultQueueNanos:  a.ParallelIOResultQueueNanos.AddChild(scope),
 		ParallelIOInFlightKeys:      a.ParallelIOInFlightKeys.AddChild(scope),
 		SinkIOInflight:              a.SinkIOInflight.AddChild(scope),
+		SinkBackpressureNanos:       a.SinkBackpressureNanos.AddChild(scope),
 		CommitLatency:               a.CommitLatency.AddChild(scope),
 		ErrorRetries:                a.ErrorRetries.AddChild(scope),
 		AdmitLatency:                a.AdmitLatency.AddChild(scope),

pkg/cmd/roachtest/tests/cdc.go

@@ -1679,6 +1679,60 @@ highwaterLoop:
 	}
 }
 
+// runCDCWebhookBackpressureMetrics tests that the sink backpressure metric
+// gets populated when using a slow webhook sink.
+func runCDCWebhookBackpressureMetrics(ctx context.Context, t test.Test, c cluster.Cluster) {
+	ct := newCDCTester(ctx, t, c)
+	defer ct.Close()
+
+	ips, err := c.ExternalIP(ctx, t.L(), c.WorkloadNode())
+	if err != nil {
+		t.Fatal(err)
+	}
+	sinkURL := fmt.Sprintf("https://%s:%d", ips[0], debug.WebhookServerPort)
+	sink := &http.Client{Transport: &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}}}
+
+	m := c.NewDeprecatedMonitor(ctx, c.All())
+
+	m.Go(func(ctx context.Context) error {
+		t.L().Printf("starting slow webhook server at %s...", sinkURL)
+		// Use webhook-server-slow with delays to create backpressure
+		return c.RunE(ctx, option.WithNodes(c.WorkloadNode()),
+			"./cockroach workload debug webhook-server-slow 100 5000 5000 5000 5000 5000")
+	})
+	defer func() {
+		_, err := sink.Get(sinkURL + "/exit")
+		t.L().Printf("exiting webhook sink status: %v", err)
+	}()
+
+	db := c.Conn(ctx, t.L(), 1)
+	defer db.Close()
+
+	ct.runTPCCWorkload(tpccArgs{warehouses: 10, duration: "1m"})
+
+	ct.newChangefeed(feedArgs{
+		sinkType:        webhookSink,
+		targets:         allTpccTargets,
+		opts:            map[string]string{"updated": "", "min_checkpoint_frequency": "'1s'", "webhook_sink_config": `'{"Flush": {"Messages": 10, "Frequency": "100ms"}}'`},
+		sinkURIOverride: fmt.Sprintf("webhook-%s/?insecure_tls_skip_verify=true", sinkURL),
+	})
+
+	// Wait a bit for metrics to be recorded
+	time.Sleep(30 * time.Second)
+
+	t.L().Printf("verifying backpressure metric...")
+
+	ct.verifyMetrics(ctx, func(metrics map[string]*prompb.MetricFamily) (ok bool) {
+		for _, m := range metrics {
+			if m.GetName() == "changefeed_sink_backpressure_nanos" {
+				count, sum := m.GetMetric()[0].GetHistogram().GetSampleCount(), m.GetMetric()[0].GetHistogram().GetSampleSum()
+				return count > 0 && sum > 0
+			}
+		}
+		return false
+	})
+}
+
 func runMessageTooLarge(ctx context.Context, t test.Test, c cluster.Cluster) {
 	ct := newCDCTester(ctx, t, c)
 	db := ct.DB()
@@ -2709,6 +2763,15 @@ func registerCDC(r registry.Registry) {
 			ct.waitForWorkload()
 		},
 	})
+	r.Add(registry.TestSpec{
+		Name:             "cdc/webhook-sink-backpressure-metrics",
+		Owner:            `cdc`,
+		Cluster:          r.MakeClusterSpec(4, spec.CPU(4), spec.WorkloadNode()),
+		Leases:           registry.MetamorphicLeases,
+		CompatibleClouds: registry.OnlyGCE,
+		Suites:           registry.Suites(registry.Nightly),
+		Run:              runCDCWebhookBackpressureMetrics,
+	})
 	r.Add(registry.TestSpec{
 		Name:             "cdc/message-too-large-error",
 		Owner:            registry.OwnerCDC,