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

Author: wenyihu6

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

@@ -24,6 +24,7 @@ func MakeTS(metrics [][]StoreMetrics) map[string][][]float64 {
 	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)
@@ -41,6 +42,7 @@ func MakeTS(metrics [][]StoreMetrics) map[string][][]float64 {
 			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,16 +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
-	CPU        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
@@ -48,12 +49,12 @@ func (sm *StoreMetrics) GetMetricValue(stat string) float64 {
 		return float64(sm.QPS)
 	case "cpu":
 		return float64(sm.CPU)
-	// case "write_bytes_per_second":
-	// 	value = float64(sm.WriteBytesPerSecond)
 	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":

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

@@ -270,6 +270,7 @@ func (s *state) capacity(storeID StoreID) roachpb.StoreCapacity {
 	capacity := store.desc.Capacity
 	capacity.QueriesPerSecond = 0
 	capacity.WritesPerSecond = 0
+	capacity.WriteBytesPerSecond = 0
 	capacity.LogicalBytes = 0
 	capacity.CPUPerSecond = 0
 	capacity.LeaseCount = 0
@@ -285,6 +286,7 @@ func (s *state) capacity(storeID StoreID) roachpb.StoreCapacity {
 		// special handling needed here.
 		capacity.QueriesPerSecond += usage.QueriesPerSecond
 		capacity.WritesPerSecond += usage.WritesPerSecond
+		capacity.WriteBytesPerSecond += usage.WriteBytesPerSecond
 		capacity.LogicalBytes += usage.LogicalBytes
 		capacity.CPUPerSecond += usage.RequestCPUNanosPerSecond + usage.RaftCPUNanosPerSecond
 		capacity.RangeCount++

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

@@ -75,6 +75,7 @@ func (rl *ReplicaLoadCounter) ApplyLoad(le workload.LoadEvent) {
 	// are used.
 	rl.loadStats.RecordReqCPUNanos(float64(le.RequestCPU))
 	rl.loadStats.RecordRaftCPUNanos(float64(le.RaftCPU))
+	rl.loadStats.RecordWriteBytes(float64(le.WriteSize))
 }
 
 // Load translates the recorded key accesses and size into range usage
@@ -83,14 +84,15 @@ func (rl *ReplicaLoadCounter) Load() allocator.RangeUsageInfo {
 	stats := rl.loadStats.Stats()
 
 	return allocator.RangeUsageInfo{
-		QueriesPerSecond:         stats.QueriesPerSecond,
+		QueriesPerSecond: stats.QueriesPerSecond,
 		// NB: WritesPerSecond is the sum of writes, rather than the rate. It is
 		// only used for testing and could be removed.
 		// TODO(wenyihu6): TestWorkloadApply and TestCapacityOverride tests this
 		// field as a way to check if replicas have received the load. It is a bit
 		// tricky to assert on the rate of writes per second. We should try
 		// refactoring.
 		WritesPerSecond:          float64(rl.WriteKeys),
+		WriteBytesPerSecond:      stats.WriteBytesPerSecond,
 		RaftCPUNanosPerSecond:    stats.RaftCPUNanosPerSecond,
 		RequestCPUNanosPerSecond: stats.RequestCPUNanosPerSecond,
 	}
@@ -133,15 +135,16 @@ type CapacityOverride roachpb.StoreCapacity
 // NewCapacityOverride returns a capacity override where no overrides are set.
 func NewCapacityOverride() CapacityOverride {
 	return CapacityOverride{
-		Capacity:         capacityOverrideSentinel,
-		Available:        capacityOverrideSentinel,
-		Used:             capacityOverrideSentinel,
-		LogicalBytes:     capacityOverrideSentinel,
-		RangeCount:       capacityOverrideSentinel,
-		LeaseCount:       capacityOverrideSentinel,
-		QueriesPerSecond: capacityOverrideSentinel,
-		WritesPerSecond:  capacityOverrideSentinel,
-		CPUPerSecond:     capacityOverrideSentinel,
+		Capacity:            capacityOverrideSentinel,
+		Available:           capacityOverrideSentinel,
+		Used:                capacityOverrideSentinel,
+		LogicalBytes:        capacityOverrideSentinel,
+		RangeCount:          capacityOverrideSentinel,
+		LeaseCount:          capacityOverrideSentinel,
+		QueriesPerSecond:    capacityOverrideSentinel,
+		WritesPerSecond:     capacityOverrideSentinel,
+		WriteBytesPerSecond: capacityOverrideSentinel,
+		CPUPerSecond:        capacityOverrideSentinel,
 		IOThresholdMax: admissionpb.IOThreshold{
 			L0NumSubLevels:           capacityOverrideSentinel,
 			L0NumSubLevelsThreshold:  capacityOverrideSentinel,
@@ -156,7 +159,7 @@ func NewCapacityOverride() CapacityOverride {
 func (co CapacityOverride) String() string {
 	return fmt.Sprintf(
 		"capacity=%d, available=%d, used=%d, logical_bytes=%d, range_count=%d, lease_count=%d, "+
-			"queries_per_sec=%.2f, writes_per_sec=%.2f, cpu_per_sec=%.2f, io_threshold_max=%v",
+			"queries_per_sec=%.2f, writes_per_sec=%.2f, write_bytes_per_sec=%.2f, cpu_per_sec=%.2f, io_threshold_max=%v",
 		co.Capacity,
 		co.Available,
 		co.Used,
@@ -165,6 +168,7 @@ func (co CapacityOverride) String() string {
 		co.LeaseCount,
 		co.QueriesPerSecond,
 		co.WritesPerSecond,
+		co.WriteBytesPerSecond,
 		co.CPUPerSecond,
 		co.IOThresholdMax,
 	)
@@ -198,6 +202,9 @@ func mergeOverride(
 	if override.WritesPerSecond != capacityOverrideSentinel {
 		ret.WritesPerSecond = override.WritesPerSecond
 	}
+	if override.WriteBytesPerSecond != capacityOverrideSentinel {
+		ret.WriteBytesPerSecond = override.WriteBytesPerSecond
+	}
 	if override.CPUPerSecond != capacityOverrideSentinel {
 		ret.CPUPerSecond = override.CPUPerSecond
 	}