Merge #148753 #149538 #149699 #149717

148753: changefeedccl: gate Kafka v2 message too large error detail behind cluster setting r=asg0451 a=elizaMkraule

A recent change added detailed logging for Kafka v2 changefeed
messages that exceed the broker's size limit. These logs now
include the message key, size, and MVCC timestamp to aid in
debugging.

To make this safe for backporting, the behavior is now gated
behind the cluster setting:

changefeed.kafka_v2_include_error_details

In the main branch, this setting defaults to true to preserve
the enhanced observability.

In release branch backports, it will default to false.

When enabled, the log will include:
- The key of the offending message
- Combined key + value size
- MVCC timestamp

When disabled, the log reverts to the previous, minimal format.

Related to: 
Jira issue: [CRDB-49646](https://cockroachlabs.atlassian.net/browse/CRDB-49646)
See also #144994


Release note (general change): Kafka v2 changefeed sinks now
support a cluster setting that enables detailed error logging
for messages exceeding Kafka v2 size limit.

149538: sql: don't throw errors for skipped auto stats jobs r=mw5h a=mw5h

Previously, auto stats jobs would throw errors and increase failed jobs
counters if they attempted to start while a stats collection was already
in progress on the table. For large clusters with
'sql.stats.automatic_job_check_before_creating_job.enabled' set to true,
this could create quite a few failed jobs. These failed jobs don't seem
to cause any performance issues, but they clutter logs, potentially
obscuring real problems and alarming customers, who then file tickets
with support to figure out why their jobs are failing.

This patch:
*  refactors the autostats checks to reduce code duplication.
*  swallows the error for concurrent auto stats creation, logging at
   INFO level instead.
*  changes the create stats jobs test so that it no longer expects these
   jobs creations to fail and instead expects the stats to not be
   collected.
*  fixes a bug in the create stats jobs test that would cause it to hang
   instead of exiting on error.
*  adds a cluster setting,
   sql.stats.error_on_concurrent_create_stats.enabled, which controls
   this new behavior. By default the old behavior is maintained.

Fixes: #148413
Release note (ops change): CockroachDB now has a cluster setting,
sql.stats.error_on_concurrent_create_stats.enabled, which modifies how
it reacts to concurrent auto stats jobs. The default, true, maintains
the previous behavior. Setting this to false will cause the concurrent
auto stats job to be skipped with just a log entry and no increased
error counters.

149699: asim: update range usage info and store capacity  r=tbg a=wenyihu6

**asim: remove unused range usage info**

This commit removes the unused field RangeUsageInfo
from TransferLeaseOp.

Epic: none
Release note: none

---

**asim: support request_cpu_per_access and raft_cpu_per_write**

This commit adds support for the request_cpu_per_access and raft_cpu_per_write
options in the gen_load command. It only adds the options to the data-driven
framework and workload generator. But no real changes have been made to
LoadEvent, and they currently have no effect on range usage or apply load.
Future commits will implement the actual impact.

Epic: none
Release note: none

---

**asim: add impact from request_cpu_per_access and raft_cpu_per_write**

Previously, request_cpu_per_access and raft_cpu_per_write were added to the
workload generator in data driven tests, but they had no actual effect to the
cluster yet. This commit makes them take effect by applying the impact from
LoadEvent including CPUPerSecond in store capacity and recording range load
stats.

Epic: none
Release note: none

---

**asim: add store capacity cpu stats  to storemetrics**

Previously, store capacity cpu was populated. This commit adds the corresponding
stats to the StoreMetrics.

Epic: none
Release note: none

---

**asim: removes redundant size assignment**

This commit removes an redundant size assignment for rangeInfo in
LoadRangeInfo, since the caller already populates rangesInfo.

Epic: none
Release note: none

---

**asim: add a comment for RangeUsageInfo.WritesPerSecond**

This commit adds a comment clarifying nuances with WritesPerSecond in
RangeUsageInfo. It is actually the sum of writes rather than the rate. It's
currently unused outside of two unit tests TestWorkloadApply and
TestCapacityOverride. They both abuse this field by verifying that the writes
reach the replicas and the sum of writes is expected. Since it is tricky to
assert on the exact per rate stat, we currently leave it as is. But we should
fix this later.

Epic: none
Release note: none

---

**asim: account for follower replica load**

Previously, asim only accounted for load on the leaseholder, ignoring
non-leaseholder replicas. This commit updates it to consider all replicas for
RangeUsageInfo and store capacity aggregation. RangeUsageInfo handles
leaseholder checks and clears request CPU and QPS stats for non-leaseholder
replicas.

Epic: none
Release note: none

---

**asim: add range info String method**

This commit adds a RangeIndo string method.

Epic: none
Release note: none

---

**asim: add write bytes per sec to capacity & range usage**

Previously, we added WriteBytesPerSecond to roachpb.StoreCapacity. This commit
plumbs it through store capacity aggregation, range load usage, and
StoreMetrics. MMA will later use to track write bandwidth usage across stores.

Epic: none
Release note: none

---

**asim: support node_cpu_rate_capacity with gen_cluster**

Previously, we added roachpb.NodeCapacity for MMA to compute resource
utilization across stores. This commit integrates it into the asim setup,
enabling gen_cluster to use the node_cpu_rate_capacity option. Note that no
functions currently access node capacity; future MMA integration commits will
utilize it.

Epic: none
Release note: none

---

**asim: add comments for datadriven**

This commit updates comments for a few options we added recently.

Epic: none
Release note: none


149717: logictest: add back assertion that was rewritten accidentally r=rafiss a=rafiss

ee263e2 rewrote this test so that it expects no spanconfig. This was likely a mistake caused by rewriting before retrying for long enough.

This patch adds back the assertion, and adds another one that should prevent accidental rewrites.

fixes #148603
Release note: None

Co-authored-by: Eliza Kraule <[email protected]>
Co-authored-by: Matt White <[email protected]>
Co-authored-by: wenyihu6 <[email protected]>
Co-authored-by: Rafi Shamim <[email protected]>

Author: 5 people

docs/generated/settings/settings-for-tenants.txt

@@ -18,6 +18,7 @@ changefeed.event_consumer_worker_queue_size	integer	16	if changefeed.event_consu
 changefeed.event_consumer_workers	integer	0	the number of workers to use when processing events: <0 disables, 0 assigns a reasonable default, >0 assigns the setting value. for experimental/core changefeeds and changefeeds using parquet format, this is disabled	application
 changefeed.fast_gzip.enabled	boolean	true	use fast gzip implementation	application
 changefeed.span_checkpoint.lag_threshold (alias: changefeed.frontier_highwater_lag_checkpoint_threshold)	duration	10m0s	the amount of time a changefeed's lagging (slowest) spans must lag behind its leading (fastest) spans before a span-level checkpoint to save leading span progress is written; if 0, span-level checkpoints due to lagging spans is disabled	application
+changefeed.kafka_v2_error_details.enabled	boolean	true	if enabled, Kafka v2 sinks will include the message key, size, and MVCC timestamp in message too large errors	application
 changefeed.memory.per_changefeed_limit	byte size	512 MiB	controls amount of data that can be buffered per changefeed	application
 changefeed.resolved_timestamp.min_update_interval (alias: changefeed.min_highwater_advance)	duration	0s	minimum amount of time that must have elapsed since the last time a changefeed's resolved timestamp was updated before it is eligible to be updated again; default of 0 means no minimum interval is enforced but updating will still be limited by the average time it takes to checkpoint progress	application
 changefeed.node_throttle_config	string		specifies node level throttling configuration for all changefeeeds	application
@@ -355,6 +356,7 @@ sql.stats.automatic_partial_collection.fraction_stale_rows	float	0.05	target fra
 sql.stats.automatic_partial_collection.min_stale_rows	integer	100	target minimum number of stale rows per table that will trigger a partial statistics refresh	application
 sql.stats.cleanup.recurrence	string	@hourly	cron-tab recurrence for SQL Stats cleanup job	application
 sql.stats.detailed_latency_metrics.enabled	boolean	false	label latency metrics with the statement fingerprint. Workloads with tens of thousands of distinct query fingerprints should leave this setting false. (experimental, affects performance for workloads with high fingerprint cardinality)	application
+sql.stats.error_on_concurrent_create_stats.enabled	boolean	true	set to true to error on concurrent CREATE STATISTICS jobs, instead of skipping them	application
 sql.stats.flush.enabled	boolean	true	if set, SQL execution statistics are periodically flushed to disk	application
 sql.stats.flush.interval	duration	10m0s	the interval at which SQL execution statistics are flushed to disk, this value must be less than or equal to 1 hour	application
 sql.stats.forecasts.enabled	boolean	true	when true, enables generation of statistics forecasts by default for all tables	application

docs/generated/settings/settings.html

@@ -23,6 +23,7 @@
 <tr><td><div id="setting-changefeed-event-consumer-workers" class="anchored"><code>changefeed.event_consumer_workers</code></div></td><td>integer</td><td><code>0</code></td><td>the number of workers to use when processing events: &lt;0 disables, 0 assigns a reasonable default, &gt;0 assigns the setting value. for experimental/core changefeeds and changefeeds using parquet format, this is disabled</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-changefeed-fast-gzip-enabled" class="anchored"><code>changefeed.fast_gzip.enabled</code></div></td><td>boolean</td><td><code>true</code></td><td>use fast gzip implementation</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-changefeed-frontier-highwater-lag-checkpoint-threshold" class="anchored"><code>changefeed.span_checkpoint.lag_threshold<br />(alias: changefeed.frontier_highwater_lag_checkpoint_threshold)</code></div></td><td>duration</td><td><code>10m0s</code></td><td>the amount of time a changefeed&#39;s lagging (slowest) spans must lag behind its leading (fastest) spans before a span-level checkpoint to save leading span progress is written; if 0, span-level checkpoints due to lagging spans is disabled</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
+<tr><td><div id="setting-changefeed-kafka-v2-error-details-enabled" class="anchored"><code>changefeed.kafka_v2_error_details.enabled</code></div></td><td>boolean</td><td><code>true</code></td><td>if enabled, Kafka v2 sinks will include the message key, size, and MVCC timestamp in message too large errors</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-changefeed-memory-per-changefeed-limit" class="anchored"><code>changefeed.memory.per_changefeed_limit</code></div></td><td>byte size</td><td><code>512 MiB</code></td><td>controls amount of data that can be buffered per changefeed</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-changefeed-min-highwater-advance" class="anchored"><code>changefeed.resolved_timestamp.min_update_interval<br />(alias: changefeed.min_highwater_advance)</code></div></td><td>duration</td><td><code>0s</code></td><td>minimum amount of time that must have elapsed since the last time a changefeed&#39;s resolved timestamp was updated before it is eligible to be updated again; default of 0 means no minimum interval is enforced but updating will still be limited by the average time it takes to checkpoint progress</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-changefeed-node-throttle-config" class="anchored"><code>changefeed.node_throttle_config</code></div></td><td>string</td><td><code></code></td><td>specifies node level throttling configuration for all changefeeeds</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
@@ -310,6 +311,7 @@
 <tr><td><div id="setting-sql-stats-automatic-partial-collection-min-stale-rows" class="anchored"><code>sql.stats.automatic_partial_collection.min_stale_rows</code></div></td><td>integer</td><td><code>100</code></td><td>target minimum number of stale rows per table that will trigger a partial statistics refresh</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-sql-stats-cleanup-recurrence" class="anchored"><code>sql.stats.cleanup.recurrence</code></div></td><td>string</td><td><code>@hourly</code></td><td>cron-tab recurrence for SQL Stats cleanup job</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-sql-stats-detailed-latency-metrics-enabled" class="anchored"><code>sql.stats.detailed_latency_metrics.enabled</code></div></td><td>boolean</td><td><code>false</code></td><td>label latency metrics with the statement fingerprint. Workloads with tens of thousands of distinct query fingerprints should leave this setting false. (experimental, affects performance for workloads with high fingerprint cardinality)</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
+<tr><td><div id="setting-sql-stats-error-on-concurrent-create-stats-enabled" class="anchored"><code>sql.stats.error_on_concurrent_create_stats.enabled</code></div></td><td>boolean</td><td><code>true</code></td><td>set to true to error on concurrent CREATE STATISTICS jobs, instead of skipping them</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-sql-stats-flush-enabled" class="anchored"><code>sql.stats.flush.enabled</code></div></td><td>boolean</td><td><code>true</code></td><td>if set, SQL execution statistics are periodically flushed to disk</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-sql-stats-flush-interval" class="anchored"><code>sql.stats.flush.interval</code></div></td><td>duration</td><td><code>10m0s</code></td><td>the interval at which SQL execution statistics are flushed to disk, this value must be less than or equal to 1 hour</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-sql-stats-forecasts-enabled" class="anchored"><code>sql.stats.forecasts.enabled</code></div></td><td>boolean</td><td><code>true</code></td><td>when true, enables generation of statistics forecasts by default for all tables</td><td>Serverless/Dedicated/Self-Hosted</td></tr>

pkg/ccl/changefeedccl/changefeedbase/settings.go

@@ -369,3 +369,14 @@ var RetryBackoffReset = settings.RegisterDurationSettingWithExplicitUnit(
 	10*time.Minute, /* defaultValue */
 	settings.DurationInRange(1*time.Second, 1*time.Hour),
 )
+
+// KafkaV2IncludeErrorDetails enables detailed error messages for Kafka v2 sinks
+// when message_too_large errors occur. This includes the message key, size,
+// and MVCC timestamp in the error.
+var KafkaV2ErrorDetailsEnabled = settings.RegisterBoolSetting(
+	settings.ApplicationLevel,
+	"changefeed.kafka_v2_error_details.enabled",
+	"if enabled, Kafka v2 sinks will include the message key, size, and MVCC timestamp in message too large errors",
+	true,
+	settings.WithPublic,
+)

pkg/ccl/changefeedccl/sink_kafka_v2.go

@@ -41,9 +41,10 @@ type kafkaSinkClientV2 struct {
 	client      KafkaClientV2
 	adminClient KafkaAdminClientV2
 
-	knobs          kafkaSinkV2Knobs
-	canTryResizing bool
-	recordResize   func(numRecords int64)
+	knobs               kafkaSinkV2Knobs
+	canTryResizing      bool
+	includeErrorDetails bool
+	recordResize        func(numRecords int64)
 
 	topicsForConnectionCheck []string
 
@@ -124,6 +125,7 @@ func newKafkaSinkClientV2(
 		knobs:                    knobs,
 		batchCfg:                 batchCfg,
 		canTryResizing:           changefeedbase.BatchReductionRetryEnabled.Get(&settings.SV),
+		includeErrorDetails:      changefeedbase.KafkaV2ErrorDetailsEnabled.Get(&settings.SV),
 		recordResize:             recordResize,
 		topicsForConnectionCheck: topicsForConnectionCheck,
 	}
@@ -165,7 +167,7 @@ func (k *kafkaSinkClientV2) Flush(ctx context.Context, payload SinkPayload) (ret
 				}
 				return nil
 			} else {
-				if len(msgs) == 1 && errors.Is(err, kerr.MessageTooLarge) {
+				if len(msgs) == 1 && errors.Is(err, kerr.MessageTooLarge) && k.includeErrorDetails {
 					msg := msgs[0]
 					mvccVal := msg.Context.Value(mvccTSKey{})
 					var ts hlc.Timestamp
@@ -273,7 +275,7 @@ func (k *kafkaSinkClientV2) maybeUpdateTopicPartitions(
 
 // MakeBatchBuffer implements SinkClient.
 func (k *kafkaSinkClientV2) MakeBatchBuffer(topic string) BatchBuffer {
-	return &kafkaBuffer{topic: topic, batchCfg: k.batchCfg}
+	return &kafkaBuffer{topic: topic, batchCfg: k.batchCfg, includeErrorDetails: k.includeErrorDetails}
 }
 
 func (k *kafkaSinkClientV2) shouldTryResizing(err error, msgs []*kgo.Record) bool {
@@ -311,7 +313,8 @@ type kafkaBuffer struct {
 	messages  []*kgo.Record
 	byteCount int
 
-	batchCfg sinkBatchConfig
+	batchCfg            sinkBatchConfig
+	includeErrorDetails bool
 }
 
 type mvccTSKey struct{}
@@ -328,7 +331,10 @@ func (b *kafkaBuffer) Append(ctx context.Context, key []byte, value []byte, attr
 		headers = append(headers, kgo.RecordHeader{Key: k, Value: v})
 	}
 
-	rctx := context.WithValue(ctx, mvccTSKey{}, attrs.mvcc)
+	var rctx context.Context
+	if b.includeErrorDetails {
+		rctx = context.WithValue(ctx, mvccTSKey{}, attrs.mvcc)
+	}
 
 	b.messages = append(b.messages, &kgo.Record{Key: key, Value: value, Topic: b.topic, Headers: headers, Context: rctx})
 	b.byteCount += len(value)

pkg/kv/kvserver/asim/gen/generator.go

@@ -118,21 +118,24 @@ func (ml MultiLoad) Generate(seed int64, settings *config.SimulationSettings) []
 
 // BasicLoad implements the LoadGen interface.
 type BasicLoad struct {
-	RWRatio        float64
-	Rate           float64
-	SkewedAccess   bool
-	MinBlockSize   int
-	MaxBlockSize   int
-	MinKey, MaxKey int64
+	RWRatio             float64
+	Rate                float64
+	SkewedAccess        bool
+	MinBlockSize        int
+	MaxBlockSize        int
+	MinKey, MaxKey      int64
+	RequestCPUPerAccess int64
+	RaftCPUPerWrite     int64
 }
 
 var _ LoadGen = BasicLoad{}
 
 func (bl BasicLoad) String() string {
 	return fmt.Sprintf(
 		"basic load with rw_ratio=%0.2f, rate=%0.2f, skewed_access=%t, min_block_size=%d, max_block_size=%d, "+
-			"min_key=%d, max_key=%d",
-		bl.RWRatio, bl.Rate, bl.SkewedAccess, bl.MinBlockSize, bl.MaxBlockSize, bl.MinKey, bl.MaxKey)
+			"min_key=%d, max_key=%d, request_cpu_per_access=%d, raft_cpu_per_write=%d",
+		bl.RWRatio, bl.Rate, bl.SkewedAccess, bl.MinBlockSize, bl.MaxBlockSize,
+		bl.MinKey, bl.MaxKey, bl.RequestCPUPerAccess, bl.RaftCPUPerWrite)
 }
 
 // Generate returns a new list of workload generators where the generator
@@ -162,6 +165,8 @@ func (bl BasicLoad) Generate(seed int64, settings *config.SimulationSettings) []
 			bl.RWRatio,
 			bl.MaxBlockSize,
 			bl.MinBlockSize,
+			bl.RaftCPUPerWrite,
+			bl.RequestCPUPerAccess,
 		),
 	}
 }
@@ -188,17 +193,19 @@ func (lc LoadedCluster) Regions() []state.Region {
 
 // BasicCluster implements the ClusterGen interace.
 type BasicCluster struct {
-	Nodes             int
-	StoresPerNode     int
-	StoreByteCapacity int64
-	Region            []string
-	NodesPerRegion    []int
+	Nodes               int
+	StoresPerNode       int
+	StoreByteCapacity   int64
+	Region              []string
+	NodesPerRegion      []int
+	NodeCPURateCapacity int64
 }
 
 func (bc BasicCluster) String() string {
 	var b strings.Builder
-	_, _ = fmt.Fprintf(&b, "basic cluster with nodes=%d, stores_per_node=%d, store_byte_capacity=%d",
-		bc.Nodes, bc.StoresPerNode, bc.StoreByteCapacity)
+	_, _ = fmt.Fprintf(&b,
+		"basic cluster with nodes=%d, stores_per_node=%d, store_byte_capacity=%d, node_cpu_rate_capacity=%d",
+		bc.Nodes, bc.StoresPerNode, bc.StoreByteCapacity, bc.NodeCPURateCapacity)
 	if len(bc.Region) != 0 {
 		_, _ = fmt.Fprintf(&b, ", region=%v, nodes_per_region=%v", bc.Region, bc.NodesPerRegion)
 	}
@@ -212,6 +219,7 @@ func (bc BasicCluster) String() string {
 func (bc BasicCluster) Generate(seed int64, settings *config.SimulationSettings) state.State {
 	info := bc.info()
 	info.StoreDiskCapacityBytes = bc.StoreByteCapacity
+	info.NodeCPURateCapacityNanos = bc.NodeCPURateCapacity
 	return state.LoadClusterInfo(info, settings)
 }
 
@@ -346,9 +354,6 @@ func (b BaseRanges) GetRangesInfo(
 
 // LoadRangeInfo loads the given state with the specified rangesInfo.
 func (b BaseRanges) LoadRangeInfo(s state.State, rangesInfo state.RangesInfo) {
-	for _, rangeInfo := range rangesInfo {
-		rangeInfo.Size = b.Bytes
-	}
 	state.LoadRangeInfo(s, rangesInfo...)
 }
 

pkg/kv/kvserver/asim/metrics/series.go

@@ -21,8 +21,10 @@ func MakeTS(metrics [][]StoreMetrics) map[string][][]float64 {
 	// custom scraper or provide definitions for each metric available. These
 	// are partially duplicated with the cluster tracker.
 	ret["qps"] = make([][]float64, stores)
+	ret["cpu"] = make([][]float64, stores)
 	ret["write"] = make([][]float64, stores)
 	ret["write_b"] = make([][]float64, stores)
+	ret["write_bytes_per_second"] = make([][]float64, stores)
 	ret["read"] = make([][]float64, stores)
 	ret["read_b"] = make([][]float64, stores)
 	ret["replicas"] = make([][]float64, stores)
@@ -37,8 +39,10 @@ func MakeTS(metrics [][]StoreMetrics) map[string][][]float64 {
 	for _, sms := range metrics {
 		for i, sm := range sms {
 			ret["qps"][i] = append(ret["qps"][i], float64(sm.QPS))
+			ret["cpu"][i] = append(ret["cpu"][i], float64(sm.CPU))
 			ret["write"][i] = append(ret["write"][i], float64(sm.WriteKeys))
 			ret["write_b"][i] = append(ret["write_b"][i], float64(sm.WriteBytes))
+			ret["write_bytes_per_second"][i] = append(ret["write_bytes_per_second"][i], float64(sm.WriteBytesPerSecond))
 			ret["read"][i] = append(ret["read"][i], float64(sm.ReadKeys))
 			ret["read_b"][i] = append(ret["read_b"][i], float64(sm.ReadBytes))
 			ret["replicas"][i] = append(ret["replicas"][i], float64(sm.Replicas))

pkg/kv/kvserver/asim/metrics/tracker.go

@@ -18,15 +18,17 @@ import (
 // StoreMetrics tracks metrics per-store in a simulation run. Each metrics
 // struct is associated with a tick.
 type StoreMetrics struct {
-	Tick       time.Time
-	StoreID    int64
-	QPS        int64
-	WriteKeys  int64
-	WriteBytes int64
-	ReadKeys   int64
-	ReadBytes  int64
-	Replicas   int64
-	Leases     int64
+	Tick                time.Time
+	StoreID             int64
+	QPS                 int64
+	CPU                 int64
+	WriteKeys           int64
+	WriteBytes          int64
+	WriteBytesPerSecond int64
+	ReadKeys            int64
+	ReadBytes           int64
+	Replicas            int64
+	Leases              int64
 	// LeaseTransfers tracks the number of lease transfer that this store has
 	// authored. Only the leaseholder store authors transfers.
 	LeaseTransfers int64
@@ -45,14 +47,14 @@ func (sm *StoreMetrics) GetMetricValue(stat string) float64 {
 	switch stat {
 	case "qps":
 		return float64(sm.QPS)
-	// case "cpu":
-	// 	value = float64(sm.CPU)
-	// case "write_bytes_per_second":
-	// 	value = float64(sm.WriteBytesPerSecond)
+	case "cpu":
+		return float64(sm.CPU)
 	case "write":
 		return float64(sm.WriteKeys)
 	case "write_b":
 		return float64(sm.WriteBytes)
+	case "write_bytes_per_second":
+		return float64(sm.WriteBytesPerSecond)
 	case "read":
 		return float64(sm.ReadKeys)
 	case "read_b":
@@ -142,6 +144,7 @@ func (mt *Tracker) Tick(ctx context.Context, tick time.Time, s state.State) {
 			Tick:               tick,
 			StoreID:            int64(storeID),
 			QPS:                int64(desc.Capacity.QueriesPerSecond),
+			CPU:                int64(desc.Capacity.CPUPerSecond),
 			WriteKeys:          u.WriteKeys,
 			WriteBytes:         u.WriteBytes,
 			ReadKeys:           u.ReadKeys,

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

@@ -15,7 +15,6 @@ go_library(
     deps = [
         "//pkg/kv/kvpb",
         "//pkg/kv/kvserver",
-        "//pkg/kv/kvserver/allocator",
         "//pkg/kv/kvserver/allocator/allocatorimpl",
         "//pkg/kv/kvserver/allocator/storepool",
         "//pkg/kv/kvserver/asim/config",
@@ -30,7 +29,6 @@ go_test(
     srcs = ["controller_test.go"],
     embed = [":op"],
     deps = [
-        "//pkg/kv/kvserver/allocator",
         "//pkg/kv/kvserver/allocator/allocatorimpl",
         "//pkg/kv/kvserver/asim/config",
         "//pkg/kv/kvserver/asim/gossip",

pkg/kv/kvserver/asim/op/controller_test.go

@@ -11,7 +11,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/allocator"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/allocator/allocatorimpl"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/asim/config"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/asim/gossip"
@@ -106,7 +105,6 @@ func TestLeaseTransferOp(t *testing.T) {
 						roachpb.RangeID(rangeID),
 						0,
 						roachpb.StoreID(target),
-						allocator.RangeUsageInfo{},
 					)
 					ticket := controller.Dispatch(ctx, state.OffsetTick(start, tick), s, op)
 					pending = append(pending, ticket)

pkg/kv/kvserver/asim/op/transfer_lease.go

@@ -8,7 +8,6 @@ package op
 import (
 	"time"
 
-	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/allocator"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/asim/state"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/errors"
@@ -19,22 +18,17 @@ type TransferLeaseOp struct {
 	baseOp
 	source, target state.StoreID
 	rangeID        state.RangeID
-	usage          allocator.RangeUsageInfo
 }
 
 // NewTransferLeaseOp returns a new TransferLeaseOp.
 func NewTransferLeaseOp(
-	tick time.Time,
-	rangeID roachpb.RangeID,
-	source, target roachpb.StoreID,
-	usage allocator.RangeUsageInfo,
+	tick time.Time, rangeID roachpb.RangeID, source, target roachpb.StoreID,
 ) *TransferLeaseOp {
 	return &TransferLeaseOp{
 		baseOp:  newBaseOp(tick),
 		source:  state.StoreID(source),
 		target:  state.StoreID(target),
 		rangeID: state.RangeID(rangeID),
-		usage:   usage,
 	}
 }