changefeedccl: add pts management performance benchmark

This change adds benchmarking for the behavior of the
protected timestamp system. The benchmark is a roachtest
that creates 500 instances of the bank workload and starts
a changefeed watching their 500 tables for 20 minutes.

We will use this same test to measure the performance of the
pts system for multi-table feeds once they create a distinct
pts record for each watched table.

Epic: CRDB-1421
Fixes: #147781
Release note: None

Author: aerfrei

pkg/cmd/roachtest/tests/cdc.go

@@ -1731,6 +1731,120 @@ func runMessageTooLarge(ctx context.Context, t test.Test, c cluster.Cluster) {
 	require.Regexp(t, `mvcc=[\d\.]+,\d+`, logStr, "log should include mvcc")
 }
 
+type multiTablePTSBenchmarkParams struct {
+	numTables int
+	numRows   int
+	duration  string
+}
+
+// runCDCMultiTablePTSBenchmark is a benchmark for changefeeds with multiple tables,
+// focusing on the performance of the PTS system. It starts a bank workload on every
+// table it creates and then runs a single changefeed that targets all of these bank tables.
+// Each of those workloads (there will be one per table) will run for the duration specified
+// in the params and have the number of rows specified in the params.
+func runCDCMultiTablePTSBenchmark(
+	ctx context.Context, t test.Test, c cluster.Cluster, params multiTablePTSBenchmarkParams,
+) {
+	ct := newCDCTester(ctx, t, c)
+	defer ct.Close()
+
+	startOpts := option.DefaultStartOpts()
+	startOpts.RoachprodOpts.ExtraArgs = append(startOpts.RoachprodOpts.ExtraArgs,
+		"--vmodule=changefeed=2,changefeed_processors=2,protected_timestamps=2",
+	)
+
+	db := ct.DB()
+	if err := configureDBForMultiTablePTSBenchmark(db); err != nil {
+		t.Fatalf("failed to set cluster settings: %v", err)
+	}
+
+	initCmd := fmt.Sprintf("./cockroach workload init bank --rows=%d --num-tables=%d {pgurl%s}",
+		params.numRows, params.numTables, ct.crdbNodes.RandNode())
+	if err := c.RunE(ctx, option.WithNodes(ct.workloadNode), initCmd); err != nil {
+		t.Fatalf("failed to initialize bank tables: %v", err)
+	}
+
+	ct.workloadWg.Add(1)
+	ct.mon.Go(func(ctx context.Context) error {
+		defer ct.workloadWg.Done()
+		workloadCmd := fmt.Sprintf("./cockroach workload run bank --rows=%d --duration=%s --num-tables=%d {pgurl%s}",
+			params.numRows, params.duration, params.numTables, ct.crdbNodes)
+		return c.RunE(ctx, option.WithNodes(ct.workloadNode), workloadCmd)
+	})
+
+	// We generate and run the changefeed, which requires rangefeeds to be enabled.
+	if _, err := db.Exec("SET CLUSTER SETTING kv.rangefeed.enabled = true"); err != nil {
+		t.Fatalf("failed to enable rangefeeds: %v", err)
+	}
+
+	targetNames := make([]string, 0, params.numTables)
+	for i := range params.numTables {
+		targetNames = append(targetNames, fmt.Sprintf("bank.bank_%d", i))
+	}
+
+	feed := ct.newChangefeed(feedArgs{
+		sinkType: nullSink,
+		targets:  targetNames,
+		opts: map[string]string{
+			"format":                   "'json'",
+			"resolved":                 "'1s'",
+			"full_table_name":          "",
+			"min_checkpoint_frequency": "'1s'",
+			"initial_scan":             "'no'",
+		},
+	})
+
+	t.Status("multi-table PTS benchmark running with jobId ", feed.jobID)
+
+	ct.waitForWorkload()
+
+	t.Status("workload finished, verifying metrics")
+
+	// These metrics are in nanoseconds, so we are asserting that both
+	// of these latency metrics are less than 10 milliseconds.
+	ct.verifyMetrics(ctx, verifyMetricsUnderThreshold([]string{
+		"changefeed_stage_pts_manage_latency",
+		"changefeed_stage_pts_create_latency",
+	}, float64(10*time.Millisecond)))
+
+	t.Status("multi-table PTS benchmark finished")
+}
+
+func configureDBForMultiTablePTSBenchmark(db *gosql.DB) error {
+	// This is used to trigger frequent garbage collection and
+	// protected timestamp updates.
+	if _, err := db.Exec("ALTER DATABASE defaultdb CONFIGURE ZONE USING gc.ttlseconds = 1"); err != nil {
+		return err
+	}
+	if _, err := db.Exec("SET CLUSTER SETTING kv.closed_timestamp.target_duration = '100ms'"); err != nil {
+		return err
+	}
+	if _, err := db.Exec("SET CLUSTER SETTING kv.protectedts.poll_interval = '10ms'"); err != nil {
+		return err
+	}
+
+	// This is used to trigger frequent protected timestamp updates.
+	if _, err := db.Exec("SET CLUSTER SETTING changefeed.protect_timestamp_interval = '10ms'"); err != nil {
+		return err
+	}
+	if _, err := db.Exec("SET CLUSTER SETTING changefeed.protect_timestamp.lag = '1ms'"); err != nil {
+		return err
+	}
+
+	// These settings are used to trigger frequent checkpoints since protected timestamp
+	// management happens on checkpointing.
+	if _, err := db.Exec("SET CLUSTER SETTING changefeed.span_checkpoint.interval = '1s'"); err != nil {
+		return err
+	}
+	if _, err := db.Exec("SET CLUSTER SETTING changefeed.frontier_highwater_lag_checkpoint_threshold = '100ms'"); err != nil {
+		return err
+	}
+	if _, err := db.Exec("SET CLUSTER SETTING changefeed.frontier_checkpoint_frequency = '1s'"); err != nil {
+		return err
+	}
+	return nil
+}
+
 func registerCDC(r registry.Registry) {
 	r.Add(registry.TestSpec{
 		Name:      "cdc/initial-scan-only",
@@ -2719,6 +2833,23 @@ func registerCDC(r registry.Registry) {
 		CompatibleClouds: registry.AllExceptIBM,
 		Run:              runMessageTooLarge,
 	})
+	r.Add(registry.TestSpec{
+		Name:             "cdc/multi-table-pts-benchmark",
+		Owner:            registry.OwnerCDC,
+		Benchmark:        true,
+		Cluster:          r.MakeClusterSpec(4, spec.CPU(16), spec.WorkloadNode()),
+		CompatibleClouds: registry.AllClouds,
+		Suites:           registry.Suites(registry.Nightly),
+		Timeout:          1 * time.Hour,
+		Run: func(ctx context.Context, t test.Test, c cluster.Cluster) {
+			params := multiTablePTSBenchmarkParams{
+				numTables: 500,
+				numRows:   10_000,
+				duration:  "20m",
+			}
+			runCDCMultiTablePTSBenchmark(ctx, t, c, params)
+		},
+	})
 }
 
 const (
@@ -4403,3 +4534,38 @@ func verifyMetricsNonZero(names ...string) func(metrics map[string]*prompb.Metri
 		return false
 	}
 }
+
+func verifyMetricsUnderThreshold(
+	names []string, threshold float64,
+) func(metrics map[string]*prompb.MetricFamily) (ok bool) {
+	namesMap := make(map[string]struct{}, len(names))
+	for _, name := range names {
+		namesMap[name] = struct{}{}
+	}
+
+	return func(metrics map[string]*prompb.MetricFamily) (ok bool) {
+		found := map[string]struct{}{}
+
+		for name, fam := range metrics {
+			if _, ok := namesMap[name]; !ok {
+				continue
+			}
+
+			for _, m := range fam.Metric {
+				if m.Histogram.GetSampleCount() == 0 {
+					continue
+				}
+
+				observedValue := m.Histogram.GetSampleSum() / float64(m.Histogram.GetSampleCount())
+				if observedValue < threshold {
+					found[name] = struct{}{}
+				}
+			}
+
+			if len(found) == len(names) {
+				return true
+			}
+		}
+		return false
+	}
+}