Merge #150157 #150192

150157: roachtest: validate number of tables in database for fingerprinting r=msbutler a=kev-cao

Continuing to get failures with missing tables in the backup fixtures, this commit adds a stricter validation to our backup fixtures to ensure that the number of tables are being fingerprinted. This should prevent invalid fingerprints and fixtures from being created, which should reduce our test failures on restore and isolate the failures to the real culprit, the fixture tests.

Informs: #149671

Fixes: #149772, #149773, #149769

Release note: None

150192: asim: follow up for #149904 r=tbg a=wenyihu6


**asim: use cluster setting LBRebalancingMode directly**

Previously, LBRebalancingMode existed both as a simulation setting and a cluster
setting, which was confusing especially now that we're plumbing the cluster
setting through. This commit removes the simulation specific setting and makes
the simulator use the cluster setting directly instead. Note that default
behavior is leases mode and replicas=2 remains unchanged for both.

Epic: none
Release note: none

---

**mmaprototypehelpers: skip mma coordination when mma is disabled**

Previously, non-mma components registered external changes with mma via
allocator sync even when mma was disabled. This led to logs like `did not
register external changes: due to range does not exist in cluster state`.
Without mma being enabled, mma can’t register external changes because it never
learns about the ranges via processStoreLeaseholderMsg in the first place, and
this external change would get dropped. This commit changes allocator sync to
skip coordinating with the mma allocator when mma is disabled. Note that we
changes still go via allocator sync even when mma is disabled to update the
local store pool correctly.

Epic: none
Release note: none

---

**asim: add more data driven tests**

This commit adds more data-driven tests. Output looks the same as
49d07ce. Note that long-running test files
below are skipped in normal CI using the skip_under_ci datadriven command but
will still run under the extra stress pipeline.

```
example_liveness
example_fulldisk
mma_high_cpu_25nodes
mma_skewed_cpu_skewed_write_more_ranges
mma_skewed_cpu_skewed_write_all_enabled
mma_one_voter_skewed_cpu_skewed_write
mma_skewed_cpu_skewed_write
mma_high_write_uniform_cpu_all_enabled
mma_high_write_uniform_cpu
example_skewed_cpu_even_ranges_mma_and_queues
mma_constraint_satisfaction1
mma_high_cpu
example_splitting
example_rebalancing
example_skewed_cpu_even_ranges_sma_and_queues
example_skewed_cpu_even_ranges_mma
example_skewed_cpu_even_ranges_sma
example_io_overload
mma_constraint_satisfaction_old_alloc
example_zone_config
```

Epic: none
Release note: none


Co-authored-by: Kevin Cao <[email protected]>
Co-authored-by: wenyihu6 <[email protected]>

Author: 3 people

pkg/cmd/roachtest/tests/backup_fixtures.go

@@ -49,6 +49,13 @@ import (
 // and rely on the managed identity to authenticate with Azure Blob Storage.
 const azureCredentialsFilePath = "/home/ubuntu/azure-credentials.yaml"
 
+// Maps a fixture database name to the expected number of tables in the
+// database, useful for verifying that the fingerprint of the fixture is as
+// expected.
+var expectedNumTables = map[string]int{
+	"tpcc": 9,
+}
+
 type BackupFixture interface {
 	Kind() string
 	// The database that is backed up.
@@ -468,6 +475,7 @@ func fingerprintDatabase(
 		t.L().Printf("no tables found in database %s", dbName)
 		return nil
 	}
+	require.Len(t, tables, expectedNumTables[dbName], "unexpected number of tables in database %s", dbName)
 	t.L().Printf("fingerprinting %d tables in database %s", len(tables), dbName)
 
 	fingerprints := make(map[string]string)
@@ -493,6 +501,10 @@ func fingerprintDatabase(
 		"fingerprinted %d tables in %s in %s",
 		len(tables), dbName, timeutil.Since(start),
 	)
+	require.Len(
+		t, fingerprints, expectedNumTables[dbName],
+		"unexpected number of fingerprints for database %s", dbName,
+	)
 	return fingerprints
 }
 
@@ -513,6 +525,7 @@ func getDatabaseTables(t test.Test, sql *sqlutils.SQLRunner, db string) []string
 		}
 		tables = append(tables, fmt.Sprintf(`%s.%s.%s`, db, schemaName, tableName))
 	}
+	require.NoError(t, rows.Err(), "error iterating over tables in database %s", db)
 	return tables
 }
 

pkg/kv/kvserver/allocator/mmaprototypehelpers/BUILD.bazel

@@ -10,10 +10,12 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//pkg/kv/kvpb",
+        "//pkg/kv/kvserver",
         "//pkg/kv/kvserver/allocator",
         "//pkg/kv/kvserver/allocator/mmaprototype",
         "//pkg/kv/kvserver/allocator/storepool",
         "//pkg/roachpb",
+        "//pkg/settings/cluster",
         "//pkg/util/log",
         "//pkg/util/syncutil",
         "//pkg/util/timeutil",

pkg/kv/kvserver/allocator/mmaprototypehelpers/kvserver_mma_integration.go

@@ -10,10 +10,12 @@ import (
 	"fmt"
 
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/allocator"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/allocator/mmaprototype"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/allocator/storepool"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
 )
@@ -62,17 +64,21 @@ func (s Author) External() bool {
 type AllocatorSync struct {
 	sp          *storepool.StorePool
 	mmAllocator mmaprototype.Allocator
+	st          *cluster.Settings
 	mu          struct {
 		syncutil.Mutex
 		changeSeqGen   SyncChangeID
 		trackedChanges map[SyncChangeID]trackedAllocatorChange
 	}
 }
 
-func NewAllocatorSync(sp *storepool.StorePool, mmAllocator mmaprototype.Allocator) *AllocatorSync {
+func NewAllocatorSync(
+	sp *storepool.StorePool, mmAllocator mmaprototype.Allocator, st *cluster.Settings,
+) *AllocatorSync {
 	as := &AllocatorSync{
 		sp:          sp,
 		mmAllocator: mmAllocator,
+		st:          st,
 	}
 	as.mu.trackedChanges = make(map[SyncChangeID]trackedAllocatorChange)
 	return as
@@ -122,9 +128,12 @@ func (as *AllocatorSync) NonMMAPreTransferLease(
 	)
 	log.Infof(ctx, "registering external lease transfer change: usage=%v changes=%v",
 		usage, replicaChanges)
-	changeIDs := as.mmAllocator.RegisterExternalChanges(replicaChanges[:])
-	if changeIDs == nil {
-		log.Info(ctx, "mma did not track lease transfer, skipping")
+	var changeIDs []mmaprototype.ChangeID
+	if kvserver.LoadBasedRebalancingMode.Get(&as.st.SV) == kvserver.LBRebalancingMultiMetric {
+		changeIDs = as.mmAllocator.RegisterExternalChanges(replicaChanges[:])
+		if changeIDs == nil {
+			log.Info(ctx, "mma did not track lease transfer, skipping")
+		}
 	}
 	trackedChange := trackedAllocatorChange{
 		typ:          AllocatorChangeTypeLeaseTransfer,
@@ -210,9 +219,12 @@ func (as *AllocatorSync) NonMMAPreChangeReplicas(
 
 	log.Infof(ctx, "registering external replica change: chgs=%v usage=%v changes=%v",
 		changes, usage, replicaChanges)
-	changeIDs := as.mmAllocator.RegisterExternalChanges(replicaChanges)
-	if changeIDs == nil {
-		log.Info(ctx, "cluster does not have a range for the external replica change, skipping")
+	var changeIDs []mmaprototype.ChangeID
+	if kvserver.LoadBasedRebalancingMode.Get(&as.st.SV) == kvserver.LBRebalancingMultiMetric {
+		changeIDs = as.mmAllocator.RegisterExternalChanges(replicaChanges)
+		if changeIDs == nil {
+			log.Info(ctx, "cluster does not have a range for the external replica change, skipping")
+		}
 	}
 	trackedChange := trackedAllocatorChange{
 		typ:       AllocatorChangeTypeChangeReplicas,
@@ -335,12 +347,14 @@ func (as *AllocatorSync) PostApply(ctx context.Context, syncChangeID SyncChangeI
 		}
 		delete(as.mu.trackedChanges, syncChangeID)
 	}()
-	if changeIDs := tracked.changeIDs; changeIDs != nil {
-		log.Infof(ctx, "PostApply: tracked=%v change_ids=%v success: %v", tracked, changeIDs, success)
-		as.updateMetrics(success, tracked.typ, tracked.author)
-		as.mmAllocator.AdjustPendingChangesDisposition(changeIDs, success)
-	} else {
-		log.Infof(ctx, "PostApply: tracked=%v no change_ids success: %v", tracked, success)
+	if kvserver.LoadBasedRebalancingMode.Get(&as.st.SV) == kvserver.LBRebalancingMultiMetric {
+		if changeIDs := tracked.changeIDs; changeIDs != nil {
+			log.Infof(ctx, "PostApply: tracked=%v change_ids=%v success: %v", tracked, changeIDs, success)
+			as.updateMetrics(success, tracked.typ, tracked.author)
+			as.mmAllocator.AdjustPendingChangesDisposition(changeIDs, success)
+		} else {
+			log.Infof(ctx, "PostApply: tracked=%v no change_ids success: %v", tracked, success)
+		}
 	}
 	as.updateMetrics(success, tracked.typ, tracked.author)
 	if !success {

pkg/kv/kvserver/asim/config/settings.go

@@ -27,7 +27,6 @@ const (
 	defaultSplitQPSThreshold       = 2500
 	defaultSplitStatRetention      = 10 * time.Minute
 	defaultSeed                    = 42
-	defaultLBRebalancingMode       = 2 // Leases and replicas.
 	defaultLBRebalancingInterval   = time.Minute
 	defaultLBRebalanceQPSThreshold = 0.1
 	defaultLBRebalancingObjective  = 0 // QPS
@@ -96,9 +95,6 @@ type SimulationSettings struct {
 	// SplitStatRetention is the duration which recorded load will be retained
 	// and factored into load based splitting decisions.
 	SplitStatRetention time.Duration
-	// LBRebalancingMode controls if and when we do store-level rebalancing
-	// based on load. It maps to kvserver.LBRebalancingMode.
-	LBRebalancingMode int64
 	// LBRebalancingObjective is the load objective to balance.
 	LBRebalancingObjective int64
 	// LBRebalancingInterval controls how often the store rebalancer will
@@ -136,7 +132,6 @@ func DefaultSimulationSettings() *SimulationSettings {
 		StateExchangeDelay:      defaultStateExchangeDelay,
 		SplitQPSThreshold:       defaultSplitQPSThreshold,
 		SplitStatRetention:      defaultSplitStatRetention,
-		LBRebalancingMode:       defaultLBRebalancingMode,
 		LBRebalancingObjective:  defaultLBRebalancingObjective,
 		LBRebalancingInterval:   defaultLBRebalancingInterval,
 		ReplicateQueueEnabled:   true,

pkg/kv/kvserver/asim/event/mutation_event.go

@@ -54,8 +54,9 @@ type SetNodeLocalityEvent struct {
 // SetSimulationSettingsEvent represents a mutation event responsible for
 // changing a simulation setting during the simulation.
 type SetSimulationSettingsEvent struct {
-	Key   string
-	Value interface{}
+	IsClusterSetting bool
+	Key              string
+	Value            interface{}
 }
 
 var _ Event = &SetSpanConfigEvent{}
@@ -140,7 +141,11 @@ func (sne SetNodeLocalityEvent) String() string {
 
 func (se SetSimulationSettingsEvent) Func() EventFunc {
 	return MutationFunc(func(ctx context.Context, s state.State) {
-		s.SetSimulationSettings(se.Key, se.Value)
+		if se.IsClusterSetting {
+			s.SetClusterSetting(se.Key, se.Value)
+		} else {
+			s.SetSimulationSettings(se.Key, se.Value)
+		}
 	})
 }
 

pkg/kv/kvserver/asim/mmaintegration/mma_store_rebalancer.go

@@ -97,7 +97,7 @@ func (msr *MMAStoreRebalancer) Tick(ctx context.Context, tick time.Time, s state
 		return
 	}
 
-	if msr.settings.LBRebalancingMode != int64(kvserver.LBRebalancingMultiMetric) {
+	if kvserver.LoadBasedRebalancingMode.Get(&msr.settings.ST.SV) != kvserver.LBRebalancingMultiMetric {
 		// When the store rebalancer isn't set to use the multi-metric mode, the
 		// legacy store rebalancer is used.
 		return

pkg/kv/kvserver/asim/state/impl.go

@@ -430,9 +430,9 @@ func (s *state) AddNode() Node {
 		nodeID:      nodeID,
 		desc:        roachpb.NodeDescriptor{NodeID: roachpb.NodeID(nodeID)},
 		stores:      []StoreID{},
-		storepool:   sp,
 		mmAllocator: mmAllocator,
-		as:          mmaprototypehelpers.NewAllocatorSync(sp, mmAllocator),
+		storepool:   sp,
+		as:          mmaprototypehelpers.NewAllocatorSync(sp, mmAllocator, s.settings.ST),
 	}
 	s.nodes[nodeID] = node
 	s.SetNodeLiveness(nodeID, livenesspb.NodeLivenessStatus_LIVE)
@@ -1393,6 +1393,15 @@ func (s *state) RegisterConfigChangeListener(listener ConfigChangeListener) {
 	s.configChangeListeners = append(s.configChangeListeners, listener)
 }
 
+func (s *state) SetClusterSetting(Key string, Value interface{}) {
+	switch Key {
+	case "LBRebalancingMode":
+		kvserver.LoadBasedRebalancingMode.Override(context.Background(), &s.settings.ST.SV, kvserver.LBRebalancingMode(Value.(int64)))
+	default:
+		panic("other cluster settings not supported")
+	}
+}
+
 // SetSimulationSettings sets the simulation setting for the given key to the
 // given value.
 func (s *state) SetSimulationSettings(Key string, Value interface{}) {

pkg/kv/kvserver/asim/state/state.go

@@ -202,6 +202,8 @@ type State interface {
 	RegisterConfigChangeListener(ConfigChangeListener)
 	// SetSimulationSettings sets the simulation settings for the state.
 	SetSimulationSettings(Key string, Value interface{})
+	// SetClusterSetting sets the cluster setting for the state.
+	SetClusterSetting(Key string, Value interface{})
 	// NodeCapacity returns the capacity of the node with ID NodeID.
 	NodeCapacity(NodeID) roachpb.NodeCapacity
 	// SetNodeCPURateCapacity sets the CPU rate capacity for the node with ID

pkg/kv/kvserver/asim/storerebalancer/store_rebalancer.go

@@ -190,7 +190,7 @@ func (src *storeRebalancerControl) phasePrologue(
 			s, src.storeID,
 			kvserver.LBRebalancingObjective(src.settings.LBRebalancingObjective).ToDimension(),
 		),
-		kvserver.LBRebalancingMode(src.settings.LBRebalancingMode),
+		kvserver.LoadBasedRebalancingMode.Get(&src.settings.ST.SV),
 	)
 
 	if !src.sr.ShouldRebalanceStore(ctx, rctx) {

pkg/kv/kvserver/asim/tests/BUILD.bazel

@@ -38,6 +38,7 @@ go_test(
     data = glob(["testdata/**"]),
     embed = [":tests"],
     deps = [
+        "//pkg/kv/kvserver",
         "//pkg/kv/kvserver/allocator/allocatorimpl",
         "//pkg/kv/kvserver/asim/assertion",
         "//pkg/kv/kvserver/asim/config",