Merge cockroachdb#39012

39012: roachtest: enable sysbench test suite r=tbg a=nvanbenschoten

Informs cockroachdb#32738.

Now that we know more about cockroachdb#32738, we know that we can safely avoid the segfault if we talk directly to a cockroach node (i.e bypass haproxy) during the preparation phase of sysbench. This commit re-enables the sysbench test suite after doing so.

The commit also passes the `--auto_inc=false` flag to sysbench. This is critical, because without this the test will use a `SERIAL` column for the primary key of each table. It then expects that the `SERIAL` column will create rows with values [1, table_size], which is not true in CRDB by default. Before this fix, I was noticing incredibly high UPDATE throughput, which was a result of never finding any real rows to update. We can use the `experimental_serial_normalization` variable to use a real SQL sequence to back the `SERIAL` column, but this slows down the import step by two orders of magnitude because updates to the sequence are not batched for multi-value INSERT statements.

One question I'd like to resolve during the review is whether we should tag this as a weekly test. We don't have the ability to parse its output to hook it up to roachperf (yet), so there's not a particularly strong reason to run it nightly.

Co-authored-by: Nathan VanBenschoten <[email protected]>

pkg/cmd/roachprod/install/cluster_synced.go

@@ -1412,16 +1412,25 @@ func (c *SyncedCluster) Get(src, dest string) {
 }
 
 func (c *SyncedCluster) pgurls(nodes []int) map[int]string {
+	hosts := c.pghosts(nodes)
+	m := make(map[int]string, len(hosts))
+	for node, host := range hosts {
+		m[node] = c.Impl.NodeURL(c, host, c.Impl.NodePort(c, node))
+	}
+	return m
+}
+
+func (c *SyncedCluster) pghosts(nodes []int) map[int]string {
 	ips := make([]string, len(nodes))
 	c.Parallel("", len(nodes), 0, func(i int) ([]byte, error) {
 		var err error
 		ips[i], err = c.GetInternalIP(nodes[i])
-		return nil, errors.Wrapf(err, "pgurls")
+		return nil, errors.Wrapf(err, "pghosts")
 	})
 
 	m := make(map[int]string, len(ips))
 	for i, ip := range ips {
-		m[nodes[i]] = c.Impl.NodeURL(c, ip, c.Impl.NodePort(c, nodes[i]))
+		m[nodes[i]] = ip
 	}
 	return m
 }

pkg/cmd/roachprod/install/expander.go

@@ -20,19 +20,21 @@ import (
 
 var parameterRe = regexp.MustCompile(`{[^}]*}`)
 var pgURLRe = regexp.MustCompile(`{pgurl(:[-,0-9]+)?}`)
+var pgHostRe = regexp.MustCompile(`{pghost(:[-,0-9]+)?}`)
 var pgPortRe = regexp.MustCompile(`{pgport(:[-,0-9]+)?}`)
 var uiPortRe = regexp.MustCompile(`{uiport(:[-,0-9]+)?}`)
 var storeDirRe = regexp.MustCompile(`{store-dir}`)
 var logDirRe = regexp.MustCompile(`{log-dir}`)
 var certsDirRe = regexp.MustCompile(`{certs-dir}`)
 
 // expander expands a string which contains templated parameters for cluster
-// attributes like pgurl, pgport, uiport, store-dir, and log-dir with the
-// corresponding values.
+// attributes like pgurl, pghost, pgport, uiport, store-dir, and log-dir with
+// the corresponding values.
 type expander struct {
 	node int
 
 	pgURLs  map[int]string
+	pgHosts map[int]string
 	pgPorts map[int]string
 	uiPorts map[int]string
 }
@@ -49,6 +51,7 @@ func (e *expander) expand(c *SyncedCluster, arg string) (string, error) {
 		}
 		expanders := []expanderFunc{
 			e.maybeExpandPgURL,
+			e.maybeExpandPgHost,
 			e.maybeExpandPgPort,
 			e.maybeExpandUIPort,
 			e.maybeExpandStoreDir,
@@ -117,6 +120,21 @@ func (e *expander) maybeExpandPgURL(c *SyncedCluster, s string) (string, bool, e
 	return s, err == nil, err
 }
 
+// maybeExpandPgHost is an expanderFunc for {pghost:<nodeSpec>}
+func (e *expander) maybeExpandPgHost(c *SyncedCluster, s string) (string, bool, error) {
+	m := pgHostRe.FindStringSubmatch(s)
+	if m == nil {
+		return s, false, nil
+	}
+
+	if e.pgHosts == nil {
+		e.pgHosts = c.pghosts(allNodes(len(c.VMs)))
+	}
+
+	s, err := e.maybeExpandMap(c, e.pgHosts, m[1])
+	return s, err == nil, err
+}
+
 // maybeExpandPgURL is an expanderFunc for {pgport:<nodeSpec>}
 func (e *expander) maybeExpandPgPort(c *SyncedCluster, s string) (string, bool, error) {
 	m := pgPortRe.FindStringSubmatch(s)

pkg/cmd/roachtest/sysbench.go

@@ -22,11 +22,12 @@ const (
 	oltpDelete sysbenchWorkload = iota
 	oltpInsert
 	oltpPointSelect
+	oltpUpdateIndex
 	oltpUpdateNonIndex
+	oltpReadOnly
 
 	// TODO(nvanbenschoten): transactional workloads are not supported
 	// because sysbench does not contain client-side retry loops.
-	// oltpReadOnly
 	// oltpReadWrite
 	// oltpWriteOnly
 
@@ -37,8 +38,9 @@ var sysbenchWorkloadName = map[sysbenchWorkload]string{
 	oltpDelete:         "oltp_delete",
 	oltpInsert:         "oltp_insert",
 	oltpPointSelect:    "oltp_point_select",
+	oltpUpdateIndex:    "oltp_update_index",
 	oltpUpdateNonIndex: "oltp_update_non_index",
-	// oltpReadOnly:  "oltp_read_only",
+	oltpReadOnly:       "oltp_read_only",
 	// oltpReadWrite: "oltp_read_write",
 	// oltpWriteOnly: "oltp_write_only",
 }
@@ -55,10 +57,14 @@ type sysbenchOptions struct {
 	rowsPerTable int
 }
 
-func (o *sysbenchOptions) cmd() string {
+func (o *sysbenchOptions) cmd(haproxy bool) string {
+	pghost := "{pghost:1}"
+	if haproxy {
+		pghost = "127.0.0.1"
+	}
 	return fmt.Sprintf(`sysbench \
 		--db-driver=pgsql \
-		--pgsql-host=127.0.0.1 \
+		--pgsql-host=%s \
 		--pgsql-port=26257 \
 		--pgsql-user=root \
 		--pgsql-password= \
@@ -68,7 +74,9 @@ func (o *sysbenchOptions) cmd() string {
 		--threads=%d \
 		--tables=%d \
 		--table_size=%d \
+		--auto_inc=false \
 		%s`,
+		pghost,
 		int(o.duration.Seconds()),
 		o.concurrency,
 		o.tables,
@@ -102,10 +110,10 @@ func runSysbench(ctx context.Context, t *test, c *cluster, opts sysbenchOptions)
 	m.Go(func(ctx context.Context) error {
 		t.Status("preparing workload")
 		c.Run(ctx, c.Node(1), `./cockroach sql --insecure -e "CREATE DATABASE sysbench"`)
-		c.Run(ctx, loadNode, opts.cmd()+" prepare")
+		c.Run(ctx, loadNode, opts.cmd(false /* haproxy */)+" prepare")
 
 		t.Status("running workload")
-		c.Run(ctx, loadNode, opts.cmd()+" run")
+		c.Run(ctx, loadNode, opts.cmd(true /* haproxy */)+" run")
 		return nil
 	})
 	m.Wait()
@@ -114,18 +122,18 @@ func runSysbench(ctx context.Context, t *test, c *cluster, opts sysbenchOptions)
 func registerSysbench(r *testRegistry) {
 	for w := sysbenchWorkload(0); w < numSysbenchWorkloads; w++ {
 		const n = 3
-		const cpus = 16
+		const cpus = 32
+		const conc = 4 * cpus
 		opts := sysbenchOptions{
 			workload:     w,
 			duration:     10 * time.Minute,
-			concurrency:  8 * cpus,
-			tables:       4,
-			rowsPerTable: 1000000,
+			concurrency:  conc,
+			tables:       10,
+			rowsPerTable: 10000000,
 		}
 
 		r.Add(testSpec{
-			Skip:    "https://github.com/cockroachdb/cockroach/issues/32738",
-			Name:    fmt.Sprintf("sysbench/%s/nodes=%d", w, n),
+			Name:    fmt.Sprintf("sysbench/%s/nodes=%d/cpu=%d/conc=%d", w, n, cpus, conc),
 			Cluster: makeClusterSpec(n+1, cpu(cpus)),
 			Run: func(ctx context.Context, t *test, c *cluster) {
 				runSysbench(ctx, t, c, opts)