Merge #77040

77040: kvserver: gossip store read amplification r=kvoli a=kvoli

Previously, other store statistics were gossipped between stores to
inform allocation and rebalancing decisions. This patch introduces an
additional gosipped value, the store read amplification. This patch
allows the allocator to avoid rebalancing ranges onto stores with poor
lsm health, indicated by a large read amplification.

resolves #74254

Release note: None

Co-authored-by: Austen McClernon <[email protected]>

pkg/kv/kvserver/store.go

@@ -2964,6 +2964,7 @@ func (s *Store) Capacity(ctx context.Context, useCached bool) (roachpb.StoreCapa
 	capacity.LogicalBytes = logicalBytes
 	capacity.QueriesPerSecond = totalQueriesPerSecond
 	capacity.WritesPerSecond = totalWritesPerSecond
+	capacity.ReadAmplification = s.metrics.RdbReadAmplification.Value()
 	capacity.BytesPerReplica = roachpb.PercentilesFromData(bytesPerReplica)
 	capacity.WritesPerReplica = roachpb.PercentilesFromData(writesPerReplica)
 	s.recordNewPerSecondStats(totalQueriesPerSecond, totalWritesPerSecond)

pkg/kv/kvserver/store_pool.go

@@ -755,6 +755,10 @@ type StoreList struct {
 	// candidateWritesPerSecond tracks writes-per-second stats for stores that are
 	// eligible to be rebalance targets.
 	candidateWritesPerSecond stat
+
+	// candidateReadAmplification tracks the read amplification stats for stores that are
+	// eligible to be rebalance targets.
+	candidateReadAmplification stat
 }
 
 // Generates a new store list based on the passed in descriptors. It will
@@ -769,28 +773,33 @@ func makeStoreList(descriptors []roachpb.StoreDescriptor) StoreList {
 		sl.candidateLogicalBytes.update(float64(desc.Capacity.LogicalBytes))
 		sl.candidateQueriesPerSecond.update(desc.Capacity.QueriesPerSecond)
 		sl.candidateWritesPerSecond.update(desc.Capacity.WritesPerSecond)
+		sl.candidateReadAmplification.update(float64(desc.Capacity.ReadAmplification))
 	}
 	return sl
 }
 
 func (sl StoreList) String() string {
 	var buf bytes.Buffer
 	fmt.Fprintf(&buf,
-		"  candidate: avg-ranges=%v avg-leases=%v avg-disk-usage=%v avg-queries-per-second=%v",
+		"  candidate: avg-ranges=%v avg-leases=%v avg-disk-usage=%v avg-queries-per-second=%v read-amplification=%v",
 		sl.candidateRanges.mean,
 		sl.candidateLeases.mean,
 		humanizeutil.IBytes(int64(sl.candidateLogicalBytes.mean)),
-		sl.candidateQueriesPerSecond.mean)
+		sl.candidateQueriesPerSecond.mean,
+		sl.candidateReadAmplification.mean,
+	)
 	if len(sl.stores) > 0 {
 		fmt.Fprintf(&buf, "\n")
 	} else {
 		fmt.Fprintf(&buf, " <no candidates>")
 	}
 	for _, desc := range sl.stores {
-		fmt.Fprintf(&buf, "  %d: ranges=%d leases=%d disk-usage=%s queries-per-second=%.2f\n",
+		fmt.Fprintf(&buf, "  %d: ranges=%d leases=%d disk-usage=%s queries-per-second=%.2f read-amplification=%d\n",
 			desc.StoreID, desc.Capacity.RangeCount,
 			desc.Capacity.LeaseCount, humanizeutil.IBytes(desc.Capacity.LogicalBytes),
-			desc.Capacity.QueriesPerSecond)
+			desc.Capacity.QueriesPerSecond,
+			desc.Capacity.ReadAmplification,
+		)
 	}
 	return buf.String()
 }

pkg/kv/kvserver/store_pool_test.go

@@ -513,26 +513,28 @@ func TestStorePoolUpdateLocalStore(t *testing.T) {
 			StoreID: 1,
 			Node:    roachpb.NodeDescriptor{NodeID: 1},
 			Capacity: roachpb.StoreCapacity{
-				Capacity:         100,
-				Available:        50,
-				RangeCount:       5,
-				LeaseCount:       1,
-				LogicalBytes:     30,
-				QueriesPerSecond: 100,
-				WritesPerSecond:  30,
+				Capacity:          100,
+				Available:         50,
+				RangeCount:        5,
+				LeaseCount:        1,
+				LogicalBytes:      30,
+				QueriesPerSecond:  100,
+				WritesPerSecond:   30,
+				ReadAmplification: 5,
 			},
 		},
 		{
 			StoreID: 2,
 			Node:    roachpb.NodeDescriptor{NodeID: 2},
 			Capacity: roachpb.StoreCapacity{
-				Capacity:         100,
-				Available:        55,
-				RangeCount:       4,
-				LeaseCount:       2,
-				LogicalBytes:     25,
-				QueriesPerSecond: 50,
-				WritesPerSecond:  25,
+				Capacity:          100,
+				Available:         55,
+				RangeCount:        4,
+				LeaseCount:        2,
+				LogicalBytes:      25,
+				QueriesPerSecond:  50,
+				WritesPerSecond:   25,
+				ReadAmplification: 10,
 			},
 		},
 	}
@@ -574,6 +576,9 @@ func TestStorePoolUpdateLocalStore(t *testing.T) {
 	if expectedWPS := 30 + WPS; desc.Capacity.WritesPerSecond != expectedWPS {
 		t.Errorf("expected WritesPerSecond %f, but got %f", expectedWPS, desc.Capacity.WritesPerSecond)
 	}
+	if expectedReadAmp := int64(5); desc.Capacity.ReadAmplification != expectedReadAmp {
+		t.Errorf("expected ReadAmplification %d, but got %d", expectedReadAmp, desc.Capacity.ReadAmplification)
+	}
 
 	sp.updateLocalStoreAfterRebalance(roachpb.StoreID(2), rangeUsageInfo, roachpb.REMOVE_VOTER)
 	desc, ok = sp.getStoreDescriptor(roachpb.StoreID(2))
@@ -592,6 +597,9 @@ func TestStorePoolUpdateLocalStore(t *testing.T) {
 	if expectedWPS := 25 - WPS; desc.Capacity.WritesPerSecond != expectedWPS {
 		t.Errorf("expected WritesPerSecond %f, but got %f", expectedWPS, desc.Capacity.WritesPerSecond)
 	}
+	if expectedReadAmp := int64(10); desc.Capacity.ReadAmplification != expectedReadAmp {
+		t.Errorf("expected ReadAmplification %d, but got %d", expectedReadAmp, desc.Capacity.ReadAmplification)
+	}
 
 	sp.updateLocalStoresAfterLeaseTransfer(roachpb.StoreID(1), roachpb.StoreID(2), rangeUsageInfo.QueriesPerSecond)
 	desc, ok = sp.getStoreDescriptor(roachpb.StoreID(1))

pkg/roachpb/metadata.go

@@ -560,12 +560,12 @@ func (sc StoreCapacity) String() string {
 // SafeFormat implements the redact.SafeFormatter interface.
 func (sc StoreCapacity) SafeFormat(w redact.SafePrinter, _ rune) {
 	w.Printf("disk (capacity=%s, available=%s, used=%s, logicalBytes=%s), "+
-		"ranges=%d, leases=%d, queries=%.2f, writes=%.2f, "+
+		"ranges=%d, leases=%d, queries=%.2f, writes=%.2f, readAmplification=%d"+
 		"bytesPerReplica={%s}, writesPerReplica={%s}",
 		humanizeutil.IBytes(sc.Capacity), humanizeutil.IBytes(sc.Available),
 		humanizeutil.IBytes(sc.Used), humanizeutil.IBytes(sc.LogicalBytes),
 		sc.RangeCount, sc.LeaseCount, sc.QueriesPerSecond, sc.WritesPerSecond,
-		sc.BytesPerReplica, sc.WritesPerReplica)
+		sc.ReadAmplification, sc.BytesPerReplica, sc.WritesPerReplica)
 }
 
 // FractionUsed computes the fraction of storage capacity that is in use.

pkg/roachpb/metadata.proto

@@ -330,6 +330,11 @@ message StoreCapacity {
   // by ranges in the store. The stat is tracked over the time period defined
   // in storage/replica_stats.go, which as of July 2018 is 30 minutes.
   optional double writes_per_second = 5 [(gogoproto.nullable) = false];
+  // read_amplification tracks the current read amplification in the store.
+  // TODO(kvoli): Use of this field will need to be version-gated, to avoid
+  // instances where overlapping node-binary versions within a cluster result
+  // in this this field missing.
+  optional int64 read_amplification = 11 [(gogoproto.nullable) = false];
   // bytes_per_replica and writes_per_replica contain percentiles for the
   // number of bytes and writes-per-second to each replica in the store.
   // This information can be used for rebalancing decisions.