Merge #155518

155518: workload: add changefeed-max-rate, changefeed-start-delay, and changefeed-resolved-target r=wenyihu6 a=stevendanna

```
--changefeed                            Optionally run a changefeed over the tables
--changefeed-max-rate float             Maximum frequency of changefeed ingestion. If 0, no limit.
--changefeed-resolved-target duration   The target frequency of resolved messages. O to disable resolved reporting and accept server defaults. (default 5s)
--changefeed-start-delay duration       How long to wait before starting the changefeed
```

`--changefeed-max-rate` can be used in conjunction with `--max-rate` to
simulate an under-provisioned or correctly provisioned sink for the
given workload.

`---changefeed-start-delay` can be used to force a catch-up scan.

Note that I've also renamed --with-changefeed to --changefeed so that
all the changefeed options show up next to each other in the help.

The resolved timestamp reporting is a bit of a hack but it seems to work
for now.

Epic: none

Release note: None

Co-authored-by: Steven Danna <[email protected]>

Author: craig[bot]

pkg/workload/changefeeds/changefeeds.go

@@ -10,6 +10,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
@@ -25,6 +26,7 @@ func AddChangefeedToQueryLoad(
 	ctx context.Context,
 	gen workload.ConnFlagser,
 	dbName string,
+	resolvedTarget time.Duration,
 	urls []string,
 	reg *histogram.Registry,
 	ql *workload.QueryLoad,
@@ -33,6 +35,14 @@ func AddChangefeedToQueryLoad(
 		Method:              gen.ConnFlags().Method,
 		MaxTotalConnections: 2,
 	}
+	cfLatency := reg.GetHandle().Get("changefeed")
+	// TODO(ssd): This metric makes no sense has a histogram. The idea that every
+	// metric we might want to report is a histogram is baked in pretty deeply so
+	// I'll address that in a different PR (maybe)
+	var cfResolved *histogram.NamedHistogram
+	if resolvedTarget > 0 {
+		cfResolved = reg.GetHandle().Get("changefeed-resolved")
+	}
 
 	mcp, err := workload.NewMultiConnPool(ctx, cfg, urls...)
 	if err != nil {
@@ -51,8 +61,6 @@ func AddChangefeedToQueryLoad(
 		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")
@@ -68,6 +76,11 @@ func AddChangefeedToQueryLoad(
 		return err
 	}
 
+	var cursorStr string
+	if err := conn.QueryRow(ctx, "SELECT cluster_logical_timestamp()").Scan(&cursorStr); err != nil {
+		return err
+	}
+
 	tableNames := strings.Builder{}
 	for i, table := range gen.Tables() {
 		if i == 0 {
@@ -77,8 +90,23 @@ func AddChangefeedToQueryLoad(
 		}
 	}
 
-	stmt := fmt.Sprintf("EXPERIMENTAL CHANGEFEED FOR %s WITH updated, no_initial_scan, schema_change_policy=nobackfill",
-		tableNames.String())
+	opts := []string{
+		"updated",
+		"no_initial_scan",
+		"schema_change_policy=nobackfill",
+		"cursor=$1",
+	}
+	args := []any{
+		cursorStr,
+	}
+	if resolvedTarget > 0 {
+		opts = append(opts, []string{"resolved=$2", "min_checkpoint_frequency=$2"}...)
+		args = append(args, resolvedTarget.String())
+	}
+	stmt := fmt.Sprintf(
+		"CREATE CHANGEFEED FOR %s WITH %s",
+		tableNames.String(), strings.Join(opts, ","),
+	)
 	cfCtx, cancel := context.WithCancel(ctx)
 
 	var doneErr error
@@ -97,33 +125,57 @@ func AddChangefeedToQueryLoad(
 			return false
 		}
 		var err error
-		rows, err = conn.Query(cfCtx, stmt)
+		rows, err = conn.Query(cfCtx, stmt, args...)
 		return maybeMarkDone(err)
 	}
-	ql.WorkerFns = append(ql.WorkerFns, func(ctx context.Context) error {
+
+	var lastResolved hlc.Timestamp
+
+	ql.ChangefeedFns = append(ql.ChangefeedFns, 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"`
+				Updated  string `json:"updated"`
+				Resolved string `json:"resolved"`
 			}
 			var v updatedJSON
 			if maybeMarkDone(json.Unmarshal(values[2].([]byte), &v)) {
 				return doneErr
 			}
-			updated, err := hlc.ParseHLC(v.Updated)
-			if maybeMarkDone(err) {
-				return doneErr
+			if v.Updated != "" {
+				updated, err := hlc.ParseHLC(v.Updated)
+				if maybeMarkDone(err) {
+					return doneErr
+				}
+				cfLatency.Record(timeutil.Since(updated.GoTime()))
+			} else if v.Resolved != "" {
+				resolved, err := hlc.ParseHLC(v.Resolved)
+				if maybeMarkDone(err) {
+					return doneErr
+				}
+				if resolved.Less(lastResolved) {
+					return errors.Errorf("resolved timestamp %s is less than last resolved timestamp %s", resolved, lastResolved)
+				}
+				lastResolved = resolved
+			} else {
+				return errors.Errorf("failed to parse CHANGEFEED event: %s", values[2])
+			}
+			// Resolved timestamps arrived infrequently. Always record the time since
+			// our lastResolved so that we don't get long periods of 0 in the
+			// histogram.
+			if cfResolved != nil {
+				cfResolved.Record(timeutil.Since(lastResolved.GoTime()))
 			}
-			cfLatency.Record(timeutil.Since(updated.GoTime()))
 			return nil
 		}
 		if maybeMarkDone(rows.Err()) {

pkg/workload/cli/run.go

@@ -71,8 +71,6 @@ 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.
@@ -108,6 +106,18 @@ var secure = securityFlags.Bool("secure", false,
 var user = securityFlags.String("user", "root", "Specify a user to run the workload as")
 var password = securityFlags.String("password", "", "Optionally specify a password for the user")
 
+// Options relating to the optional changefeed.
+var (
+	changefeed = runFlags.Bool("changefeed", false,
+		"Optionally run a changefeed over the tables")
+	changefeedStartDelay = runFlags.Duration("changefeed-start-delay", 0*time.Second,
+		"How long to wait before starting the changefeed")
+	changefeedMaxRate = runFlags.Float64(
+		"changefeed-max-rate", 0, "Maximum frequency of changefeed ingestion. If 0, no limit.")
+	changefeedResolvedTarget = runFlags.Duration("changefeed-resolved-target", 5*time.Second,
+		"The target frequency of resolved messages. O to disable resolved reporting and accept server defaults.")
+)
+
 func init() {
 
 	_ = sharedFlags.MarkHidden("pprofport")
@@ -427,6 +437,10 @@ func runRun(gen workload.Generator, urls []string, dbName string) error {
 		// with allowed burst of 1 at the maximum allowed rate.
 		limiter = rate.NewLimiter(rate.Limit(*maxRate), 1)
 	}
+	var changefeedLimiter *rate.Limiter
+	if *changefeedMaxRate > 0 {
+		changefeedLimiter = rate.NewLimiter(rate.Limit(*changefeedMaxRate), 1)
+	}
 
 	maybeLogRandomSeed(ctx, gen)
 	o, ok := gen.(workload.Opser)
@@ -491,9 +505,9 @@ func runRun(gen workload.Generator, urls []string, dbName string) error {
 				return errors.Wrapf(err, "failed to initialize the load generator")
 			}
 
-			if *withChangefeed {
+			if *changefeed {
 				log.Dev.Infof(ctx, "adding changefeed to query load...")
-				err = changefeeds.AddChangefeedToQueryLoad(ctx, gen.(workload.ConnFlagser), dbName, urls, reg, &ops)
+				err = changefeeds.AddChangefeedToQueryLoad(ctx, gen.(workload.ConnFlagser), dbName, *changefeedResolvedTarget, urls, reg, &ops)
 				if err != nil && !*tolerateErrors {
 					return errors.Wrapf(err, "failed to initialize changefeed")
 				}
@@ -537,8 +551,17 @@ func runRun(gen workload.Generator, urls []string, dbName string) error {
 	workersCtx, cancelWorkers := context.WithCancel(ctx)
 	defer cancelWorkers()
 	var wg sync.WaitGroup
-	wg.Add(len(ops.WorkerFns))
+	wg.Add(len(ops.WorkerFns) + len(ops.ChangefeedFns))
 	go func() {
+		for _, workFn := range ops.ChangefeedFns {
+			go func(workFn func(context.Context) error) {
+				if *changefeedStartDelay > 0 {
+					time.Sleep(*changefeedStartDelay)
+				}
+				workerRun(workersCtx, errCh, &wg, changefeedLimiter, workFn)
+			}(workFn)
+		}
+
 		// If a ramp period was specified, start all the workers gradually
 		// with a new context.
 		var rampCtx context.Context
@@ -609,6 +632,13 @@ func runRun(gen workload.Generator, urls []string, dbName string) error {
 						log.Dev.Warningf(ctx, "histogram: %v", err)
 					}
 				}
+				// TODO(ssd): Ugly hack. Until we support something other than
+				// histograms here, for this particular metric, if we reset the
+				// histogram. We don't reset the Cumulative histogram which lets us see
+				// pMax.
+				if t.Name == "changefeed-resolved" {
+					t.Hist.Reset()
+				}
 			})
 
 		// Once the load generator is fully ramped up, we reset the histogram

pkg/workload/workload.go

@@ -419,7 +419,8 @@ func (l requiresCCLBinaryDataLoader) InitialDataLoad(
 type QueryLoad struct {
 	// WorkerFns is one function per worker. It is to be called once per unit of
 	// work to be done.
-	WorkerFns []func(context.Context) error
+	WorkerFns     []func(context.Context) error
+	ChangefeedFns []func(context.Context) error
 
 	// Close, if set, is called before the process exits, giving workloads a
 	// chance to print some information or perform cleanup.