Merge #150666 #150861 #150863

150666: changefeedccl: add sink backpressure time metric r=aerfrei a=asg0451

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.


150861: storage: use AssertionFailedf for panics r=RaduBerinde a=jbowens

Update a few panics that threw a simple string to throw an error constructed through AssertionFailedf. Where applicable, this error is propagated through a return value instead.

This is motivated by #150058, an instance where the panic string gets redacted, requiring a bit more work to identify the origin of the panic (through mapping the line number on the appropriate commit SHA).

Epic: none
Release note: none

150863: CLAUDE: use verbose output in filtered tests r=rickystewart a=dhartunian

I observed a `./dev test pkg -f=x` scenario where no test matched, and false success was declared.

Release note: None

Co-authored-by: Miles Frankel <[email protected]>
Co-authored-by: Jackson Owens <[email protected]>
Co-authored-by: David Hartunian <[email protected]>

Author: 4 people

CLAUDE.md

@@ -23,15 +23,19 @@ CockroachDB is a distributed SQL database written in Go, built with Bazel and ma
 
 **Testing:**
 ```bash
-./dev test pkg/sql                    # Run unit tests for SQL package
-./dev test pkg/sql -f=TestParse      # Run specific test pattern
+./dev test pkg/sql                   # Run unit tests for SQL package
+./dev test pkg/sql -f=TestParse -v   # Run specific test pattern.
 ./dev test pkg/sql --race            # Run with race detection
 ./dev test pkg/sql --stress          # Run repeatedly until failure
 ./dev testlogic                      # Run all SQL logic tests
 ./dev testlogic ccl                  # Run enterprise logic tests
 ./dev testlogic --files='prepare|fk' # Run specific test files
 ```
 
+Note that when filtering tests via `-f` to include the `-v` flag which
+will warn you in the output if your filter didn't match anything. Look
+for `testing: warning: no tests to run` in the output.
+
 **Code Generation and Linting:**
 ```bash
 ./dev generate            # Generate all code (protobuf, parsers, etc.)
@@ -129,4 +133,4 @@ Always run `./dev generate` after modifying `.proto` files, SQL grammar, or opti
   - Require the user to specify an epic number (or None) which should be included at the bottom of the commit record following "Epic:".
   - Prefix the subject line with the package in which the bulk of the changes occur.
   - For multi-commit PRs, summarize each commit in the PR record.
-  - Do not include a test plan unless explicitly asked by the user.
+  - Do not include a test plan unless explicitly asked by the user.

docs/generated/metrics/metrics.yaml

@@ -1530,6 +1530,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,

pkg/storage/backup_compaction_iterator.go

@@ -116,7 +116,7 @@ func (f *BackupCompactionIterator) ValueLen() int {
 func (f *BackupCompactionIterator) HasPointAndRange() (bool, bool) {
 	hasPoint, hasRange := f.iter.HasPointAndRange()
 	if hasRange {
-		panic("unexpected range tombstone")
+		panic(errors.AssertionFailedf("unexpected range tombstone"))
 	}
 	return hasPoint, hasRange
 }

pkg/storage/batch.go

@@ -128,7 +128,7 @@ func (r *BatchReader) EngineKey() (EngineKey, error) {
 func (r *BatchReader) Value() []byte {
 	switch r.kind {
 	case pebble.InternalKeyKindDelete, pebble.InternalKeyKindSingleDelete:
-		panic("cannot call Value on a deletion entry")
+		panic(errors.AssertionFailedf("cannot call Value on a deletion entry"))
 	default:
 		return r.value
 	}