roachtest: refactor mixedversion tests to use global test monitor

Change the mixedversion tests to use the global test monitor.
We want to move all tests to the global test monitor eventually
and the mixedversion tests are a good target to ensure our
monitor changes targeting specific processes work as they
create separate process clusters.

Author: DarrylWong

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

@@ -283,20 +283,6 @@ func (h *Helper) GoCommand(cmd string, nodes option.NodeListOption) context.Canc
 	}, task.Name(desc))
 }
 
-// ExpectDeath alerts the testing infrastructure that a node is
-// expected to die. Regular restarts as part of the mixedversion
-// testing are already taken into account. This function should only
-// be used by tests that perform their own node restarts or chaos
-// events.
-func (h *Helper) ExpectDeath() {
-	h.ExpectDeaths(1)
-}
-
-// ExpectDeaths is the general version of `ExpectDeath()`.
-func (h *Helper) ExpectDeaths(n int) {
-	h.runner.monitor.ExpectDeaths(n)
-}
-
 // ClusterVersion returns the currently active cluster version. Avoids
 // querying the database if we are not running migrations, since the
 // test runner has cached version of the cluster versions.

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

@@ -642,6 +642,10 @@ func NewTest(
 	crdbNodes option.NodeListOption,
 	options ...CustomOption,
 ) *Test {
+	if !t.Spec().(*registry.TestSpec).Monitor {
+		t.Fatal("mixedversion tests require enabling the global test monitor in the test spec")
+	}
+
 	opts := defaultTestOptions()
 	for _, fn := range options {
 		fn(&opts)
@@ -858,7 +862,7 @@ func (t *Test) Run() {
 }
 
 func (t *Test) run(plan *TestPlan) error {
-	return newTestRunner(t.ctx, t.cancel, plan, t.options.tag, t.logger, t.cluster).run()
+	return newTestRunner(t.ctx, t.cancel, plan, t.rt, t.options.tag, t.logger, t.cluster).run()
 }
 
 func (t *Test) plan() (plan *TestPlan, retErr error) {

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

@@ -14,7 +14,6 @@ import (
 	"path/filepath"
 	"regexp"
 	"strings"
-	"sync"
 	"sync/atomic"
 	"text/tabwriter"
 	"time"
@@ -25,6 +24,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/registry"
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/roachtestutil/clusterupgrade"
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/roachtestutil/task"
+	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/test"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/roachprod/install"
 	"github.com/cockroachdb/cockroach/pkg/roachprod/logger"
@@ -33,23 +33,9 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/errors"
-	"golang.org/x/exp/maps"
 )
 
 type (
-	// crdbMonitor is a thin wrapper around the roachtest monitor API
-	// (cluster.NewMonitor) that produces error events through a channel
-	// whenever an unexpected node death happens. It also allows us to
-	// provide an API for test authors to inform the framework that a
-	// node death is expected if the test performs its own restarts or
-	// chaos events.
-	crdbMonitor struct {
-		once      sync.Once
-		crdbNodes option.NodeListOption
-		monitor   cluster.Monitor
-		errCh     chan error
-	}
-
 	serviceRuntime struct {
 		descriptor      *ServiceDescriptor
 		binaryVersions  *atomic.Value
@@ -72,7 +58,7 @@ type (
 		logger        *logger.Logger
 
 		background task.Manager
-		monitor    *crdbMonitor
+		monitor    test.Monitor
 
 		// ranUserHooks keeps track of whether the runner has run any
 		// user-provided hooks so far.
@@ -111,6 +97,7 @@ func newTestRunner(
 	ctx context.Context,
 	cancel context.CancelFunc,
 	plan *TestPlan,
+	rt test.Test,
 	tag string,
 	l *logger.Logger,
 	c cluster.Cluster,
@@ -142,7 +129,7 @@ func newTestRunner(
 		tenantService: tenantService,
 		cluster:       c,
 		background:    task.NewManager(ctx, l),
-		monitor:       newCRDBMonitor(ctx, c, maps.Keys(allCRDBNodes)),
+		monitor:       rt.Monitor(),
 		ranUserHooks:  &ranUserHooks,
 	}
 }
@@ -193,9 +180,6 @@ func (tr *testRunner) run() (retErr error) {
 			}
 
 			return fmt.Errorf("background step `%s` returned error: %w", event.Name, event.Err)
-
-		case err := <-tr.monitor.Err():
-			return tr.testFailure(tr.ctx, err, tr.logger, nil)
 		}
 	}
 }
@@ -388,11 +372,6 @@ func (tr *testRunner) teardown(stepsChan chan error, testFailed bool) {
 	tr.logger.Printf("stopping background functions")
 	tr.background.Terminate(tr.logger)
 
-	tr.logger.Printf("stopping node monitor")
-	if err := tr.monitor.Stop(); err != nil {
-		tr.logger.Printf("monitor returned error: %v", err)
-	}
-
 	// If the test failed, we wait for any currently running steps to
 	// return before passing control back to the roachtest
 	// framework. This achieves a test.log that does not contain any
@@ -614,22 +593,6 @@ func (tr *testRunner) refreshServiceData(ctx context.Context, service *serviceRu
 		return err
 	}
 
-	// We only want to start the monitor once we know every relevant
-	// cockroach binary is running. This is due to a limitation on the
-	// roachprod monitor: it is only able to monitor cockroach processes
-	// that are running at the time the monitor is created.
-	//
-	// For system-only and separate-process deployments, we can
-	// initialize the monitor right away, since this function is only
-	// called once the storage cluster is running. For separate-process
-	// deployments, we start the monitor if this function is called with
-	// the tenant service. The system is always started first, so when
-	// this function is called with the tenant service, we know that
-	// every relevant cockroach binary is running at this point.
-	if tr.plan.deploymentMode != SeparateProcessDeployment || !isSystem {
-		tr.monitor.Init()
-	}
-
 	return nil
 }
 
@@ -749,46 +712,6 @@ func (tr *testRunner) addGrafanaAnnotation(
 	return tr.cluster.AddGrafanaAnnotation(ctx, l, req)
 }
 
-func newCRDBMonitor(
-	ctx context.Context, c cluster.Cluster, crdbNodes option.NodeListOption,
-) *crdbMonitor {
-	return &crdbMonitor{
-		crdbNodes: crdbNodes,
-		monitor:   c.NewMonitor(ctx, crdbNodes),
-		errCh:     make(chan error),
-	}
-}
-
-// Init must be called once the cluster is initialized and the
-// cockroach process is running on the nodes. Init is idempotent.
-func (cm *crdbMonitor) Init() {
-	cm.once.Do(func() {
-		go func() {
-			if err := cm.monitor.WaitForNodeDeath(); err != nil {
-				cm.errCh <- err
-			}
-		}()
-	})
-}
-
-// Err returns a channel that will receive errors whenever an
-// unexpected node death is observed.
-func (cm *crdbMonitor) Err() chan error {
-	return cm.errCh
-}
-
-func (cm *crdbMonitor) ExpectDeaths(n int) {
-	cm.monitor.ExpectDeaths(int32(n))
-}
-
-func (cm *crdbMonitor) Stop() error {
-	if cm.monitor == nil { // test-only
-		return nil
-	}
-
-	return cm.monitor.WaitE()
-}
-
 // tableWriter is a thin wrapper around the `tabwriter` package used
 // by the test runner to display logical and released binary versions
 // in a tabular format.

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

@@ -119,10 +119,6 @@ func Test_run(t *testing.T) {
 				startSystemID: 9999,
 			}
 
-			runnerCh := make(chan error)
-			defer close(runnerCh)
-			runner.monitor = &crdbMonitor{errCh: runnerCh}
-
 			runErr := runner.run()
 			require.Error(t, runErr)
 

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

@@ -227,7 +227,6 @@ func (s restartVirtualClusterStep) Run(
 	l.Printf("restarting node %d (tenant %s) into version %s", s.node, s.virtualCluster, s.version)
 	node := h.runner.cluster.Node(s.node)
 
-	h.ExpectDeath()
 	stopOpts := option.StopVirtualClusterOpts(s.virtualCluster, node, option.Graceful(maxWait))
 	if err := h.runner.cluster.StopServiceForVirtualClusterE(ctx, l, stopOpts); err != nil {
 		return errors.Wrap(err, "failed to stop cockroach process for tenant")
@@ -359,13 +358,13 @@ func (s restartWithNewBinaryStep) Run(
 		install.TagOption(systemTag),
 	}, s.settings...)
 
-	h.ExpectDeath()
+	node := h.runner.cluster.Node(s.node)
 	if err := clusterupgrade.RestartNodesWithNewBinary(
 		startCtx,
 		s.rt,
 		l,
 		h.runner.cluster,
-		h.runner.cluster.Node(s.node),
+		node,
 		startOpts(customStartOpts...),
 		s.version,
 		settings...,
@@ -377,7 +376,7 @@ func (s restartWithNewBinaryStep) Run(
 		// If we are in shared-process mode and the tenant is already
 		// running at this point, we wait for the server on the restarted
 		// node to be up before moving on.
-		return waitForTenantProcess(ctx, l, h, h.runner.cluster.Node(s.node), s.deploymentMode)
+		return waitForTenantProcess(ctx, l, h, node, s.deploymentMode)
 	}
 
 	return nil
@@ -815,7 +814,7 @@ func (s panicNodeStep) Run(ctx context.Context, l *logger.Logger, rng *rand.Rand
 	)
 	customStartOpts := restartSystemSettings(true, s.initTarget)
 
-	h.ExpectDeath()
+	h.runner.monitor.ExpectProcessDead(s.targetNode)
 
 	const stmt = "SELECT crdb_internal.force_panic('expected panic from panicNodeMutator')"
 	err = h.System.ExecWithGateway(

pkg/cmd/roachtest/tests/acceptance.go

@@ -29,6 +29,7 @@ func registerAcceptance(r registry.Registry) {
 		timeout            time.Duration
 		encryptionSupport  registry.EncryptionSupport
 		defaultLeases      bool
+		monitor            bool
 		randomized         bool
 		workloadNode       bool
 		incompatibleClouds registry.CloudSet
@@ -75,6 +76,7 @@ func registerAcceptance(r registry.Registry) {
 				fn:            runVersionUpgrade,
 				timeout:       2 * time.Hour, // actually lower in local runs; see `runVersionUpgrade`
 				defaultLeases: true,
+				monitor:       true,
 				randomized:    true,
 				suites:        []string{registry.MixedVersion},
 				// Disabled on IBM because s390x is only built on master
@@ -103,6 +105,7 @@ func registerAcceptance(r registry.Registry) {
 				fn:            runValidateSystemSchemaAfterVersionUpgrade,
 				timeout:       60 * time.Minute,
 				defaultLeases: true,
+				monitor:       true,
 				randomized:    true,
 				numNodes:      1,
 				suites:        []string{registry.MixedVersion},
@@ -157,6 +160,7 @@ func registerAcceptance(r registry.Registry) {
 				Timeout:           10 * time.Minute,
 				CompatibleClouds:  registry.AllClouds.Remove(tc.incompatibleClouds),
 				Suites:            registry.Suites(suites...),
+				Monitor:           tc.monitor,
 				Randomized:        tc.randomized,
 			}
 

pkg/cmd/roachtest/tests/admission_control_elastic_mixed_version.go

@@ -39,6 +39,7 @@ func registerElasticWorkloadMixedVersion(r registry.Registry) {
 		Owner:            registry.OwnerKV,
 		Timeout:          3 * time.Hour,
 		Benchmark:        true,
+		Monitor:          true,
 		CompatibleClouds: registry.OnlyGCE,
 		Suites:           registry.Suites(registry.MixedVersion, registry.Nightly),
 		Cluster: r.MakeClusterSpec(4, spec.CPU(8),

pkg/cmd/roachtest/tests/db_console_cypress.go

@@ -206,6 +206,7 @@ func registerDbConsoleCypress(r registry.Registry) {
 		// is impossible to test as of 05/2025.
 		CompatibleClouds: registry.AllClouds.NoIBM(),
 		Suites:           registry.Suites(registry.MixedVersion, registry.Nightly),
+		Monitor:          true,
 		Randomized:       false,
 		Run:              runDbConsoleCypressMixedVersions,
 		Timeout:          2 * time.Hour,

pkg/cmd/roachtest/tests/db_console_endpoints.go

@@ -100,6 +100,7 @@ func registerDBConsoleEndpointsMixedVersion(r registry.Registry) {
 		Cluster:          r.MakeClusterSpec(5, spec.WorkloadNode()),
 		CompatibleClouds: registry.AllClouds,
 		Suites:           registry.Suites(registry.MixedVersion, registry.Nightly),
+		Monitor:          true,
 		Randomized:       true,
 		Run:              runDBConsoleMixedVersion,
 		Timeout:          1 * time.Hour,

pkg/cmd/roachtest/tests/decommission.go

@@ -119,6 +119,8 @@ func registerDecommission(r registry.Registry) {
 			Cluster:          r.MakeClusterSpec(numNodes),
 			CompatibleClouds: registry.AllClouds.NoAWS().NoIBM(),
 			Suites:           registry.Suites(registry.MixedVersion, registry.Nightly),
+			Monitor:          true,
+			Randomized:       true,
 			Run: func(ctx context.Context, t test.Test, c cluster.Cluster) {
 				runDecommissionMixedVersions(ctx, t, c)
 			},