Merge cockroachdb#132263 cockroachdb#132341

132263: lint: fix `TestForbiddenImports` r=rail,herkolategan a=rickystewart We need to set `Dir` in the `packages.Load()` call or else the whole thing doesn't work. It would be really nice if it threw an error instead of simply doing nothing, but it is what it is. To guard against this in the future I added an error if the list is empty. Fix the test then all existing violations, except for `raftlogger` and `rafttest` which will need some extra TLC by the owning team (see cockroachdb#132262) Closes cockroachdb#132258 Epic: none Release note: None 132341: roachtest: run perf tests with sql and kv mode r=jeffswenson a=msbutler Epic: none Release note: none Co-authored-by: Ricky Stewart <[email protected]> Co-authored-by: Michael Butler <[email protected]>
mw5h · Oct 10, 2024 · a69a2f3 · a69a2f3
3 parents 88600d3 + 4e662c4 + 904f0ed
commit a69a2f3
Show file tree

Hide file tree

Showing 21 changed files with 107 additions and 37 deletions.
diff --git a/pkg/ccl/crosscluster/logical/BUILD.bazel b/pkg/ccl/crosscluster/logical/BUILD.bazel
@@ -86,7 +86,6 @@ go_library(
         "//pkg/util/timeutil",
         "//pkg/util/tracing",
         "@com_github_cockroachdb_errors//:errors",
-        "@com_github_cockroachdb_errors//issuelink",
         "@com_github_cockroachdb_logtags//:logtags",
         "@com_github_cockroachdb_redact//:redact",
         "@com_github_lib_pq//oid",

diff --git a/pkg/ccl/crosscluster/logical/create_logical_replication_stmt.go b/pkg/ccl/crosscluster/logical/create_logical_replication_stmt.go
@@ -38,7 +38,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/errors"
-	"github.com/cockroachdb/errors/issuelink"
 )
 
 func init() {
@@ -94,7 +93,7 @@ func createLogicalReplicationStreamPlanHook(
 		}
 
 		if stmt.From.Database != "" {
-			return errors.UnimplementedErrorf(issuelink.IssueLink{}, "logical replication streams on databases are unsupported")
+			return errors.UnimplementedErrorf(errors.IssueLink{}, "logical replication streams on databases are unsupported")
 		}
 		if len(stmt.From.Tables) != len(stmt.Into.Tables) {
 			return pgerror.New(pgcode.InvalidParameterValue, "the same number of source and destination tables must be specified")

diff --git a/pkg/ccl/crosscluster/streamclient/BUILD.bazel b/pkg/ccl/crosscluster/streamclient/BUILD.bazel
@@ -38,7 +38,6 @@ go_library(
         "@com_github_golang_snappy//:snappy",
         "@com_github_jackc_pgconn//:pgconn",
         "@com_github_jackc_pgx_v4//:pgx",
-        "@com_github_pkg_errors//:errors",
     ],
 )
 

diff --git a/pkg/ccl/crosscluster/streamclient/client_helpers.go b/pkg/ccl/crosscluster/streamclient/client_helpers.go
@@ -11,9 +11,9 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/ccl/crosscluster"
 	"github.com/cockroachdb/cockroach/pkg/repstream/streampb"
 	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
+	"github.com/cockroachdb/errors"
 	"github.com/golang/snappy"
 	"github.com/jackc/pgx/v4"
-	"github.com/pkg/errors"
 )
 
 func subscribeInternal(

diff --git a/pkg/cmd/roachtest/tests/logical_data_replication.go b/pkg/cmd/roachtest/tests/logical_data_replication.go
@@ -71,7 +71,7 @@ type LDRWorkload struct {
 func registerLogicalDataReplicationTests(r registry.Registry) {
 	specs := []ldrTestSpec{
 		{
-			name: "ldr/kv0/workload=both/ingestion",
+			name: "ldr/kv0/workload=both/basic/immediate",
 			clusterSpec: multiClusterSpec{
 				leftNodes:  3,
 				rightNodes: 3,
@@ -82,10 +82,11 @@ func registerLogicalDataReplicationTests(r registry.Registry) {
 					spec.VolumeSize(100),
 				},
 			},
-			run: TestLDRBasic,
+			mode: ModeImmediate,
+			run:  TestLDRBasic,
 		},
 		{
-			name: "ldr/kv0/workload=both/update_heavy",
+			name: "ldr/kv0/workload=both/basic/validated",
 			clusterSpec: multiClusterSpec{
 				leftNodes:  3,
 				rightNodes: 3,
@@ -96,7 +97,38 @@ func registerLogicalDataReplicationTests(r registry.Registry) {
 					spec.VolumeSize(100),
 				},
 			},
-			run: TestLDRUpdateHeavy,
+			mode: ModeValidated,
+			run:  TestLDRBasic,
+		},
+		{
+			name: "ldr/kv0/workload=both/update_heavy/immediate",
+			clusterSpec: multiClusterSpec{
+				leftNodes:  3,
+				rightNodes: 3,
+				clusterOpts: []spec.Option{
+					spec.CPU(8),
+					spec.WorkloadNode(),
+					spec.WorkloadNodeCPU(8),
+					spec.VolumeSize(100),
+				},
+			},
+			mode: ModeImmediate,
+			run:  TestLDRUpdateHeavy,
+		},
+		{
+			name: "ldr/kv0/workload=both/update_heavy/validated",
+			clusterSpec: multiClusterSpec{
+				leftNodes:  3,
+				rightNodes: 3,
+				clusterOpts: []spec.Option{
+					spec.CPU(8),
+					spec.WorkloadNode(),
+					spec.WorkloadNodeCPU(8),
+					spec.VolumeSize(100),
+				},
+			},
+			mode: ModeValidated,
+			run:  TestLDRUpdateHeavy,
 		},
 		{
 			name: "ldr/kv0/workload=both/shutdown_node",
@@ -143,6 +175,7 @@ func registerLogicalDataReplicationTests(r registry.Registry) {
 	}
 
 	for _, sp := range specs {
+
 		r.Add(registry.TestSpec{
 			Name:             sp.name,
 			Owner:            registry.OwnerDisasterRecovery,
@@ -162,14 +195,15 @@ func registerLogicalDataReplicationTests(r registry.Registry) {
 				}
 				setup, cleanup := mc.Start(ctx, t)
 				defer cleanup()
-				sp.run(ctx, t, c, setup)
-
+				sp.run(ctx, t, c, setup, sp.mode)
 			},
 		})
 	}
 }
 
-func TestLDRBasic(ctx context.Context, t test.Test, c cluster.Cluster, setup multiClusterSetup) {
+func TestLDRBasic(
+	ctx context.Context, t test.Test, c cluster.Cluster, setup multiClusterSetup, mode mode,
+) {
 	duration := 15 * time.Minute
 	initRows := 1000
 	maxBlockBytes := 1024
@@ -191,7 +225,7 @@ func TestLDRBasic(ctx context.Context, t test.Test, c cluster.Cluster, setup mul
 		tableName: "kv",
 	}
 
-	leftJobID, rightJobID := setupLDR(ctx, t, c, setup, ldrWorkload)
+	leftJobID, rightJobID := setupLDR(ctx, t, c, setup, ldrWorkload, mode)
 
 	// Setup latency verifiers
 	maxExpectedLatency := 2 * time.Minute
@@ -236,7 +270,7 @@ func TestLDRBasic(ctx context.Context, t test.Test, c cluster.Cluster, setup mul
 }
 
 func TestLDRSchemaChange(
-	ctx context.Context, t test.Test, c cluster.Cluster, setup multiClusterSetup,
+	ctx context.Context, t test.Test, c cluster.Cluster, setup multiClusterSetup, mode mode,
 ) {
 	duration := 15 * time.Minute
 	if c.IsLocal() {
@@ -254,7 +288,7 @@ func TestLDRSchemaChange(
 		tableName: "kv",
 	}
 
-	leftJobID, rightJobID := setupLDR(ctx, t, c, setup, ldrWorkload)
+	leftJobID, rightJobID := setupLDR(ctx, t, c, setup, ldrWorkload, mode)
 
 	// Setup latency verifiers
 	maxExpectedLatency := 2 * time.Minute
@@ -312,7 +346,7 @@ func TestLDRSchemaChange(
 }
 
 func TestLDRUpdateHeavy(
-	ctx context.Context, t test.Test, c cluster.Cluster, setup multiClusterSetup,
+	ctx context.Context, t test.Test, c cluster.Cluster, setup multiClusterSetup, mode mode,
 ) {
 
 	duration := 10 * time.Minute
@@ -331,7 +365,7 @@ func TestLDRUpdateHeavy(
 		tableName: "usertable",
 	}
 
-	leftJobID, rightJobID := setupLDR(ctx, t, c, setup, ldrWorkload)
+	leftJobID, rightJobID := setupLDR(ctx, t, c, setup, ldrWorkload, mode)
 
 	// Setup latency verifiers
 	maxExpectedLatency := 3 * time.Minute
@@ -376,7 +410,7 @@ func TestLDRUpdateHeavy(
 }
 
 func TestLDROnNodeShutdown(
-	ctx context.Context, t test.Test, c cluster.Cluster, setup multiClusterSetup,
+	ctx context.Context, t test.Test, c cluster.Cluster, setup multiClusterSetup, mode mode,
 ) {
 
 	duration := 10 * time.Minute
@@ -395,7 +429,7 @@ func TestLDROnNodeShutdown(
 		tableName: "kv",
 	}
 
-	leftJobID, rightJobID := setupLDR(ctx, t, c, setup, ldrWorkload)
+	leftJobID, rightJobID := setupLDR(ctx, t, c, setup, ldrWorkload, mode)
 
 	// Setup latency verifiers, remembering to account for latency spike from killing a node
 	maxExpectedLatency := 5 * time.Minute
@@ -472,7 +506,7 @@ func TestLDROnNodeShutdown(
 // aim to keep the workload going on both sides and wait for reconciliation
 // once the network partition has completed
 func TestLDROnNetworkPartition(
-	ctx context.Context, t test.Test, c cluster.Cluster, setup multiClusterSetup,
+	ctx context.Context, t test.Test, c cluster.Cluster, setup multiClusterSetup, mode mode,
 ) {
 	duration := 10 * time.Minute
 	if c.IsLocal() {
@@ -490,7 +524,7 @@ func TestLDROnNetworkPartition(
 		tableName: "kv",
 	}
 
-	leftJobID, rightJobID := setupLDR(ctx, t, c, setup, ldrWorkload)
+	leftJobID, rightJobID := setupLDR(ctx, t, c, setup, ldrWorkload, mode)
 
 	monitor := c.NewMonitor(ctx, setup.CRDBNodes())
 	monitor.Go(func(ctx context.Context) error {
@@ -552,9 +586,29 @@ func getLogicalDataReplicationJobInfo(db *gosql.DB, jobID int) (jobInfo, error)
 type ldrTestSpec struct {
 	name        string
 	clusterSpec multiClusterSpec
-	run         func(context.Context, test.Test, cluster.Cluster, multiClusterSetup)
+	run         func(context.Context, test.Test, cluster.Cluster, multiClusterSetup, mode)
+	mode        mode
+}
+
+type mode int
+
+func (m mode) String() string {
+	switch m {
+	case ModeImmediate:
+		return "immediate"
+	case ModeValidated:
+		return "validated"
+	default:
+		return "default"
+	}
 }
 
+const (
+	Default = iota
+	ModeImmediate
+	ModeValidated
+)
+
 type multiClusterSpec struct {
 	clusterOpts []spec.Option
 
@@ -669,6 +723,7 @@ func setupLDR(
 	c cluster.Cluster,
 	setup multiClusterSetup,
 	ldrWorkload LDRWorkload,
+	mode mode,
 ) (int, int) {
 	c.Run(ctx,
 		option.WithNodes(setup.workloadNode),
@@ -680,9 +735,13 @@ func setupLDR(
 	dbName, tableName := ldrWorkload.dbName, ldrWorkload.tableName
 
 	startLDR := func(targetDB *sqlutils.SQLRunner, sourceURL string) int {
+		options := ""
+		if mode.String() != "" {
+			options = fmt.Sprintf("WITH mode='%s'", mode)
+		}
 		targetDB.Exec(t, fmt.Sprintf("USE %s", dbName))
 		r := targetDB.QueryRow(t,
-			fmt.Sprintf("CREATE LOGICAL REPLICATION STREAM FROM TABLE %s ON $1 INTO TABLE %s", tableName, tableName),
+			fmt.Sprintf("CREATE LOGICAL REPLICATION STREAM FROM TABLE %s ON $1 INTO TABLE %s %s", tableName, tableName, options),
 			sourceURL,
 		)
 		var jobID int

diff --git a/pkg/raft/BUILD.bazel b/pkg/raft/BUILD.bazel
@@ -26,6 +26,7 @@ go_library(
         "//pkg/raft/raftstoreliveness",
         "//pkg/raft/tracker",
         "//pkg/util/hlc",
+        "@com_github_cockroachdb_errors//:errors",
         "@org_golang_x_exp//maps",
     ],
 )

diff --git a/pkg/raft/bootstrap.go b/pkg/raft/bootstrap.go
@@ -18,9 +18,8 @@
 package raft
 
 import (
-	"errors"
-
 	pb "github.com/cockroachdb/cockroach/pkg/raft/raftpb"
+	"github.com/cockroachdb/errors"
 )
 
 // Bootstrap initializes the RawNode for first use by appending configuration

diff --git a/pkg/raft/confchange/BUILD.bazel b/pkg/raft/confchange/BUILD.bazel
@@ -12,6 +12,7 @@ go_library(
         "//pkg/raft/quorum",
         "//pkg/raft/raftpb",
         "//pkg/raft/tracker",
+        "@com_github_cockroachdb_errors//:errors",
     ],
 )
 
@@ -29,5 +30,6 @@ go_test(
         "//pkg/raft/raftpb",
         "//pkg/raft/tracker",
         "@com_github_cockroachdb_datadriven//:datadriven",
+        "@com_github_cockroachdb_errors//:errors",
     ],
 )
diff --git a/pkg/raft/confchange/confchange.go b/pkg/raft/confchange/confchange.go
@@ -18,13 +18,13 @@
 package confchange
 
 import (
-	"errors"
 	"fmt"
 	"strings"
 
 	"github.com/cockroachdb/cockroach/pkg/raft/quorum"
 	pb "github.com/cockroachdb/cockroach/pkg/raft/raftpb"
 	"github.com/cockroachdb/cockroach/pkg/raft/tracker"
+	"github.com/cockroachdb/errors"
 )
 
 // Changer facilitates configuration changes. It exposes methods to handle

diff --git a/pkg/raft/confchange/datadriven_test.go b/pkg/raft/confchange/datadriven_test.go
@@ -18,7 +18,6 @@
 package confchange
 
 import (
-	"errors"
 	"fmt"
 	"strconv"
 	"strings"
@@ -28,6 +27,7 @@ import (
 	pb "github.com/cockroachdb/cockroach/pkg/raft/raftpb"
 	"github.com/cockroachdb/cockroach/pkg/raft/tracker"
 	"github.com/cockroachdb/datadriven"
+	"github.com/cockroachdb/errors"
 )
 
 func TestConfChangeDataDriven(t *testing.T) {

diff --git a/pkg/raft/raft.go b/pkg/raft/raft.go
@@ -21,7 +21,6 @@ import (
 	"bytes"
 	"context"
 	"crypto/rand"
-	"errors"
 	"fmt"
 	"math"
 	"math/big"
@@ -36,6 +35,7 @@ import (
 	pb "github.com/cockroachdb/cockroach/pkg/raft/raftpb"
 	"github.com/cockroachdb/cockroach/pkg/raft/raftstoreliveness"
 	"github.com/cockroachdb/cockroach/pkg/raft/tracker"
+	"github.com/cockroachdb/errors"
 	"golang.org/x/exp/maps"
 )
 

diff --git a/pkg/raft/rafttest/interaction_env_handler_add_nodes.go b/pkg/raft/rafttest/interaction_env_handler_add_nodes.go
@@ -19,7 +19,6 @@ package rafttest
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"reflect"
 	"testing"
@@ -31,6 +30,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/datadriven"
+	"github.com/cockroachdb/errors"
 )
 
 func (env *InteractionEnv) handleAddNodes(t *testing.T, d datadriven.TestData) error {

diff --git a/pkg/raft/rafttest/interaction_env_handler_process_append_thread.go b/pkg/raft/rafttest/interaction_env_handler_process_append_thread.go
@@ -18,13 +18,13 @@
 package rafttest
 
 import (
-	"errors"
 	"fmt"
 	"testing"
 
 	"github.com/cockroachdb/cockroach/pkg/raft"
 	"github.com/cockroachdb/cockroach/pkg/raft/raftpb"
 	"github.com/cockroachdb/datadriven"
+	"github.com/cockroachdb/errors"
 )
 
 func (env *InteractionEnv) handleProcessAppendThread(t *testing.T, d datadriven.TestData) error {

diff --git a/pkg/raft/rafttest/interaction_env_handler_report_unreachable.go b/pkg/raft/rafttest/interaction_env_handler_report_unreachable.go
@@ -18,10 +18,10 @@
 package rafttest
 
 import (
-	"errors"
 	"testing"
 
 	"github.com/cockroachdb/datadriven"
+	"github.com/cockroachdb/errors"
 )
 
 func (env *InteractionEnv) handleReportUnreachable(t *testing.T, d datadriven.TestData) error {