Merge #153989 #154275 #154435

153989: roachtest: add variations of c2c/bulk with diff settings r=dt a=dt

We may linger in a place where we tell some customers to customize non-public
cluster settings to optimize for certain characteristics (pcr lag vs bulk op perf)
until we can develop a better set of default behaviors. While we're in this state,
we should test some of the more common configs we think we might recommend, even though
these will still be custom configs, recommended on case-by-case basis.

Release note: none.
Epic: none.

154275: util/intsets: fix outdated comment r=mgartner a=mgartner

Release note: None

154435: allocatorimpl: remove alwaysAllowDecisionWithoutStats r=tbg a=wenyihu6

Previously, we called TransferLeaseTarget with alwaysAllowDecisionWithoutStats,
but it was only relevant when the lease transfer goal was
allocator.FollowTheWorkload. The only case where we passed true for
alwaysAllowDecisionWithoutStats was from storeRebalancer.chooseLeaseToTransfer,
where opt.Goal was set to allocator.LoadConvergence. This commit removes it
since it is unclear whether flag was useful.

Epic: none 
Release note: none 

Co-authored-by: David Taylor <[email protected]>
Co-authored-by: Marcus Gartner <[email protected]>
Co-authored-by: wenyihu6 <[email protected]>

Author: 4 people

pkg/cmd/roachtest/cluster.go

@@ -325,7 +325,10 @@ func initBinariesAndLibraries() {
 	cockroachPath := roachtestflags.CockroachPath
 	cockroachEAPath := roachtestflags.CockroachEAPath
 	workloadPath := roachtestflags.WorkloadPath
-	cockroach[defaultArch], _ = resolveBinary("cockroach", cockroachPath, defaultArch, true, false)
+	// Skip cockroach binary validation if using --cockroach-stage
+	if roachtestflags.CockroachStage == "" {
+		cockroach[defaultArch], _ = resolveBinary("cockroach", cockroachPath, defaultArch, true, false)
+	}
 	// Let the test runner verify the workload binary exists if TestSpec.RequiresDeprecatedWorkload is true.
 	workload[defaultArch], _ = resolveBinary("workload", workloadPath, defaultArch, false, false)
 	cockroachEA[defaultArch], err = resolveBinary("cockroach-ea", cockroachEAPath, defaultArch, false, true)
@@ -336,7 +339,9 @@ func initBinariesAndLibraries() {
 	if roachtestflags.ARM64Probability > 0 && defaultArch != vm.ArchARM64 {
 		fmt.Printf("Locating and verifying binaries for os=%q, arch=%q\n", defaultOSName, vm.ArchARM64)
 		// We need to verify we have all the required binaries for arm64.
-		cockroach[vm.ArchARM64], _ = resolveBinary("cockroach", cockroachPath, vm.ArchARM64, true, false)
+		if roachtestflags.CockroachStage == "" {
+			cockroach[vm.ArchARM64], _ = resolveBinary("cockroach", cockroachPath, vm.ArchARM64, true, false)
+		}
 		workload[vm.ArchARM64], _ = resolveBinary("workload", workloadPath, vm.ArchARM64, true, false)
 		cockroachEA[vm.ArchARM64], err = resolveBinary("cockroach-ea", cockroachEAPath, vm.ArchARM64, false, true)
 		if err != nil {
@@ -346,7 +351,9 @@ func initBinariesAndLibraries() {
 	if roachtestflags.FIPSProbability > 0 && defaultArch != vm.ArchFIPS {
 		fmt.Printf("Locating and verifying binaries for os=%q, arch=%q\n", defaultOSName, vm.ArchFIPS)
 		// We need to verify we have all the required binaries for fips.
-		cockroach[vm.ArchFIPS], _ = resolveBinary("cockroach", cockroachPath, vm.ArchFIPS, true, false)
+		if roachtestflags.CockroachStage == "" {
+			cockroach[vm.ArchFIPS], _ = resolveBinary("cockroach", cockroachPath, vm.ArchFIPS, true, false)
+		}
 		workload[vm.ArchFIPS], _ = resolveBinary("workload", workloadPath, vm.ArchFIPS, true, false)
 		cockroachEA[vm.ArchFIPS], err = resolveBinary("cockroach-ea", cockroachEAPath, vm.ArchFIPS, false, true)
 		if err != nil {
@@ -1939,7 +1946,11 @@ func (c *clusterImpl) PutE(
 func (c *clusterImpl) PutCockroach(ctx context.Context, l *logger.Logger, t *testImpl) error {
 	if roachtestflags.CockroachStage != "" {
 		// Use staging instead of upload when --cockroach-stage is specified
-		return c.Stage(ctx, l, "cockroach", roachtestflags.CockroachStage, ".", c.All())
+		stageVersion := roachtestflags.CockroachStage
+		if stageVersion == "latest" {
+			stageVersion = "" // Stage() expects empty string for latest
+		}
+		return c.Stage(ctx, l, "cockroach", stageVersion, ".", c.All())
 	}
 	return c.PutE(ctx, l, t.Cockroach(), test.DefaultCockroachPath, c.All())
 }

pkg/cmd/roachtest/main.go

@@ -581,8 +581,4 @@ func validateAndConfigure(cmd *cobra.Command, args []string) {
 		printErrAndExit(fmt.Errorf("--cockroach and --cockroach-stage are mutually exclusive. Use one or the other"))
 	}
 
-	// Normalize "latest" to empty string for staging system
-	if roachtestflags.CockroachStage == "latest" {
-		roachtestflags.CockroachStage = ""
-	}
 }

pkg/cmd/roachtest/tests/cluster_to_cluster.go

@@ -398,6 +398,8 @@ type replicateBulkOps struct {
 
 	// debugSkipRollback skips all rollback steps during the test.
 	debugSkipRollback bool
+
+	withSettings []struct{ setting, value string }
 }
 
 func (bo replicateBulkOps) sourceInitCmd(tenantName string, nodes option.NodeListOption) string {
@@ -411,6 +413,18 @@ func (bo replicateBulkOps) sourceRunCmd(tenantName string, nodes option.NodeList
 func (bo replicateBulkOps) runDriver(
 	workloadCtx context.Context, c cluster.Cluster, t test.Test, setup *c2cSetup,
 ) error {
+	mainTenantConn := c.Conn(workloadCtx, t.L(), 1, option.VirtualClusterName(setup.src.name))
+	for _, pair := range bo.withSettings {
+		settingStmt := fmt.Sprintf("SET CLUSTER SETTING %s = '%s'", pair.setting, pair.value)
+		t.L().Printf("Setting on sys/main/standby-sys: %s", settingStmt)
+		setup.src.sysSQL.Exec(t, settingStmt)
+		// PCR settings are system-only; assume others are app-level.
+		if !strings.Contains(pair.setting, "physical_replication") {
+			if _, err := mainTenantConn.ExecContext(workloadCtx, settingStmt); err != nil {
+				return err
+			}
+		}
+	}
 	runBackupMVCCRangeTombstones(workloadCtx, t, c, mvccRangeTombstoneConfig{
 		skipBackupRestore: true,
 		skipClusterSetup:  true,
@@ -1529,7 +1543,7 @@ func registerClusterToCluster(r registry.Registry) {
 			suites:                    registry.Suites(registry.Nightly),
 		},
 		{
-			name:               "c2c/BulkOps",
+			name:               "c2c/BulkOps/settings=none",
 			srcNodes:           4,
 			dstNodes:           4,
 			cpus:               8,
@@ -1553,6 +1567,63 @@ func registerClusterToCluster(r registry.Registry) {
 			clouds:                    registry.OnlyGCE,
 			suites:                    registry.Suites(registry.Nightly),
 		},
+		{
+			name:     "c2c/BulkOps/settings=ac-import",
+			srcNodes: 4,
+			dstNodes: 4,
+			cpus:     8,
+			pdSize:   100,
+			workload: replicateBulkOps{withSettings: []struct{ setting, value string }{
+				{"bulkio.import.elastic_control.enabled", "true"},
+				{"bulkio.elastic_cpu_control.request_duration", "3ms"},
+			}},
+			timeout:            2 * time.Hour,
+			additionalDuration: 0,
+			// Cutover currently takes around 4 minutes, perhaps because we need to
+			// revert 10 GB of replicated data.
+			//
+			// TODO(msbutler): investigate further if cutover can be sped up.
+			cutoverTimeout: 20 * time.Minute,
+			cutover:        5 * time.Minute,
+			// In a few ad hoc runs, the max latency hikes up to 27 minutes before lag
+			// replanning and distributed catch up scans fix the poor initial plan. If
+			// max accepted latency doubles, then there's likely a regression.
+			maxAcceptedLatency: 1 * time.Hour,
+			// Skipping node distribution check because there is little data on the
+			// source when the replication stream begins.
+			skipNodeDistributionCheck: true,
+			clouds:                    registry.OnlyGCE,
+			suites:                    registry.Suites(registry.Nightly),
+		},
+		{
+			name:     "c2c/BulkOps/settings=ac-and-splits",
+			srcNodes: 4,
+			dstNodes: 4,
+			cpus:     8,
+			pdSize:   100,
+			workload: replicateBulkOps{withSettings: []struct{ setting, value string }{
+				{"bulkio.import.elastic_control.enabled", "true"},
+				{"bulkio.elastic_cpu_control.request_duration", "3ms"},
+				{"physical_replication.consumer.ingest_split_event.enabled", "true"},
+			}},
+			timeout:            2 * time.Hour,
+			additionalDuration: 0,
+			// Cutover currently takes around 4 minutes, perhaps because we need to
+			// revert 10 GB of replicated data.
+			//
+			// TODO(msbutler): investigate further if cutover can be sped up.
+			cutoverTimeout: 20 * time.Minute,
+			cutover:        5 * time.Minute,
+			// In a few ad hoc runs, the max latency hikes up to 27 minutes before lag
+			// replanning and distributed catch up scans fix the poor initial plan. If
+			// max accepted latency doubles, then there's likely a regression.
+			maxAcceptedLatency: 1 * time.Hour,
+			// Skipping node distribution check because there is little data on the
+			// source when the replication stream begins.
+			skipNodeDistributionCheck: true,
+			clouds:                    registry.OnlyGCE,
+			suites:                    registry.Suites(registry.Nightly),
+		},
 		{
 			name:               "c2c/BulkOps/singleImport",
 			srcNodes:           4,

pkg/cmd/roachtest/tests/latency_verifier.go

@@ -44,6 +44,7 @@ type latencyVerifier struct {
 	catchupScanEveryN roachtestutil.EveryN
 
 	maxSeenSteadyLatency time.Duration
+	tooLargeEveryN       roachtestutil.EveryN
 	maxSeenSteadyEveryN  roachtestutil.EveryN
 	latencyBecameSteady  bool
 
@@ -74,8 +75,9 @@ func makeLatencyVerifier(
 		setTestStatus:            setTestStatus,
 		latencyHist:              hist,
 		tolerateErrors:           tolerateErrors,
-		maxSeenSteadyEveryN:      roachtestutil.Every(10 * time.Second),
-		catchupScanEveryN:        roachtestutil.Every(2 * time.Second),
+		tooLargeEveryN:           roachtestutil.Every(120 * time.Second),
+		maxSeenSteadyEveryN:      roachtestutil.Every(30 * time.Second),
+		catchupScanEveryN:        roachtestutil.Every(10 * time.Second),
 	}
 }
 
@@ -132,7 +134,9 @@ func (lv *latencyVerifier) noteHighwater(highwaterTime time.Time) {
 		return
 	}
 	if err := lv.latencyHist.RecordValue(latency.Nanoseconds()); err != nil {
-		lv.logger.Printf("%s: could not record value %s: %s\n", lv.name, latency, err)
+		if lv.tooLargeEveryN.ShouldLog() {
+			lv.logger.Printf("%s: could not record value %s: %s\n", lv.name, latency, err)
+		}
 	}
 	if latency > lv.maxSeenSteadyLatency {
 		lv.maxSeenSteadyLatency = latency

pkg/kv/kvserver/allocator/allocatorimpl/allocator.go

@@ -2389,7 +2389,6 @@ func (a *Allocator) TransferLeaseTarget(
 		SendStreamStats(*rac2.RangeSendStreamStats)
 	},
 	usageInfo allocator.RangeUsageInfo,
-	forceDecisionWithoutStats bool,
 	opts allocator.TransferLeaseOptions,
 ) roachpb.ReplicaDescriptor {
 	if a.knobs != nil {
@@ -2443,10 +2442,7 @@ func (a *Allocator) TransferLeaseTarget(
 		if !excludeLeaseRepl {
 			switch transferDec {
 			case shouldNotTransfer:
-				if !forceDecisionWithoutStats {
-					return roachpb.ReplicaDescriptor{}
-				}
-				fallthrough
+				return roachpb.ReplicaDescriptor{}
 			case decideWithoutStats:
 				if !a.shouldTransferLeaseForLeaseCountConvergence(ctx, storePool, sl, source, validTargets) {
 					return roachpb.ReplicaDescriptor{}

pkg/kv/kvserver/allocator/allocatorimpl/allocator_test.go

@@ -2052,7 +2052,6 @@ func TestAllocatorTransferLeaseTarget(t *testing.T) {
 					storeID:           c.leaseholder,
 				},
 				allocator.RangeUsageInfo{}, /* stats */
-				false,                      /* forceDecisionWithoutStats */
 				allocator.TransferLeaseOptions{
 					ExcludeLeaseRepl:       c.excludeLeaseRepl,
 					CheckCandidateFullness: true,
@@ -2184,7 +2183,6 @@ func TestAllocatorTransferLeaseTargetIOOverloadCheck(t *testing.T) {
 					storeID:           tc.leaseholder,
 				},
 				allocator.RangeUsageInfo{}, /* stats */
-				false,                      /* forceDecisionWithoutStats */
 				allocator.TransferLeaseOptions{
 					CheckCandidateFullness: true,
 				},
@@ -2297,7 +2295,6 @@ func TestAllocatorTransferLeaseToReplicasNeedingSnapshot(t *testing.T) {
 				c.existing,
 				repl,
 				allocator.RangeUsageInfo{},
-				false, /* alwaysAllowDecisionWithoutStats */
 				allocator.TransferLeaseOptions{
 					ExcludeLeaseRepl:       c.excludeLeaseRepl,
 					CheckCandidateFullness: true,
@@ -2441,7 +2438,6 @@ func TestAllocatorTransferLeaseToReplicasNeedingCatchup(t *testing.T) {
 				c.existing,
 				repl,
 				allocator.RangeUsageInfo{},
-				false, /* alwaysAllowDecisionWithoutStats */
 				allocator.TransferLeaseOptions{
 					ExcludeLeaseRepl:       c.excludeLeaseRepl,
 					CheckCandidateFullness: true,
@@ -2537,7 +2533,6 @@ func TestAllocatorTransferLeaseTargetConstraints(t *testing.T) {
 					storeID:           c.leaseholder,
 				},
 				allocator.RangeUsageInfo{},
-				false, /* forceDecisionWithoutStats */
 				allocator.TransferLeaseOptions{
 					ExcludeLeaseRepl:       false,
 					CheckCandidateFullness: true,
@@ -2650,7 +2645,6 @@ func TestAllocatorTransferLeaseTargetDraining(t *testing.T) {
 					storeID:           c.leaseholder,
 				},
 				allocator.RangeUsageInfo{},
-				false, /* forceDecisionWithoutStats */
 				allocator.TransferLeaseOptions{
 					ExcludeLeaseRepl:       c.excludeLeaseRepl,
 					CheckCandidateFullness: true,
@@ -3312,7 +3306,6 @@ func TestAllocatorLeasePreferences(t *testing.T) {
 					storeID:           c.leaseholder,
 				},
 				allocator.RangeUsageInfo{},
-				false, /* forceDecisionWithoutStats */
 				allocator.TransferLeaseOptions{
 					ExcludeLeaseRepl:       false,
 					CheckCandidateFullness: true,
@@ -3332,7 +3325,6 @@ func TestAllocatorLeasePreferences(t *testing.T) {
 					storeID:           c.leaseholder,
 				},
 				allocator.RangeUsageInfo{},
-				false, /* forceDecisionWithoutStats */
 				allocator.TransferLeaseOptions{
 					ExcludeLeaseRepl:       true,
 					CheckCandidateFullness: true,
@@ -3426,7 +3418,6 @@ func TestAllocatorLeasePreferencesMultipleStoresPerLocality(t *testing.T) {
 					storeID:           c.leaseholder,
 				},
 				allocator.RangeUsageInfo{},
-				false, /* forceDecisionWithoutStats */
 				allocator.TransferLeaseOptions{
 					ExcludeLeaseRepl:       false,
 					CheckCandidateFullness: true,
@@ -3447,7 +3438,6 @@ func TestAllocatorLeasePreferencesMultipleStoresPerLocality(t *testing.T) {
 					storeID:           c.leaseholder,
 				},
 				allocator.RangeUsageInfo{},
-				false, /* forceDecisionWithoutStats */
 				allocator.TransferLeaseOptions{
 					ExcludeLeaseRepl:       true,
 					CheckCandidateFullness: true,
@@ -6084,7 +6074,6 @@ func TestAllocatorTransferLeaseTargetLoadBased(t *testing.T) {
 					storeID:           c.leaseholder,
 				},
 				usage,
-				false,
 				allocator.TransferLeaseOptions{
 					ExcludeLeaseRepl:       c.excludeLeaseRepl,
 					CheckCandidateFullness: true,

pkg/kv/kvserver/allocator/plan/lease.go

@@ -142,7 +142,6 @@ func (lp LeasePlanner) PlanOneChange(
 		existingVoters,
 		repl,
 		usage,
-		false, /* forceDecisionWithoutStats */
 		allocator.TransferLeaseOptions{
 			Goal:                   allocator.FollowTheWorkload,
 			ExcludeLeaseRepl:       false,

pkg/kv/kvserver/allocator/plan/replicate.go

@@ -900,7 +900,6 @@ func (rp ReplicaPlanner) maybeTransferLeaseAwayTarget(
 		desc.Replicas().VoterDescriptors(),
 		repl,
 		usageInfo,
-		false, /* forceDecisionWithoutStats */
 		allocator.TransferLeaseOptions{
 			Goal: allocator.LeaseCountConvergence,
 			// NB: This option means that the allocator is asked to not consider the

pkg/kv/kvserver/replicate_queue.go

@@ -1051,7 +1051,6 @@ func (rq *replicateQueue) shedLease(
 		desc.Replicas().VoterDescriptors(),
 		repl,
 		rangeUsageInfo,
-		false, /* forceDecisionWithoutStats */
 		opts,
 	)
 	if targetDesc == (roachpb.ReplicaDescriptor{}) {

pkg/kv/kvserver/store_rebalancer.go

@@ -814,7 +814,6 @@ func (sr *StoreRebalancer) chooseLeaseToTransfer(
 			candidates,
 			candidateReplica,
 			candidateReplica.RangeUsageInfo(),
-			true, /* forceDecisionWithoutStats */
 			allocator.TransferLeaseOptions{
 				Goal:             allocator.LoadConvergence,
 				ExcludeLeaseRepl: false,