roachtest: add more expansive panic node mutator

This change adds a new mutator that is similar to the panic node
mutator, but will also upreplicate to an RF of 5 before panicking, and
drop back down to 3 after recovering. This allows for a much larger
duration of leaving the node down before recovering, as the cluster can
now survive 2 nodes being down. The current implementation is only
limited to tests that already have 5 nodes in the cluster, so this
mutator remains disabled by default.

Epic: none

Release note: none

Author: Dev-Kyle

pkg/cmd/roachtest/roachtestutil/mixedversion/BUILD.bazel

@@ -25,6 +25,7 @@ go_library(
         "//pkg/cmd/roachtest/spec",
         "//pkg/cmd/roachtest/test",
         "//pkg/roachpb",
+        "//pkg/roachprod",
         "//pkg/roachprod/failureinjection/failures",
         "//pkg/roachprod/install",
         "//pkg/roachprod/logger",

pkg/cmd/roachtest/roachtestutil/mixedversion/mixedversion.go

@@ -349,6 +349,7 @@ type (
 		enabledDeploymentModes         []DeploymentMode
 		tag                            string
 		overriddenMutatorProbabilities map[string]float64
+		hooksSupportFailureInjection   bool
 	}
 
 	CustomOption func(*testOptions)
@@ -400,6 +401,12 @@ type (
 	DeploymentMode string
 )
 
+// EnableHooksDuringFailureInjection is an option that can be passed to
+// `NewTest` to enable the use of mixed-version hooks during failure injections.
+func EnableHooksDuringFailureInjection(opts *testOptions) {
+	opts.hooksSupportFailureInjection = true
+}
+
 // NeverUseFixtures is an option that can be passed to `NewTest` to
 // disable the use of fixtures in the test. Necessary if the test
 // wants to use a number of cockroach nodes other than 4.

pkg/cmd/roachtest/roachtestutil/mixedversion/mutators.go

@@ -409,6 +409,10 @@ func (m panicNodeMutator) Generate(
 	idx := newStepIndex(plan)
 	nodeList := planner.currentContext.System.Descriptor.Nodes
 
+	// If we have at least 5 nodes, we can safely upreplicate to 5X before panicking a node.
+	// This allows for a longer panic duration before recovery.
+	supportsUpReplication := len(nodeList) >= 5
+
 	for _, upgrade := range upgrades {
 		possiblePointsInTime := upgrade.
 			// We don't want to panic concurrently with other steps, and inserting before a concurrent step
@@ -426,13 +430,21 @@ func (m panicNodeMutator) Generate(
 
 		isIncompatibleStep := func(s *singleStep) bool {
 			// Restarting the system on a different node while our panicked node is still dead can
-			// cause the cluster to lose quorum, so we avoid any system restarts.
-			_, restart := s.impl.(restartWithNewBinaryStep)
+			// cause the cluster to lose quorum, so we avoid any system restarts. If
+			// the cluster has a high enough node count however, we can upreplicate
+			// to 5X before panicking, allowing us to safely restart other nodes.
+			restartImpl, restart := s.impl.(restartWithNewBinaryStep)
+			if supportsUpReplication {
+				// We can restart other nodes, but we do not want to restart the
+				// node that is being panicked, as the panic recover step expects
+				// the node to be down in order to restart it.
+				restart = restart && restartImpl.node == targetNode[0]
+			}
 			// Waiting for stable cluster version targets every node in
 			// the cluster, so a node cannot be dead during this step.
 			_, waitForStable := s.impl.(waitForStableClusterVersionStep)
 			// Many hook steps do not support running with a dead node,
-			// so we avoid inserting after a hook step.
+			// so we avoid inserting after an incompatible hook step.
 			_, runHook := s.impl.(runHookStep)
 
 			if idx.IsConcurrent(s) {
@@ -445,7 +457,7 @@ func (m panicNodeMutator) Generate(
 				firstStepInConcurrentBlock = nil
 			}
 
-			return restart || waitForStable || runHook || s.context.System.hasUnavailableNodes
+			return restart || waitForStable || (runHook && !planner.options.hooksSupportFailureInjection) || s.context.System.hasUnavailableNodes
 		}
 
 		// The node should be restarted after the panic, but before any steps that are
@@ -469,19 +481,35 @@ func (m panicNodeMutator) Generate(
 
 		restartDesc := fmt.Sprintf("restarting node %d after panic", targetNode[0])
 
+		var addUpReplicateStep []mutation
+		if supportsUpReplication {
+			addUpReplicateStep = stepToPanic.
+				InsertBefore(alterReplicationFactorStep{5, targetNode})
+		}
 		addPanicStep := stepToPanic.
 			InsertBefore(panicNodeStep{planner.currentContext.System.Descriptor.Nodes[0], targetNode})
 		var addRestartStep []mutation
+		var addDownReplicateStep []mutation
 		var restartStep stepSelector
 		// If validEndStep is nil, it means that there are no steps after the panic step that
 		// are compatible with a dead node, so we immediately restart the node after the panic.
 		if validEndStep == nil {
 			restartStep = cutStep
-			addRestartStep = cutStep.InsertBefore(restartNodeStep{planner.currentContext.System.Descriptor.Nodes[0], targetNode, planner.rt, restartDesc})
+			addRestartStep = restartStep.InsertBefore(restartNodeStep{planner.currentContext.System.Descriptor.Nodes[0], targetNode, planner.rt, restartDesc})
 		} else {
 			restartStep = validEndStep.RandomStep(rng)
-			addRestartStep = restartStep.
-				Insert(rng, restartNodeStep{planner.currentContext.System.Descriptor.Nodes[0], targetNode, planner.rt, restartDesc})
+			if supportsUpReplication {
+				addRestartStep = restartStep.
+					InsertBefore(restartNodeStep{planner.currentContext.System.Descriptor.Nodes[0], targetNode, planner.rt, restartDesc})
+			} else {
+				addRestartStep = restartStep.
+					Insert(rng, restartNodeStep{planner.currentContext.System.Descriptor.Nodes[0], targetNode, planner.rt, restartDesc})
+			}
+		}
+
+		if supportsUpReplication {
+			addDownReplicateStep = restartStep.
+				InsertBefore(alterReplicationFactorStep{3, targetNode})
 		}
 
 		failureContextSteps, _ := validStartStep.CutBefore(func(s *singleStep) bool {
@@ -493,6 +521,10 @@ func (m panicNodeMutator) Generate(
 
 		mutations = append(mutations, addPanicStep...)
 		mutations = append(mutations, addRestartStep...)
+		if supportsUpReplication {
+			mutations = append(addUpReplicateStep, mutations...)
+			mutations = append(mutations, addDownReplicateStep...)
+		}
 	}
 
 	return mutations, nil
@@ -562,7 +594,8 @@ func (m networkPartitionMutator) Generate(
 			_, restartSystem := s.impl.(restartWithNewBinaryStep)
 			_, restartTenant := s.impl.(restartVirtualClusterStep)
 			// Many hook steps require communication between specific nodes, so we
-			// should recover the network partition before running them.
+			// should recover the network partition before running any incompatible
+			// hook steps.
 			_, runHook := s.impl.(runHookStep)
 			// Waiting for stable cluster version requires communication between
 			// all nodes in the cluster, so we should recover the network partition
@@ -585,7 +618,7 @@ func (m networkPartitionMutator) Generate(
 			} else {
 				unavailableNodes = s.context.System.hasUnavailableNodes
 			}
-			return unavailableNodes || restartTenant || restartSystem || runHook || waitForStable
+			return unavailableNodes || restartTenant || restartSystem || (runHook && !planner.options.hooksSupportFailureInjection) || waitForStable
 		}
 
 		_, validStartStep := upgrade.CutAfter(func(s *singleStep) bool {

pkg/cmd/roachtest/roachtestutil/mixedversion/steps.go

@@ -14,8 +14,10 @@ import (
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/option"
+	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/roachtestutil"
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/roachtestutil/clusterupgrade"
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/test"
+	"github.com/cockroachdb/cockroach/pkg/roachprod"
 	"github.com/cockroachdb/cockroach/pkg/roachprod/failureinjection/failures"
 	"github.com/cockroachdb/cockroach/pkg/roachprod/install"
 	"github.com/cockroachdb/cockroach/pkg/roachprod/logger"
@@ -969,3 +971,42 @@ func (s networkPartitionRecoveryStep) Run(
 func (s networkPartitionRecoveryStep) ConcurrencyDisabled() bool {
 	return false
 }
+
+type alterReplicationFactorStep struct {
+	replicationFactor int
+	targetNode        option.NodeListOption
+}
+
+func (s alterReplicationFactorStep) Background() shouldStop { return nil }
+
+func (s alterReplicationFactorStep) Description() string {
+	return fmt.Sprintf("alter replication factor to %d", s.replicationFactor)
+}
+
+func (s alterReplicationFactorStep) Run(
+	ctx context.Context, l *logger.Logger, rng *rand.Rand, h *Helper,
+) error {
+	stmt := fmt.Sprintf("ALTER RANGE default CONFIGURE ZONE USING num_replicas = %d", s.replicationFactor)
+	if err := h.System.Exec(
+		rng,
+		stmt,
+	); err != nil {
+		return errors.Wrap(err, "failed to change replication factor")
+	}
+
+	replicationLogger, loggerName, err := roachtestutil.LoggerForCmd(l, s.targetNode, "range-replication")
+	if err != nil {
+		return errors.Wrapf(err, "failed to create logger %s", loggerName)
+	}
+
+	l.Printf("waiting to reach replication factor of %dX; details in %s.log", s.replicationFactor, loggerName)
+	db := h.System.Connect(s.targetNode[0])
+	if err := roachtestutil.WaitForReplication(ctx, replicationLogger, db, s.replicationFactor, roachprod.AtLeastReplicationFactor); err != nil {
+		return errors.Wrapf(err, "failed to reach replication factor of %dX", s.replicationFactor)
+	}
+	return nil
+}
+
+func (s alterReplicationFactorStep) ConcurrencyDisabled() bool {
+	return false
+}

pkg/cmd/roachtest/tests/tpcc.go

@@ -518,6 +518,9 @@ var tpccSupportedWarehouses = []struct {
 	{hardware: "ibm-n5cpu16", v: version.MustParse(`v19.1.0-alpha.0`), warehouses: 1300},
 	// Ditto.
 	{hardware: "gce-n5cpu16", v: version.MustParse(`v2.1.0-alpha.0`), warehouses: 1300},
+
+	{hardware: "gce-n6cpu16", v: version.MustParse(`v19.1.0-alpha.0`), warehouses: 2000},
+	{hardware: "gce-n6cpu16", v: version.MustParse(`v2.1.0-alpha.0`), warehouses: 2000},
 }
 
 // tpccMaxRate calculates the max rate of the workload given a number of warehouses.
@@ -560,7 +563,7 @@ func maxSupportedTPCCWarehouses(
 // workload is running. The number of database upgrades is randomized
 // by the mixed-version framework which chooses a random predecessor version
 // and upgrades until it reaches the current version.
-func runTPCCMixedHeadroom(ctx context.Context, t test.Test, c cluster.Cluster) {
+func runTPCCMixedHeadroom(ctx context.Context, t test.Test, c cluster.Cluster, chaos bool) {
 	maxWarehouses := maxSupportedTPCCWarehouses(*t.BuildVersion(), c.Cloud(), c.Spec())
 	headroomWarehouses := int(float64(maxWarehouses) * 0.7)
 
@@ -569,20 +572,39 @@ func runTPCCMixedHeadroom(ctx context.Context, t test.Test, c cluster.Cluster) {
 	// The full 6.5m import ran into out of disk errors (on 250gb machines),
 	// hence division by two.
 	bankRows := 65104166 / 2
+
 	if c.IsLocal() {
 		bankRows = 1000
 	}
+	// If the test is a chaos test, we decrease the number of rows and warehouses
+	// in order to lower the time it takes to reach replication with a larger cluster.
+	if chaos {
+		bankRows = 1000
+		headroomWarehouses = 200
+	}
 
-	mvt := mixedversion.NewTest(
-		ctx, t, t.L(), c, c.CRDBNodes(),
+	customOpts := []mixedversion.CustomOption{
 		// We test only upgrades from 23.2 in this test because it uses
 		// the `workload fixtures import` command, which is only supported
 		// reliably multi-tenant mode starting from that version.
 		mixedversion.MinimumSupportedVersion("v23.2.0"),
 		// We limit the total number of plan steps to 70, which is roughly 80% of all plan lengths.
 		// See #138014 for more details.
 		mixedversion.MaxNumPlanSteps(70),
-	)
+	}
+
+	// If the test is a chaos test, we want to opt for the more expansive panic
+	// mutator, as well any other appropriate test opts for the unique test.
+	if chaos {
+		customOpts = append([]mixedversion.CustomOption{
+			mixedversion.NeverUseFixtures,
+			mixedversion.EnableHooksDuringFailureInjection,
+		},
+			customOpts...)
+	}
+
+	mvt := mixedversion.NewTest(
+		ctx, t, t.L(), c, c.CRDBNodes(), customOpts...)
 
 	tenantFeaturesEnabled := make(chan struct{})
 	enableTenantFeatures := func(ctx context.Context, l *logger.Logger, rng *rand.Rand, h *mixedversion.Helper) error {
@@ -594,7 +616,7 @@ func runTPCCMixedHeadroom(ctx context.Context, t test.Test, c cluster.Cluster) {
 		l.Printf("waiting for tenant features to be enabled")
 		<-tenantFeaturesEnabled
 
-		randomNode := c.Node(c.CRDBNodes().SeededRandNode(rng)[0])
+		randomNode := c.Node(h.AvailableNodes().SeededRandNode(rng)[0])
 		cmd := tpccImportCmdWithCockroachBinary(test.DefaultCockroachPath, "", "tpcc", headroomWarehouses, fmt.Sprintf("{pgurl%s}", randomNode))
 		return c.RunE(ctx, option.WithNodes(randomNode), cmd)
 	}
@@ -603,7 +625,7 @@ func runTPCCMixedHeadroom(ctx context.Context, t test.Test, c cluster.Cluster) {
 	// upgrade machinery, in which a) all ranges are touched and b) work proportional
 	// to the amount data may be carried out.
 	importLargeBank := func(ctx context.Context, l *logger.Logger, rng *rand.Rand, h *mixedversion.Helper) error {
-		randomNode := c.Node(c.CRDBNodes().SeededRandNode(rng)[0])
+		randomNode := c.Node(h.AvailableNodes().SeededRandNode(rng)[0])
 		// Upload a versioned cockroach binary to the random node. The bank workload
 		// is no longer backwards compatible after #149374, so we need to use the same
 		// version as the cockroach cluster.
@@ -647,7 +669,7 @@ func runTPCCMixedHeadroom(ctx context.Context, t test.Test, c cluster.Cluster) {
 			labelsMap = getTpccLabels(headroomWarehouses, rampDur, workloadDur/time.Millisecond, nil)
 		}
 		cmd := roachtestutil.NewCommand("./cockroach workload run tpcc").
-			Arg("{pgurl%s}", c.CRDBNodes()).
+			Arg("{pgurl%s}", h.AvailableNodes()).
 			Flag("duration", workloadDur).
 			Flag("warehouses", headroomWarehouses).
 			Flag("histograms", histogramsPath).
@@ -728,6 +750,21 @@ func registerTPCC(r registry.Registry) {
 			})
 		},
 	})
+	mixedHeadroomChaosSpec := r.MakeClusterSpec(6, spec.CPU(16), spec.WorkloadNode(), spec.RandomlyUseZfs())
+	r.Add(registry.TestSpec{
+		Name:              "tpcc/mixed-headroom/chaos/" + mixedHeadroomChaosSpec.String(),
+		Timeout:           7 * time.Hour,
+		Owner:             registry.OwnerTestEng,
+		CompatibleClouds:  registry.AllClouds.NoAWS().NoIBM(),
+		Suites:            registry.Suites(registry.MixedVersion),
+		Cluster:           mixedHeadroomChaosSpec,
+		EncryptionSupport: registry.EncryptionMetamorphic,
+		Monitor:           true,
+		Randomized:        true,
+		Run: func(ctx context.Context, t test.Test, c cluster.Cluster) {
+			runTPCCMixedHeadroom(ctx, t, c, true)
+		},
+	})
 
 	mixedHeadroomSpec := r.MakeClusterSpec(5, spec.CPU(16), spec.WorkloadNode(), spec.RandomlyUseZfs())
 	r.Add(registry.TestSpec{
@@ -749,7 +786,7 @@ func registerTPCC(r registry.Registry) {
 		Monitor:           true,
 		Randomized:        true,
 		Run: func(ctx context.Context, t test.Test, c cluster.Cluster) {
-			runTPCCMixedHeadroom(ctx, t, c)
+			runTPCCMixedHeadroom(ctx, t, c, false)
 		},
 	})