diff --git a/pkg/kv/kvserver/asim/asim.go b/pkg/kv/kvserver/asim/asim.go
index 6c239161747f..878eafdab266 100644
--- a/pkg/kv/kvserver/asim/asim.go
+++ b/pkg/kv/kvserver/asim/asim.go
@@ -68,6 +68,7 @@ type Simulator struct {
 // TODO(kvoli): Add a range log like structure to the history.
 type History struct {
 	Recorded [][]metrics.StoreMetrics
+	S        state.State
 }
 
 // Listen implements the metrics.StoreMetricListener interface.
@@ -109,7 +110,7 @@ func NewSimulator(
 		shuffler:       state.NewShuffler(settings.Seed),
 		// TODO(kvoli): Keeping the state around is a bit hacky, find a better
 		// method of reporting the ranges.
-		history:  History{Recorded: [][]metrics.StoreMetrics{}},
+		history:  History{Recorded: [][]metrics.StoreMetrics{}, S: initialState},
 		events:   events,
 		settings: settings,
 	}
@@ -121,6 +122,7 @@ func NewSimulator(
 	s.state.RegisterConfigChangeListener(s)
 
 	m.Register(&s.history)
+	s.AddLogTag("asim", nil)
 	return s
 }
 
diff --git a/pkg/kv/kvserver/asim/asim_test.go b/pkg/kv/kvserver/asim/asim_test.go
index db05c9ba2f09..0e786eeadbb9 100644
--- a/pkg/kv/kvserver/asim/asim_test.go
+++ b/pkg/kv/kvserver/asim/asim_test.go
@@ -87,6 +87,6 @@ func TestAllocatorSimulatorDeterministic(t *testing.T) {
 			refRun = history
 			continue
 		}
-		require.Equal(t, refRun, history)
+		require.Equal(t, refRun.Recorded, history.Recorded)
 	}
 }
diff --git a/pkg/kv/kvserver/asim/gen/BUILD.bazel b/pkg/kv/kvserver/asim/gen/BUILD.bazel
index 5b47af5fcc6d..01e79229a76b 100644
--- a/pkg/kv/kvserver/asim/gen/BUILD.bazel
+++ b/pkg/kv/kvserver/asim/gen/BUILD.bazel
@@ -8,6 +8,7 @@ go_library(
     deps = [
         "//pkg/kv/kvserver/asim",
         "//pkg/kv/kvserver/asim/config",
+        "//pkg/kv/kvserver/asim/event",
         "//pkg/kv/kvserver/asim/metrics",
         "//pkg/kv/kvserver/asim/state",
         "//pkg/kv/kvserver/asim/workload",
diff --git a/pkg/kv/kvserver/asim/gen/generator.go b/pkg/kv/kvserver/asim/gen/generator.go
index 8ca1a1ac4066..5d2705b061a0 100644
--- a/pkg/kv/kvserver/asim/gen/generator.go
+++ b/pkg/kv/kvserver/asim/gen/generator.go
@@ -12,10 +12,12 @@ package gen
 
 import (
 	"math/rand"
+	"sort"
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/asim"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/asim/config"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/asim/event"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/asim/metrics"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/asim/state"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/asim/workload"
@@ -36,26 +38,53 @@ type LoadGen interface {
 	Generate(seed int64, settings *config.SimulationSettings) []workload.Generator
 }
 
-// StateGen provides a method to generate a state given a seed and simulation
-// settings.
-type StateGen interface {
-	// Generate returns a state that is parameterized randomly by the seed and
-	// simulation settings provided.
+// ClusterGen provides a method to generate the initial cluster state,  given a
+// seed and simulation settings. The initial cluster state includes: nodes
+// (including locality) and stores.
+type ClusterGen interface {
+	// Generate returns a new State that is parameterized randomly by the seed
+	// and simulation settings provided.
 	Generate(seed int64, settings *config.SimulationSettings) state.State
 }
 
+// RangeGen provides a method to generate the initial range splits, range
+// replica and lease placement within a cluster.
+type RangeGen interface {
+	// Generate returns an updated state, given the initial state, seed and
+	// simulation settings provided. In the updated state, ranges will have been
+	// created, replicas and leases assigned to stores in the cluster.
+	Generate(seed int64, settings *config.SimulationSettings, s state.State) state.State
+}
+
+// EventGen provides a  method to generate a list of events that will apply to
+// the simulated cluster. Currently, only delayed (fixed time) events are
+// supported.
+type EventGen interface {
+	// Generate returns a list of events, which should be exectued at the delay specified.
+	Generate(seed int64) event.DelayedEventList
+}
+
 // GenerateSimulation is a utility function that creates a new allocation
 // simulation using the provided state, workload, settings generators and seed.
 func GenerateSimulation(
-	duration time.Duration, stateGen StateGen, loadGen LoadGen, settingsGen SettingsGen, seed int64,
+	duration time.Duration,
+	clusterGen ClusterGen,
+	rangeGen RangeGen,
+	loadGen LoadGen,
+	settingsGen SettingsGen,
+	eventGen EventGen,
+	seed int64,
 ) *asim.Simulator {
 	settings := settingsGen.Generate(seed)
+	s := clusterGen.Generate(seed, &settings)
+	s = rangeGen.Generate(seed, &settings, s)
 	return asim.NewSimulator(
 		duration,
 		loadGen.Generate(seed, &settings),
-		stateGen.Generate(seed, &settings),
+		s,
 		&settings,
 		metrics.NewTracker(settings.MetricsInterval),
+		eventGen.Generate(seed)...,
 	)
 }
 
@@ -89,6 +118,10 @@ type BasicLoad struct {
 // SkewedAccess is true. The returned workload generators are seeded with the
 // provided seed.
 func (bl BasicLoad) Generate(seed int64, settings *config.SimulationSettings) []workload.Generator {
+	if bl.Rate == 0 {
+		return []workload.Generator{}
+	}
+
 	var keyGen workload.KeyGenerator
 	rand := rand.New(rand.NewSource(seed))
 	if bl.SkewedAccess {
@@ -110,25 +143,94 @@ func (bl BasicLoad) Generate(seed int64, settings *config.SimulationSettings) []
 	}
 }
 
-// BasicState implements the StateGen interface.
-type BasicState struct {
-	Stores            int
+// LoadedCluster implements the ClusterGen interface.
+type LoadedCluster struct {
+	Info state.ClusterInfo
+}
+
+// Generate returns a new simulator state, where the cluster is loaded based on
+// the cluster info the loaded cluster generator is created with. There is no
+// randomness in this cluster generation.
+func (lc LoadedCluster) Generate(seed int64, settings *config.SimulationSettings) state.State {
+	return state.LoadClusterInfo(lc.Info, settings)
+}
+
+// BasicCluster implements the ClusterGen interace.
+type BasicCluster struct {
+	Nodes         int
+	StoresPerNode int
+}
+
+// Generate returns a new simulator state, where the cluster is created with all
+// nodes having the same locality and with the specified number of stores/nodes
+// created. The cluster is created based on the stores and stores-per-node
+// values the basic cluster generator is created with.
+func (lc BasicCluster) Generate(seed int64, settings *config.SimulationSettings) state.State {
+	info := state.ClusterInfoWithStoreCount(lc.Nodes, lc.StoresPerNode)
+	return state.LoadClusterInfo(info, settings)
+}
+
+// LoadedRanges implements the RangeGen interface.
+type LoadedRanges struct {
+	Info state.RangesInfo
+}
+
+// Generate returns an updated simulator state, where the cluster is loaded
+// with the range info that the generator was created with. There is no
+// randomness in this cluster generation.
+func (lr LoadedRanges) Generate(
+	seed int64, settings *config.SimulationSettings, s state.State,
+) state.State {
+	state.LoadRangeInfo(s, lr.Info...)
+	return s
+}
+
+// PlacementType represents a type of placement distribution.
+type PlacementType int
+
+const (
+	Uniform PlacementType = iota
+	Skewed
+)
+
+// BasicRanges implements the RangeGen interface.
+type BasicRanges struct {
 	Ranges            int
-	SkewedPlacement   bool
+	PlacementType     PlacementType
 	KeySpace          int
 	ReplicationFactor int
+	Bytes             int64
 }
 
-// Generate returns a new state that is created with the number of stores,
-// ranges, keyspace and replication factor from the basic state fields. The
-// initial assignment of replicas and leases for ranges follows either a
-// uniform or powerlaw distribution depending on if SkewedPlacement is true.
-func (bs BasicState) Generate(seed int64, settings *config.SimulationSettings) state.State {
-	var s state.State
-	if bs.SkewedPlacement {
-		s = state.NewStateSkewedDistribution(bs.Stores, bs.Ranges, bs.ReplicationFactor, bs.KeySpace, settings)
-	} else {
-		s = state.NewStateEvenDistribution(bs.Stores, bs.Ranges, bs.ReplicationFactor, bs.KeySpace, settings)
+// Generate returns an updated simulator state, where the cluster is loaded
+// with ranges based on the parameters of basic ranges.
+func (br BasicRanges) Generate(
+	seed int64, settings *config.SimulationSettings, s state.State,
+) state.State {
+	stores := len(s.Stores())
+	var rangesInfo state.RangesInfo
+	switch br.PlacementType {
+	case Uniform:
+		rangesInfo = state.RangesInfoEvenDistribution(stores, br.Ranges, br.KeySpace, br.ReplicationFactor, br.Bytes)
+	case Skewed:
+		rangesInfo = state.RangesInfoSkewedDistribution(stores, br.Ranges, br.KeySpace, br.ReplicationFactor, br.Bytes)
 	}
+	for _, rangeInfo := range rangesInfo {
+		rangeInfo.Size = br.Bytes
+	}
+	state.LoadRangeInfo(s, rangesInfo...)
 	return s
 }
+
+// StaticEvents implements the EventGen interface.
+// TODO(kvoli): introduce conditional events.
+type StaticEvents struct {
+	DelayedEvents event.DelayedEventList
+}
+
+// Generate returns a list of events, exactly the same as the events
+// StaticEvents was created with.
+func (se StaticEvents) Generate(seed int64) event.DelayedEventList {
+	sort.Sort(se.DelayedEvents)
+	return se.DelayedEvents
+}
diff --git a/pkg/kv/kvserver/asim/state/new_state.go b/pkg/kv/kvserver/asim/state/new_state.go
index c2b700b9559e..cee7959759b1 100644
--- a/pkg/kv/kvserver/asim/state/new_state.go
+++ b/pkg/kv/kvserver/asim/state/new_state.go
@@ -153,14 +153,14 @@ func RangesInfoWithDistribution(
 // their weights, a best effort apporach is taken so that the total number of
 // aggregate matches numNodes.
 func ClusterInfoWithDistribution(
-	numNodes int, storesPerNode int, regions []string, regionNodeWeights []float64,
+	nodeCount int, storesPerNode int, regions []string, regionNodeWeights []float64,
 ) ClusterInfo {
 	ret := ClusterInfo{}
 
 	ret.Regions = make([]Region, len(regions))
-	availableNodes := numNodes
+	availableNodes := nodeCount
 	for i, name := range regions {
-		allocatedNodes := int(float64(numNodes) * (regionNodeWeights[i]))
+		allocatedNodes := int(float64(nodeCount) * (regionNodeWeights[i]))
 		if allocatedNodes > availableNodes {
 			allocatedNodes = availableNodes
 		}
@@ -174,11 +174,11 @@ func ClusterInfoWithDistribution(
 	return ret
 }
 
-// ClusterInfoWithStores returns a new ClusterInfo with the specified number of
-// stores. There will be only one store per node and a single region and zone.
-func ClusterInfoWithStoreCount(stores int, storesPerNode int) ClusterInfo {
+// ClusterInfoWithStoreCount returns a new ClusterInfo with the specified number of
+// stores. There will be storesPerNode stores per node and a single region and zone.
+func ClusterInfoWithStoreCount(nodeCount int, storesPerNode int) ClusterInfo {
 	return ClusterInfoWithDistribution(
-		stores,
+		nodeCount,
 		storesPerNode,
 		[]string{"AU_EAST"}, /* regions */
 		[]float64{1},        /* regionNodeWeights */
diff --git a/pkg/kv/kvserver/asim/tests/BUILD.bazel b/pkg/kv/kvserver/asim/tests/BUILD.bazel
index 4a40b14662cd..c3b00577b998 100644
--- a/pkg/kv/kvserver/asim/tests/BUILD.bazel
+++ b/pkg/kv/kvserver/asim/tests/BUILD.bazel
@@ -7,11 +7,18 @@ go_test(
     data = glob(["testdata/**"]),
     embed = [":tests"],
     deps = [
+        "//pkg/kv/kvserver/allocator/allocatorimpl",
         "//pkg/kv/kvserver/asim",
         "//pkg/kv/kvserver/asim/config",
+        "//pkg/kv/kvserver/asim/event",
         "//pkg/kv/kvserver/asim/gen",
         "//pkg/kv/kvserver/asim/metrics",
+        "//pkg/kv/kvserver/asim/state",
+        "//pkg/kv/kvserver/liveness/livenesspb",
+        "//pkg/roachpb",
+        "//pkg/spanconfig/spanconfigtestutils",
         "//pkg/testutils/datapathutils",
+        "//pkg/util/log",
         "@com_github_cockroachdb_datadriven//:datadriven",
         "@com_github_guptarohit_asciigraph//:asciigraph",
         "@com_github_stretchr_testify//require",
@@ -26,6 +33,8 @@ go_library(
     deps = [
         "//pkg/kv/kvserver/asim",
         "//pkg/kv/kvserver/asim/metrics",
+        "//pkg/roachpb",
+        "//pkg/spanconfig/spanconfigtestutils",
         "//pkg/util/log",
         "@com_github_montanaflynn_stats//:stats",
     ],
diff --git a/pkg/kv/kvserver/asim/tests/assert.go b/pkg/kv/kvserver/asim/tests/assert.go
index a17d86092dec..85775775c3d9 100644
--- a/pkg/kv/kvserver/asim/tests/assert.go
+++ b/pkg/kv/kvserver/asim/tests/assert.go
@@ -18,6 +18,8 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/asim"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/asim/metrics"
+	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/spanconfig/spanconfigtestutils"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/montanaflynn/stats"
 )
@@ -181,3 +183,168 @@ func (ba balanceAssertion) String() string {
 		"balance stat=%s threshold=%.2f ticks=%d",
 		ba.stat, ba.threshold, ba.ticks)
 }
+
+type storeStatAssertion struct {
+	ticks         int
+	stat          string
+	stores        []int
+	acceptedValue float64
+}
+
+// Assert looks at a simulation run history and returns true if the
+// assertion holds and false if not. When the assertion does not hold, the
+// reason is also returned.
+func (sa storeStatAssertion) Assert(
+	ctx context.Context, h asim.History,
+) (holds bool, reason string) {
+	m := h.Recorded
+	ticks := len(m)
+	if sa.ticks > ticks {
+		log.VInfof(ctx, 2,
+			"The history to run assertions against (%d) is shorter than "+
+				"the assertion duration (%d)", ticks, sa.ticks)
+		return true, ""
+	}
+
+	ts := metrics.MakeTS(m)
+	statTs := ts[sa.stat]
+	holds = true
+	// Set holds to be true initially, holds is set to false if the steady
+	// state assertion doesn't hold on any store.
+	holds = true
+	buf := strings.Builder{}
+
+	for _, store := range sa.stores {
+		trimmedStoreStats := statTs[store-1][ticks-sa.ticks-1:]
+		for _, stat := range trimmedStoreStats {
+			if stat != sa.acceptedValue {
+				if holds {
+					holds = false
+					fmt.Fprintf(&buf, "  %s\n", sa)
+				}
+				fmt.Fprintf(&buf,
+					"\tstore=%d stat=%.2f\n",
+					store, stat)
+			}
+		}
+	}
+	return holds, buf.String()
+}
+
+// String returns the string representation of the assertion.
+func (sa storeStatAssertion) String() string {
+	return fmt.Sprintf("stat=%s value=%.2f ticks=%d",
+		sa.stat, sa.acceptedValue, sa.ticks)
+}
+
+type conformanceAssertion struct {
+	underreplicated int
+	overreplicated  int
+	violating       int
+	unavailable     int
+}
+
+// conformanceAssertionSentinel declares a sentinel value which when any of the
+// conformanceAssertion parameters are set to, we ignore the conformance
+// reports value for that type of conformance.
+const conformanceAssertionSentinel = -1
+
+// Assert looks at a simulation run history and returns true if the
+// assertion holds and false if not. When the assertion does not hold, the
+// reason is also returned.
+func (ca conformanceAssertion) Assert(
+	ctx context.Context, h asim.History,
+) (holds bool, reason string) {
+	report := h.S.Report()
+	buf := strings.Builder{}
+	holds = true
+
+	unavailable, under, over, violating := len(report.Unavailable), len(report.UnderReplicated), len(report.OverReplicated), len(report.ViolatingConstraints)
+
+	maybeInitHolds := func() {
+		if holds {
+			holds = false
+			fmt.Fprintf(&buf, "  %s\n", ca)
+			fmt.Fprintf(&buf, "  actual unavailable=%d under=%d, over=%d violating=%d\n",
+				unavailable, under, over, violating,
+			)
+		}
+	}
+
+	if ca.unavailable != conformanceAssertionSentinel &&
+		ca.unavailable != unavailable {
+		maybeInitHolds()
+		buf.WriteString(PrintSpanConfigConformanceList(
+			"unavailable", report.Unavailable))
+	}
+	if ca.underreplicated != conformanceAssertionSentinel &&
+		ca.underreplicated != under {
+		maybeInitHolds()
+		buf.WriteString(PrintSpanConfigConformanceList(
+			"under replicated", report.UnderReplicated))
+	}
+	if ca.overreplicated != conformanceAssertionSentinel &&
+		ca.overreplicated != over {
+		maybeInitHolds()
+		buf.WriteString(PrintSpanConfigConformanceList(
+			"over replicated", report.OverReplicated))
+	}
+	if ca.violating != conformanceAssertionSentinel &&
+		ca.violating != violating {
+		maybeInitHolds()
+		buf.WriteString(PrintSpanConfigConformanceList(
+			"violating constraints", report.ViolatingConstraints))
+	}
+
+	return holds, buf.String()
+}
+
+// String returns the string representation of the assertion.
+func (ca conformanceAssertion) String() string {
+	buf := strings.Builder{}
+	fmt.Fprintf(&buf, "conformance ")
+	if ca.unavailable != conformanceAssertionSentinel {
+		fmt.Fprintf(&buf, "unavailable=%d ", ca.unavailable)
+	}
+	if ca.underreplicated != conformanceAssertionSentinel {
+		fmt.Fprintf(&buf, "under=%d ", ca.underreplicated)
+	}
+	if ca.overreplicated != conformanceAssertionSentinel {
+		fmt.Fprintf(&buf, "over=%d ", ca.overreplicated)
+	}
+	if ca.violating != conformanceAssertionSentinel {
+		fmt.Fprintf(&buf, "violating=%d ", ca.violating)
+	}
+	return buf.String()
+}
+
+func printRangeDesc(r roachpb.RangeDescriptor) string {
+	var buf strings.Builder
+	buf.WriteString(fmt.Sprintf("r%d:", r.RangeID))
+	buf.WriteString(r.RSpan().String())
+	buf.WriteString(" [")
+	if allReplicas := r.Replicas().Descriptors(); len(allReplicas) > 0 {
+		for i, rep := range allReplicas {
+			if i > 0 {
+				buf.WriteString(", ")
+			}
+			buf.WriteString(rep.String())
+		}
+	} else {
+		buf.WriteString("<no replicas>")
+	}
+	buf.WriteString("]")
+	return buf.String()
+}
+
+func PrintSpanConfigConformanceList(tag string, ranges []roachpb.ConformanceReportedRange) string {
+	var buf strings.Builder
+	for i, r := range ranges {
+		if i == 0 {
+			buf.WriteString(fmt.Sprintf("%s:\n", tag))
+		}
+		buf.WriteString(fmt.Sprintf("  %s applying %s\n", printRangeDesc(r.RangeDescriptor),
+			spanconfigtestutils.PrintSpanConfigDiffedAgainstDefaults(r.Config)))
+	}
+	return buf.String()
+}
diff --git a/pkg/kv/kvserver/asim/tests/datadriven_simulation_test.go b/pkg/kv/kvserver/asim/tests/datadriven_simulation_test.go
index 2989358081ce..4087de818a57 100644
--- a/pkg/kv/kvserver/asim/tests/datadriven_simulation_test.go
+++ b/pkg/kv/kvserver/asim/tests/datadriven_simulation_test.go
@@ -19,11 +19,18 @@ import (
 	"testing"
 	"time"
 
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/allocator/allocatorimpl"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/asim"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/asim/config"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/asim/event"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/asim/gen"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/asim/metrics"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/asim/state"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/liveness/livenesspb"
+	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/spanconfig/spanconfigtestutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/datapathutils"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/datadriven"
 	"github.com/guptarohit/asciigraph"
 	"github.com/stretchr/testify/require"
@@ -42,14 +49,50 @@ import (
 //     simulation. The default values are: rw_ratio=0 rate=0 min_block=1
 //     max_block=1 min_key=1 max_key=10_000 access_skew=false.
 //
-//   - "gen_state" [stores=<int>] [ranges=<int>] [placement_skew=<bool>]
-//     [repl_factor=<int>] [keyspace=<int>]
-//     Initialize the state generator parameters. On the next call to eval, the
-//     state generator is called to create the initial state used in the
-//     simulation. The default values are: stores=3 ranges=1 repl_factor=3
+//   - "ken_cluster" [nodes=<int>] [stores_per_node=<int>]
+//     Initialize the cluster generator parameters. On the next call to eval,
+//     the cluster generator is called to create the initial state used in the
+//     simulation. The default values are: nodes=3 stores_per_node=1.
+//
+//   - "load_cluster": config=<name>
+//     Load a defined cluster configuration to be the generated cluster in the
+//     simulation. The available confiurations are: single_region: 15 nodes in
+//     region=US, 5 in each zone US_1/US_2/US_3. single_region_multi_store: 3
+//     nodes, 5 stores per node with the same zone/region configuration as
+//     above. multi_region: 36 nodes, 12 in each region and 4 in each zone,
+//     regions having 3 zones. complex: 28 nodes, 3 regions with a skewed
+//     number of nodes per region.
+//
+//   - "gen_ranges" [ranges=<int>] [placement_skew=<bool>] [repl_factor=<int>]
+//     [keyspace=<int>] [range_bytes=<int>]
+//     Initialize the range generator parameters. On the next call to eval, the
+//     range generator is called to assign an ranges and their replica
+//     placement. The default values are ranges=1 repl_factor=3
 //     placement_skew=false keyspace=10000.
 //
-//   - "assertion" type=<string> stat=<string> ticks=<int> threshold=<float>
+//   - set_liveness node=<int> [delay=<duration>]
+//     status=(dead|decommisssioning|draining|unavailable)
+//     Set the liveness status of the node with ID NodeID. This applies at the
+//     start of the simulation or with some delay after the simulation starts,
+//     if specified.
+//
+//   - add_node: [stores=<int>] [locality=<string>] [delay=<duration>]
+//     Add a node to the cluster after initial generation with some delay,
+//     locality and number of stores on the node. The default values are
+//     stores=0 locality=none delay=0.
+//
+//   - set_span_config [delay=<duration>]
+//     [startKey, endKey): <span_config> Provide a new line separated list
+//     of spans and span configurations e.g.
+//     [0,100): num_replicas=5 num_voters=3 constraints={'+region=US_East'}
+//     [100, 500): num_replicas=3
+//     ...
+//     This will update the span config for the span [0,100) to specify 3
+//     voting replicas and 2 non-voting replicas, with a constraint that all
+//     replicas are in the region US_East.
+//
+//   - "assertion" type=<string> [stat=<string>] [ticks=<int>] [threshold=<float>]
+//     [store=<int>] [(under|over|unavailable|violating)=<int>]
 //     Add an assertion to the list of assertions that run against each
 //     sample on subsequent calls to eval. When every assertion holds during eval,
 //     OK is printed, otherwise the reason the assertion(s) failed is printed.
@@ -68,6 +111,16 @@ import (
 //     threshold (e.g. threshold=0.05) % of the mean, the assertion fails. This
 //     assertion applies per-store, over 'ticks' duration.
 //
+//     For type=stat assertions, if the stat (e.g. stat=replicas) value of the
+//     last ticks (e.g. ticks=5) duration is not exactly equal to threshold,
+//     the assertion fails. This applies for a specified store which must be
+//     provided with store=storeID.
+//
+//     For type=conformance assertions, you may assert on the number of
+//     replicas that you expect to be underreplicated (under),
+//     overreplicated(over), unavailable(unavailable) and violating
+//     constraints(violating) at the end of the evaluation.
+//
 //   - "setting" [rebalance_mode=<int>] [rebalance_interval=<duration>]
 //     [rebalance_qps_threshold=<float>] [split_qps_threshold=<float>]
 //     [rebalance_range_threshold=<float>] [gossip_delay=<duration>]
@@ -89,14 +142,31 @@ import (
 //     is the simulated time and the y axis is the stat value. A series is
 //     rendered per-store, so if there are 10 stores, 10 series will be
 //     rendered.
+//
+//   - "topology" [sample=<int>]
+//     Print the cluster locality topology of the sample given (default=last).
+//     e.g. for the load_cluster config=single_region
+//     US
+//     ..US_1
+//     ....└── [1 2 3 4 5]
+//     ..US_2
+//     ....└── [6 7 8 9 10]
+//     ..US_3
+//     ....└── [11 12 13 14 15]
 func TestDataDriven(t *testing.T) {
 	ctx := context.Background()
 	dir := datapathutils.TestDataPath(t, ".")
 	datadriven.Walk(t, dir, func(t *testing.T, path string) {
 		const defaultKeyspace = 10000
 		loadGen := gen.BasicLoad{}
-		stateGen := gen.BasicState{}
+		var clusterGen gen.ClusterGen
+		var rangeGen gen.RangeGen = gen.BasicRanges{
+			Ranges:            1,
+			ReplicationFactor: 1,
+			KeySpace:          defaultKeyspace,
+		}
 		settingsGen := gen.StaticSettings{Settings: config.DefaultSimulationSettings()}
+		eventGen := gen.StaticEvents{DelayedEvents: event.DelayedEventList{}}
 		assertions := []SimulationAssertion{}
 		runs := []asim.History{}
 		datadriven.RunTest(t, path, func(t *testing.T, d *datadriven.TestData) string {
@@ -123,21 +193,177 @@ func TestDataDriven(t *testing.T) {
 				loadGen.MaxBlockSize = maxBlock
 				loadGen.MinBlockSize = minBlock
 				return ""
-			case "gen_state":
-				var stores, ranges, replFactor, keyspace = 3, 1, 3, defaultKeyspace
+			case "gen_ranges":
+				var ranges, replFactor, keyspace = 1, 3, defaultKeyspace
+				var bytes int64 = 0
 				var placementSkew bool
 
-				scanIfExists(t, d, "stores", &stores)
 				scanIfExists(t, d, "ranges", &ranges)
 				scanIfExists(t, d, "repl_factor", &replFactor)
 				scanIfExists(t, d, "placement_skew", &placementSkew)
 				scanIfExists(t, d, "keyspace", &keyspace)
+				scanIfExists(t, d, "bytes", &bytes)
+
+				var placementType gen.PlacementType
+				if placementSkew {
+					placementType = gen.Skewed
+				} else {
+					placementType = gen.Uniform
+				}
+				rangeGen = gen.BasicRanges{
+					Ranges:            ranges,
+					PlacementType:     placementType,
+					KeySpace:          keyspace,
+					ReplicationFactor: replFactor,
+					Bytes:             bytes,
+				}
+				return ""
+			case "topology":
+				var sample = len(runs)
+				scanIfExists(t, d, "sample", &sample)
+				top := runs[sample-1].S.Topology()
+				return (&top).String()
+			case "gen_cluster":
+				var nodes = 3
+				var storesPerNode = 1
+				scanIfExists(t, d, "nodes", &nodes)
+				scanIfExists(t, d, "stores_per_node", &storesPerNode)
+				clusterGen = gen.BasicCluster{
+					Nodes:         nodes,
+					StoresPerNode: storesPerNode,
+				}
+				return ""
+			case "load_cluster":
+				var config string
+				var clusterInfo state.ClusterInfo
+				scanArg(t, d, "config", &config)
+
+				switch config {
+				case "single_region":
+					clusterInfo = state.SingleRegionConfig
+				case "single_region_multi_store":
+					clusterInfo = state.SingleRegionMultiStoreConfig
+				case "multi_region":
+					clusterInfo = state.MultiRegionConfig
+				case "complex":
+					clusterInfo = state.ComplexConfig
+				default:
+					panic(fmt.Sprintf("unknown cluster config %s", config))
+				}
+
+				clusterGen = gen.LoadedCluster{
+					Info: clusterInfo,
+				}
+				return ""
+			case "add_node":
+				var delay time.Duration
+				var numStores = 1
+				var localityString string
+				scanIfExists(t, d, "delay", &delay)
+				scanIfExists(t, d, "stores", &numStores)
+				scanIfExists(t, d, "locality", &localityString)
+
+				addEvent := event.DelayedEvent{
+					EventFn: func(ctx context.Context, tick time.Time, s state.State) {
+						node := s.AddNode()
+						if localityString != "" {
+							var locality roachpb.Locality
+							if err := locality.Set(localityString); err != nil {
+								panic(fmt.Sprintf("unable to set node locality %s", err.Error()))
+							}
+							s.SetNodeLocality(node.NodeID(), locality)
+						}
+						for i := 0; i < numStores; i++ {
+							if _, ok := s.AddStore(node.NodeID()); !ok {
+								panic(fmt.Sprintf("adding store to node=%d failed", node))
+							}
+						}
+					},
+					At: settingsGen.Settings.StartTime.Add(delay),
+				}
+				eventGen.DelayedEvents = append(eventGen.DelayedEvents, addEvent)
+				return ""
+			case "set_span_config":
+				var delay time.Duration
+				scanIfExists(t, d, "delay", &delay)
+				for _, line := range strings.Split(d.Input, "\n") {
+					line = strings.TrimSpace(line)
+					if line == "" {
+						continue
+					}
+					tag, data, found := strings.Cut(line, ":")
+					require.True(t, found)
+					tag, data = strings.TrimSpace(tag), strings.TrimSpace(data)
+					span := spanconfigtestutils.ParseSpan(t, tag)
+					conf := spanconfigtestutils.ParseZoneConfig(t, data).AsSpanConfig()
+					eventGen.DelayedEvents = append(eventGen.DelayedEvents, event.DelayedEvent{
+						EventFn: func(ctx context.Context, tick time.Time, s state.State) {
+							s.SetSpanConfig(span, conf)
+						},
+						At: settingsGen.Settings.StartTime.Add(delay),
+					})
+				}
+				return ""
+			case "set_liveness":
+				var nodeID int
+				var liveness string
+				var delay time.Duration
+				livenessStatus := 3
+				scanArg(t, d, "node", &nodeID)
+				scanArg(t, d, "liveness", &liveness)
+				scanIfExists(t, d, "delay", &delay)
+				switch liveness {
+				case "unknown":
+					livenessStatus = 0
+				case "dead":
+					livenessStatus = 1
+				case "unavailable":
+					livenessStatus = 2
+				case "live":
+					livenessStatus = 3
+				case "decommissioning":
+					livenessStatus = 4
+				case "draining":
+					livenessStatus = 5
+					panic(fmt.Sprintf("unkown liveness status: %s", liveness))
+				}
+				eventGen.DelayedEvents = append(eventGen.DelayedEvents, event.DelayedEvent{
+					EventFn: func(ctx context.Context, tick time.Time, s state.State) {
+						s.SetNodeLiveness(
+							state.NodeID(nodeID),
+							livenesspb.NodeLivenessStatus(livenessStatus),
+						)
+					},
+					At: settingsGen.Settings.StartTime.Add(delay),
+				})
+				return ""
+			case "set_capacity":
+				var store int
+				var ioThreshold float64 = -1
+				var capacity, available int64 = -1, -1
+				var delay time.Duration
+
+				scanArg(t, d, "store", &store)
+				scanIfExists(t, d, "io_threshold", &ioThreshold)
+				scanIfExists(t, d, "capacity", &capacity)
+				scanIfExists(t, d, "available", &available)
+				scanIfExists(t, d, "delay", &delay)
+
+				capacityOverride := state.NewCapacityOverride()
+				capacityOverride.Capacity = capacity
+				capacityOverride.Available = available
+				if ioThreshold != -1 {
+					capacityOverride.IOThreshold = allocatorimpl.TestingIOThresholdWithScore(ioThreshold)
+				}
+
+				eventGen.DelayedEvents = append(eventGen.DelayedEvents, event.DelayedEvent{
+					EventFn: func(ctx context.Context, tick time.Time, s state.State) {
+						log.Infof(ctx, "setting capacity override %+v", capacityOverride)
+						s.SetCapacityOverride(state.StoreID(store), capacityOverride)
+					},
+					At: settingsGen.Settings.StartTime.Add(delay),
+				})
 
-				stateGen.Stores = stores
-				stateGen.ReplicationFactor = replFactor
-				stateGen.KeySpace = keyspace
-				stateGen.Ranges = ranges
-				stateGen.SkewedPlacement = placementSkew
 				return ""
 			case "eval":
 				samples := 1
@@ -158,7 +384,8 @@ func TestDataDriven(t *testing.T) {
 				for sample := 0; sample < samples; sample++ {
 					assertionFailures := []string{}
 					simulator := gen.GenerateSimulation(
-						duration, stateGen, loadGen, settingsGen, seedGen.Int63(),
+						duration, clusterGen, rangeGen, loadGen,
+						settingsGen, eventGen, seedGen.Int63(),
 					)
 					simulator.RunSim(ctx)
 					history := simulator.History()
@@ -195,23 +422,55 @@ func TestDataDriven(t *testing.T) {
 				var threshold float64
 
 				scanArg(t, d, "type", &typ)
-				scanArg(t, d, "stat", &stat)
-				scanArg(t, d, "ticks", &ticks)
-				scanArg(t, d, "threshold", &threshold)
 
 				switch typ {
 				case "balance":
+					scanArg(t, d, "stat", &stat)
+					scanArg(t, d, "ticks", &ticks)
+					scanArg(t, d, "threshold", &threshold)
 					assertions = append(assertions, balanceAssertion{
 						ticks:     ticks,
 						stat:      stat,
 						threshold: threshold,
 					})
 				case "steady":
+					scanArg(t, d, "stat", &stat)
+					scanArg(t, d, "ticks", &ticks)
+					scanArg(t, d, "threshold", &threshold)
 					assertions = append(assertions, steadyStateAssertion{
 						ticks:     ticks,
 						stat:      stat,
 						threshold: threshold,
 					})
+				case "stat":
+					var store int
+					scanArg(t, d, "stat", &stat)
+					scanArg(t, d, "ticks", &ticks)
+					scanArg(t, d, "threshold", &threshold)
+					scanArg(t, d, "store", &store)
+					assertions = append(assertions, storeStatAssertion{
+						ticks:         ticks,
+						stat:          stat,
+						acceptedValue: threshold,
+						// TODO(kvoli): support setting multiple stores.
+						stores: []int{store},
+					})
+				case "conformance":
+					var under, over, unavailable, violating int
+					under = conformanceAssertionSentinel
+					over = conformanceAssertionSentinel
+					unavailable = conformanceAssertionSentinel
+					violating = conformanceAssertionSentinel
+					scanIfExists(t, d, "under", &under)
+					scanIfExists(t, d, "over", &over)
+					scanIfExists(t, d, "unavailable", &unavailable)
+					scanIfExists(t, d, "violating", &violating)
+					assertions = append(assertions, conformanceAssertion{
+						underreplicated: under,
+						overreplicated:  over,
+						violating:       violating,
+						unavailable:     unavailable,
+					})
 				}
 				return ""
 			case "setting":
@@ -221,6 +480,7 @@ func TestDataDriven(t *testing.T) {
 				scanIfExists(t, d, "split_qps_threshold", &settingsGen.Settings.SplitQPSThreshold)
 				scanIfExists(t, d, "rebalance_range_threshold", &settingsGen.Settings.RangeRebalanceThreshold)
 				scanIfExists(t, d, "gossip_delay", &settingsGen.Settings.StateExchangeDelay)
+				scanIfExists(t, d, "range_size_split_threshold", &settingsGen.Settings.RangeSizeSplitThreshold)
 				return ""
 			case "plot":
 				var stat string
@@ -228,7 +488,7 @@ func TestDataDriven(t *testing.T) {
 				var buf strings.Builder
 
 				scanArg(t, d, "stat", &stat)
-				scanArg(t, d, "sample", &sample)
+				scanIfExists(t, d, "sample", &sample)
 				scanIfExists(t, d, "height", &height)
 				scanIfExists(t, d, "width", &width)
 
diff --git a/pkg/kv/kvserver/asim/tests/testdata/example_add_node b/pkg/kv/kvserver/asim/tests/testdata/example_add_node
new file mode 100644
index 000000000000..83f828b79b41
--- /dev/null
+++ b/pkg/kv/kvserver/asim/tests/testdata/example_add_node
@@ -0,0 +1,60 @@
+# This test simulates the behavior of the roachtest replicate/1to3. Where
+# initially there is one store, two new stores are added and the the test
+# asserts the replica counts between the 3 stores eventually balances.
+gen_cluster nodes=1
+----
+
+# Generate 300 ranges, where each range is 100mb (logical).
+gen_ranges ranges=300 range_bytes=100000000 repl_factor=1
+----
+
+# Add the two new nodes that won't be in the initial cluster, however added as
+# soon as the simulation  evaluation begins i.e. with delay=0.
+add_node
+----
+
+add_node
+----
+
+# Assert that the replica counts balance within 5% of each other among stores.
+assertion type=balance stat=replicas ticks=6 threshold=1.05
+----
+
+# Update the replication factor for the keyspace to be 3, instead of the
+# initial replication factor of 1 set during generation.
+set_span_config
+[0,10000): num_replicas=3 num_voters=3
+----
+
+eval duration=20m samples=1 seed=42
+----
+OK
+
+# Plot the replica count from the evaluation. Since there are 300 replicas on
+# s1 and the default RF=3, we expect the other stores to be up-replicated to
+# 300 replicas as well.
+plot stat=replicas sample=1
+----
+----
+
+ 301 ┼──────────────────────────────────────╭────────────────────────────────────────
+ 281 ┤                                   ╭╭─╯
+ 261 ┤                                ╭╭──╯
+ 241 ┤                              ╭╭─╯
+ 221 ┤                           ╭───╯
+ 201 ┤                        ╭╭─╯
+ 181 ┤                      ╭──╯
+ 161 ┤                   ╭──╯
+ 140 ┤                ╭──╯╯
+ 120 ┤              ╭─╯╯
+ 100 ┤           ╭──╯
+  80 ┤         ╭─╯╯
+  60 ┤      ╭──╯
+  40 ┤    ╭─╯
+  20 ┤ ╭──╯
+   0 ┼─╯
+                                          replicas
+----
+----
+
+# vim:ft=sh
diff --git a/pkg/kv/kvserver/asim/tests/testdata/example_fulldisk b/pkg/kv/kvserver/asim/tests/testdata/example_fulldisk
new file mode 100644
index 000000000000..6654796c7829
--- /dev/null
+++ b/pkg/kv/kvserver/asim/tests/testdata/example_fulldisk
@@ -0,0 +1,68 @@
+gen_cluster nodes=5
+----
+
+gen_ranges ranges=500 bytes=300000000
+----
+
+gen_load rate=500 max_block=128000 min_block=128000
+----
+
+set_capacity store=5 capacity=45000000000
+----
+
+eval duration=30m seed=42
+----
+OK
+
+# Plot the replicas over time per store. With a steady state of writes, we will
+# repeatedly hit the disk fullness threshold which causes shedding replicas on
+# store 5. This is shown below as it sheds replicas.
+plot stat=replicas
+----
+----
+
+ 336 ┤                                       ╭╮   ╭╭╮╭╮─╭╮╭╭╮              ╭╮╭──╮╮╭╭─
+ 325 ┤                            ╭╮╭────╮╭────────╯╰╯╰─╯╰─╯╰──────────────╮╭╯─╯╰──╯╯
+ 314 ┤                   ╭╭╭╮╭─╭──╯╰╯╯╰╯╰╰╯             ╰╯       ╰╯ ╰╯╰╯╰╯ ╰╯
+ 302 ┼───────────────╮─────────╯
+ 291 ┤               ╰───╮
+ 280 ┤                   ╰╮ ╭╮
+ 269 ┤                    ╰─╯╰╮
+ 258 ┤                        ╰╮
+ 246 ┤                         ╰──╮
+ 235 ┤                            ╰╮
+ 224 ┤                             ╰╮
+ 213 ┤                              ╰╮╭────╮                            ╭╮
+ 202 ┤                               ╰╯    ╰╮                ╭──────────╯╰───╮
+ 190 ┤                                      ╰─────╮   ╭───╮ ╭╯               │
+ 179 ┤                                            ╰───╯   ╰─╯                ╰─╮ ╭──╮
+ 168 ┤                                                                         ╰─╯  ╰
+                                          replicas
+----
+----
+
+# Plot the % of disk storage capacity used. We should see s5 hovering right
+# around 92.5-95% (the storage capacity threshold value).
+plot stat=disk_fraction_used
+----
+----
+
+ 0.98 ┤              ╭─╮ ╭╮  ╭╮╭─╮╭──╮  ╭──────╮╭─╮    ╭───╮    ╭╮  ╭╮╭─╮  ╭───╮   ╭─╮
+ 0.91 ┤      ╭───────╯ ╰─╯╰──╯╰╯ ╰╯  ╰──╯      ╰╯ ╰────╯   ╰────╯╰──╯╰╯ ╰──╯   ╰───╯ ╰
+ 0.85 ┼──────╯
+ 0.78 ┤
+ 0.72 ┤
+ 0.65 ┤
+ 0.59 ┤
+ 0.52 ┤
+ 0.46 ┤
+ 0.39 ┤
+ 0.33 ┤
+ 0.26 ┤
+ 0.20 ┤
+ 0.13 ┤
+ 0.07 ┤
+ 0.00 ┼───────────────────────────────────────────────────────────────────────────────
+                                      disk_fraction_used
+----
+----
diff --git a/pkg/kv/kvserver/asim/tests/testdata/example_io_overload b/pkg/kv/kvserver/asim/tests/testdata/example_io_overload
new file mode 100644
index 000000000000..0e1adcb16822
--- /dev/null
+++ b/pkg/kv/kvserver/asim/tests/testdata/example_io_overload
@@ -0,0 +1,41 @@
+gen_cluster nodes=5
+----
+
+gen_ranges ranges=500 placement_skew=true
+----
+
+set_capacity store=5 io_threshold=1
+----
+
+assertion type=stat stat=replicas store=5 threshold=0 ticks=5
+----
+
+eval duration=10m seed=42
+----
+OK
+
+# Expect s5 to get no replicas due to IO overload. The plot below should show a
+# solid line at 0, which will be s5's replica count.
+plot stat=replicas
+----
+----
+
+ 500 ┼────────╮╮
+ 467 ┤     ╰──╰───────────╮╮
+ 433 ┤                   ╰╰────────────╮╮
+ 400 ┤                                ╰╰───────────╮─╮    ╭╮   ╭╮  ╭╮╭──╮
+ 367 ┤                                             ╰╭─────╯╰───╯╰──╯╰╯──╰────────────
+ 333 ┤                                         ╭────╯
+ 300 ┤                                     ╭───╯
+ 267 ┤                                 ╭───╯
+ 233 ┤                            ╭────╯
+ 200 ┤                        ╭───╯
+ 167 ┤                   ╭────╯
+ 133 ┤               ╭───╯
+ 100 ┤           ╭───╯
+  67 ┤      ╭────╯
+  33 ┤  ╭───╯
+   0 ┼───────────────────────────────────────────────────────────────────────────────
+                                          replicas
+----
+----
diff --git a/pkg/kv/kvserver/asim/tests/testdata/example_liveness b/pkg/kv/kvserver/asim/tests/testdata/example_liveness
new file mode 100644
index 000000000000..9a4e5d970027
--- /dev/null
+++ b/pkg/kv/kvserver/asim/tests/testdata/example_liveness
@@ -0,0 +1,87 @@
+# This example sets n7 to dead initially and n5 to decommissioning after 2
+# minutes. The output of replicas per store is then plotted.
+#
+# Create 7 stores, with 700 ranges (RF=3). Each store should have approx 300
+# replicas and 100 leases.
+gen_cluster nodes=7
+----
+
+gen_ranges ranges=700
+----
+
+# n7 is dead and remains dead forever. It will still have its initial (3000)
+# replicas.
+set_liveness node=7 liveness=dead
+----
+
+# n6 becomes decommissioning after 3 minutes and remains decommissioning
+# thereafter.
+set_liveness node=6 liveness=decommissioning delay=3m
+----
+
+# The number of replicas on the dead store should be 0, assert this.
+assertion type=stat stat=replicas ticks=6 threshold=0 store=7
+----
+
+# The number of replicas on the decommissioning store should be 0, assert this.
+assertion type=stat stat=replicas ticks=6 threshold=0 store=6
+----
+
+eval duration=12m seed=42
+----
+OK
+
+# We expect one node(store) (n7) to immediately start losing replicas, whilst
+# other stores gain replicas evenly. After 3 minutes, we expect another
+# node(store) (n6) to begin losing replicas in a similar manner.
+plot stat=replicas
+----
+----
+
+ 432 ┤                                                      ╭────╭───────────────────
+ 403 ┤                                   ╭──────╭───╭────────────────────────────────
+ 374 ┤                         ╭─╭──╭───────────────╯╯
+ 346 ┤                ╭─╭╭──────────╯
+ 317 ┤╭╭╭─────────────────────╮
+ 288 ┼──╮                     ╰───╮
+ 259 ┤  ╰──╮                      ╰────╮
+ 230 ┤     ╰─╮                         ╰──╮
+ 202 ┤       ╰──╮                         ╰──╮
+ 173 ┤          ╰───╮                        ╰────╮
+ 144 ┤              ╰──╮                          ╰──╮
+ 115 ┤                 ╰─╮                           ╰──╮
+  86 ┤                   ╰───╮                          ╰──╮
+  58 ┤                       ╰──╮                          ╰────╮
+  29 ┤                          ╰───╮                           ╰───────╮
+   0 ┤                              ╰────────────────────────────────────────────────
+                                          replicas
+----
+----
+
+# Both nodes should begin losing leases immediately after their liveness status
+# is changed to dead or decommissioning (5 minutes later).
+plot stat=leases
+----
+----
+
+ 148 ┤                                                       ╭───────────────────────
+ 138 ┤                                             ╭───╭─────╭───────────────────────
+ 128 ┤                                      ╭────╭───────────╯╯──╯
+ 118 ┤              ╭╮╭─────────────────╮────────╯────╯
+ 109 ┤     ╭──────────╯───────────────╯ ╰─╮
+  99 ┼──╮──╯────────╯                     ╰─╮
+  89 ┤  ╰───╮                               ╰──╮
+  79 ┤      ╰─╮                                ╰─╮
+  69 ┤        ╰──╮                               ╰─╮
+  59 ┤           ╰───╮                             ╰╮
+  49 ┤               ╰─╮                            ╰──╮
+  39 ┤                 ╰───╮                           ╰─╮
+  30 ┤                     ╰──╮                          ╰─╮
+  20 ┤                        ╰───╮                        ╰──╮
+  10 ┤                            ╰──╮                        ╰──╮
+   0 ┤                               ╰───────────────────────────────────────────────
+                                           leases
+----
+----
+
+# vim:ft=sh
diff --git a/pkg/kv/kvserver/asim/tests/testdata/example_load_cluster b/pkg/kv/kvserver/asim/tests/testdata/example_load_cluster
new file mode 100644
index 000000000000..99a2fc6c38f7
--- /dev/null
+++ b/pkg/kv/kvserver/asim/tests/testdata/example_load_cluster
@@ -0,0 +1,49 @@
+# This test shows how configurations may be loaded from the existing catalog.
+# This test also demonstrates how to use conformance assertions to check
+# replication meet expectations.
+load_cluster config=complex
+----
+
+# Load just a single range into state, with a RF=5.
+gen_ranges ranges=1 repl_factor=5
+----
+
+# Set the span config so that there are only voters, with 3 voters in US_East
+# and 1 voter each in US_West and EU.
+set_span_config
+[0,10000): num_replicas=5 num_voters=5 constraints={'+region=US_East':3,'+region=US_West':1,'+region=EU':1} voter_constraints={'+region=US_East':3,'+region=US_West':1,'+region=EU':1}
+----
+
+# This assertion will fail if there are more than 0 unavailable, under
+# replicated, over replicated or constraint violating ranges, once the
+# simulation evaluation ends.
+assertion type=conformance unavailable=0 under=0 over=0  violating=0
+----
+
+eval duration=2m samples=1 seed=42
+----
+OK
+
+topology
+----
+EU
+  EU_1
+  │ └── [19 20 21]
+  EU_2
+  │ └── [22 23 24]
+  EU_3
+  │ └── [25 26 27 28]
+US_East
+  US_East_1
+  │ └── [1]
+  US_East_2
+  │ └── [2 3]
+  US_East_3
+  │ └── [4 5 6 7 8 9 10 11 12 13 14 15 16]
+US_West
+  US_West_1
+    └── [17 18]
+
+
+
+# vim:ft=sh
diff --git a/pkg/kv/kvserver/asim/tests/testdata/example_multi_store b/pkg/kv/kvserver/asim/tests/testdata/example_multi_store
new file mode 100644
index 000000000000..93a0f1cd9012
--- /dev/null
+++ b/pkg/kv/kvserver/asim/tests/testdata/example_multi_store
@@ -0,0 +1,46 @@
+# This test simulates identical parameters as the rebalance_load multi-store
+# test. The number of leases per store should be equal to 1. We assert on this
+# with a balance threshold of 1 (i.e. identical number of leases) and a steady
+# state threshold of 0 (i.e. doesn't change).
+gen_cluster nodes=7 stores_per_node=2
+----
+
+gen_ranges ranges=14 placement_skew=true
+----
+
+gen_load rate=7000
+----
+
+assertion stat=leases type=balance ticks=6 threshold=1
+----
+
+assertion stat=leases type=steady ticks=6 threshold=0
+----
+
+eval duration=5m seed=42
+----
+OK
+
+plot stat=leases
+----
+----
+
+ 14.00 ┼╮
+ 13.07 ┤╰╮
+ 12.13 ┤ ╰╮
+ 11.20 ┤  │
+ 10.27 ┤  │
+  9.33 ┤  │
+  8.40 ┤  ╰╮
+  7.47 ┤   ╰╮
+  6.53 ┤    │
+  5.60 ┤    │
+  4.67 ┤    │
+  3.73 ┤    ╰───────────╮
+  2.80 ┤                │
+  1.87 ┤    ╭───────────╮──────────────╮
+  0.93 ┤╭╭╭────────────────────────────────────────────────────────────────────────────
+  0.00 ┼──╯─────────────╯──────────────╯
+                                             leases
+----
+----
diff --git a/pkg/kv/kvserver/asim/tests/testdata/example_rebalancing b/pkg/kv/kvserver/asim/tests/testdata/example_rebalancing
index d26af0187a23..99b155512bed 100644
--- a/pkg/kv/kvserver/asim/tests/testdata/example_rebalancing
+++ b/pkg/kv/kvserver/asim/tests/testdata/example_rebalancing
@@ -2,7 +2,10 @@
 # where there are 7 stores, 7 ranges and initially the replicas are placed
 # following a skewed distribution (where s1 has the most replicas, s2 has half
 # as many as s1...).
-gen_state stores=7 ranges=7 placement_skew=true
+gen_cluster nodes=7
+----
+
+gen_ranges ranges=7 placement_skew=true
 ----
 
 # Create a load generator, where there are 7k ops/s and the access follows a
diff --git a/pkg/kv/kvserver/asim/tests/testdata/example_splitting b/pkg/kv/kvserver/asim/tests/testdata/example_splitting
index 4f644d434dd6..51e5d1b1ceb2 100644
--- a/pkg/kv/kvserver/asim/tests/testdata/example_splitting
+++ b/pkg/kv/kvserver/asim/tests/testdata/example_splitting
@@ -1,6 +1,9 @@
 # Explore how load based and sized based splitting occur in isolation. In this
 # example, there is only one store so no rebalancing activity should occur. 
-gen_state stores=1 ranges=1 repl_factor=1
+gen_cluster nodes=1
+----
+
+gen_ranges ranges=1 repl_factor=1
 ----
 
 # Create a load generator, where there is higher ops/s than the qps split