Merge #155516

155516: workload: add mechanism to run changefeed alongside workloads r=wenyihu6 a=stevendanna

This is a manual foward-port of #73034 with a few small modifications from me.

This allows a user to run a changefeed with any workload which listens for writes to all tables. It adds a `--with-changefeed` flag to `workload <name> run` commands and then outputs a time series which indicates the number of rows and the difference in time between when the client sees the row and its commit timestamp.

Note that using the commit timestamp may be controversial. That timestamp will generally include all of the client latency. That's got some nice properties. If the latency added no additional latency, then the duration distribution would be exactly that of the client's writes.

Example:

```bash
./bin/workload run kv  --concurrency 10 --with-changefeed=true
```

```
_elapsed___errors__ops/sec(inst)___ops/sec(cum)__p50(ms)__p95(ms)__p99(ms)_pMax(ms)
   31.0s        0        15544.3        12910.1   3221.2   5905.6   6442.5   6710.9 changefeed
   31.0s        0         7504.6         7405.3      1.2      2.0      9.4     17.8 write
   32.0s        0        14963.0        12974.2   2952.8   5905.6   6174.0   6710.9 changefeed
   32.0s        0         7408.5         7405.4      1.2      1.9      8.9     19.9 write
   33.0s        0        14197.2        13011.3   2818.6   6174.0   6442.5   6979.3 changefeed
   33.0s        0         7039.6         7394.3      1.2      2.2      7.6     37.7 write
   34.0s        0        15786.2        13092.9   2550.1   5637.1   6174.0   6442.5 changefeed
   34.0s        0         7162.6         7387.5      1.2      2.4      7.6     35.7 write
   35.0s        0        10789.0        13027.1   2818.6   6174.0   6442.5   6710.9 changefeed
   35.0s        0         3513.0         7276.8      1.2      2.5      7.6    352.3 write
   36.0s        0        16003.6        13109.7   3221.2   6174.0   7247.8   7516.2 changefeed
   36.0s        0         8260.3         7304.1      1.1      1.7      4.5     16.3 write
   37.0s        0        15547.7        13175.6   3087.0   5100.3   6174.0   6710.9 changefeed
   37.0s        0         8218.4         7328.8      1.0      1.5      5.8     18.9 write
   38.0s        0        14046.1        13198.5   2684.4   6710.9   6979.3   7516.2 changefeed
   38.0s        0         8062.1         7348.1      1.1      1.6      5.2     16.8 write
   39.0s        0        12096.6        13170.3   1677.7   1879.0   2080.4   2147.5 changefeed
   39.0s        0         8285.4         7372.1      1.1      1.5      4.7     17.8 write
   40.0s        0        11868.9        13137.8   1208.0   1409.3   1409.3   1409.3 changefeed
   40.0s        0         8661.9         7404.4      1.0      1.5      3.4     12.6 write
_elapsed___errors__ops/sec(inst)___ops/sec(cum)__p50(ms)__p95(ms)__p99(ms)_pMax(ms)
   41.0s        0         8675.6        13028.9    838.9   1006.6   1006.6   1006.6 changefeed
   41.0s        0         8846.6         7439.6      1.0      1.4      3.3     12.1 write
```

Release note (cli change): `cockroach workload <name> run` commands now offer a `--with-changefeed` flag to additionally run a changefeed watching for writes to the workload's tables.

Co-authored-by: Andrew Werner <[email protected]>

Author: craig[bot]

pkg/BUILD.bazel

@@ -2865,6 +2865,7 @@ GO_TARGETS = [
     "//pkg/workload/bank:bank",
     "//pkg/workload/bank:bank_test",
     "//pkg/workload/bulkingest:bulkingest",
+    "//pkg/workload/changefeeds:changefeeds",
     "//pkg/workload/cli:cli",
     "//pkg/workload/cli:cli_test",
     "//pkg/workload/conflict:conflict",

pkg/workload/changefeeds/BUILD.bazel

@@ -0,0 +1,16 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "changefeeds",
+    srcs = ["changefeeds.go"],
+    importpath = "github.com/cockroachdb/cockroach/pkg/workload/changefeeds",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/util/hlc",
+        "//pkg/util/timeutil",
+        "//pkg/workload",
+        "//pkg/workload/histogram",
+        "@com_github_cockroachdb_errors//:errors",
+        "@com_github_jackc_pgx_v5//:pgx",
+    ],
+)

pkg/workload/changefeeds/changefeeds.go

@@ -0,0 +1,154 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package changefeeds
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+
+	"github.com/cockroachdb/cockroach/pkg/util/hlc"
+	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
+	"github.com/cockroachdb/cockroach/pkg/workload"
+	"github.com/cockroachdb/cockroach/pkg/workload/histogram"
+	"github.com/cockroachdb/errors"
+	"github.com/jackc/pgx/v5"
+)
+
+// AddChangefeedToQueryLoad augments the passed QueryLoad to contain an extra
+// worker to run a changefeed over the tables of the generator.
+func AddChangefeedToQueryLoad(
+	ctx context.Context,
+	gen workload.ConnFlagser,
+	dbName string,
+	urls []string,
+	reg *histogram.Registry,
+	ql *workload.QueryLoad,
+) error {
+	cfg := workload.MultiConnPoolCfg{
+		Method:              gen.ConnFlags().Method,
+		MaxTotalConnections: 2,
+	}
+
+	mcp, err := workload.NewMultiConnPool(ctx, cfg, urls...)
+	if err != nil {
+		return err
+	}
+	conn, err := mcp.Get().Acquire(ctx)
+	if err != nil {
+		return err
+	}
+	if _, err := conn.Exec(ctx, fmt.Sprintf("USE %q", dbName)); err != nil {
+		return err
+	}
+
+	setAppName := fmt.Sprintf("SET application_name='%s_changefeed'", gen.Meta().Name)
+	if _, err := conn.Exec(ctx, setAppName); err != nil {
+		return err
+	}
+
+	cfLatency := reg.GetHandle().Get("changefeed")
+
+	var sessionID string
+	if err := conn.QueryRow(ctx, "SHOW session_id").Scan(&sessionID); err != nil {
+		return errors.Wrap(err, "getting session_id")
+	}
+
+	// Create a second connection to close the first connection by issuing a
+	// cancel request.
+	closeConn, err := mcp.Get().Acquire(ctx)
+	if err != nil {
+		return err
+	}
+	if _, err := closeConn.Exec(ctx, setAppName); err != nil {
+		return err
+	}
+
+	tableNames := strings.Builder{}
+	for i, table := range gen.Tables() {
+		if i == 0 {
+			fmt.Fprintf(&tableNames, "%q", table.Name)
+		} else {
+			fmt.Fprintf(&tableNames, ", %q", table.Name)
+		}
+	}
+
+	stmt := fmt.Sprintf("EXPERIMENTAL CHANGEFEED FOR %s WITH updated, no_initial_scan, schema_change_policy=nobackfill",
+		tableNames.String())
+	cfCtx, cancel := context.WithCancel(ctx)
+
+	var doneErr error
+	maybeMarkDone := func(err error) (done bool) {
+		if err == nil {
+			return false
+		}
+		cancel()
+		_ = conn.Conn().Close(ctx)
+		doneErr, conn = err, nil
+		return true
+	}
+	var rows pgx.Rows
+	maybeSetupRows := func() (done bool) {
+		if rows != nil {
+			return false
+		}
+		var err error
+		rows, err = conn.Query(cfCtx, stmt)
+		return maybeMarkDone(err)
+	}
+	ql.WorkerFns = append(ql.WorkerFns, func(ctx context.Context) error {
+		if doneErr != nil {
+			return doneErr
+		}
+		if maybeSetupRows() {
+			return doneErr
+		}
+		if rows.Next() {
+			values, err := rows.Values()
+			if maybeMarkDone(err) {
+				return doneErr
+			}
+			type updatedJSON struct {
+				Updated string `json:"updated"`
+			}
+			var v updatedJSON
+			if maybeMarkDone(json.Unmarshal(values[2].([]byte), &v)) {
+				return doneErr
+			}
+			updated, err := hlc.ParseHLC(v.Updated)
+			if maybeMarkDone(err) {
+				return doneErr
+			}
+			cfLatency.Record(timeutil.Since(updated.GoTime()))
+			return nil
+		}
+		if maybeMarkDone(rows.Err()) {
+			return doneErr
+		}
+		maybeMarkDone(errors.New("changefeed ended"))
+		return doneErr
+	})
+
+	prevClose := ql.Close
+	ql.Close = func(ctx context.Context) error {
+		cancel()
+		_, _ = closeConn.Exec(ctx, "CANCEL SESSION $1", sessionID)
+		if err := closeConn.Conn().Close(ctx); err != nil {
+			return err
+		}
+		if conn != nil {
+			if err := conn.Conn().Close(ctx); err != nil {
+				return err
+			}
+		}
+		if err := prevClose(ctx); err != nil {
+			return err
+		}
+		return nil
+	}
+	return nil
+}

pkg/workload/cli/BUILD.bazel

@@ -24,6 +24,7 @@ go_library(
         "//pkg/util/retry",
         "//pkg/util/timeutil",
         "//pkg/workload",
+        "//pkg/workload/changefeeds",
         "//pkg/workload/histogram",
         "//pkg/workload/histogram/exporter",
         "//pkg/workload/workloadsql",

pkg/workload/cli/run.go

@@ -29,6 +29,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/retry"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/cockroach/pkg/workload"
+	"github.com/cockroachdb/cockroach/pkg/workload/changefeeds"
 	"github.com/cockroachdb/cockroach/pkg/workload/histogram"
 	"github.com/cockroachdb/cockroach/pkg/workload/histogram/exporter"
 	"github.com/cockroachdb/cockroach/pkg/workload/workloadsql"
@@ -70,6 +71,8 @@ var prometheusPort = sharedFlags.Int(
 	2112,
 	"Port to expose prometheus metrics if the workload has a prometheus gatherer set.",
 )
+var withChangefeed = runFlags.Bool("with-changefeed", false,
+	"Optionally run a changefeed over the tables")
 
 // individualOperationReceiverAddr is an address to send latency
 // measurements to. By default it will not send anything.
@@ -484,13 +487,20 @@ func runRun(gen workload.Generator, urls []string, dbName string) error {
 				log.Dev.Warningf(ctx, "retrying after error while creating load: %v", err)
 			}
 			ops, err = o.Ops(ctx, urls, reg)
+			if err != nil && !*tolerateErrors {
+				return errors.Wrapf(err, "failed to initialize the load generator")
+			}
+
+			if *withChangefeed {
+				log.Dev.Infof(ctx, "adding changefeed to query load...")
+				err = changefeeds.AddChangefeedToQueryLoad(ctx, gen.(workload.ConnFlagser), dbName, urls, reg, &ops)
+				if err != nil && !*tolerateErrors {
+					return errors.Wrapf(err, "failed to initialize changefeed")
+				}
+			}
 			if err == nil {
 				return nil
 			}
-			err = errors.Wrapf(err, "failed to initialize the load generator")
-			if !*tolerateErrors {
-				return err
-			}
 		}
 		if ctx.Err() != nil {
 			// Don't retry endlessly. Note that this retry loop is not under the