kvserver: decline rebalance snapshots on receivers with poor LSM

This commit introduces two new cluster settings:
```
kv.snapshot_decline.read_amp_threshold

server.declined_snapshot_timeout
```

With this commit, stores with a read amplification level higher than
`kv.snapshot_decline.read_amp_threshold` will decline all `REBALANCE`
snapshots. Upon receiving a `DECLINED` response, the senders of these snapshots
will consider these receivers `throttled` for
`server.declined_snapshot_timeout`.

This means that stores with poor LSM health will not be considered as valid
candidates for replica rebalancing.

Fixes cockroachdb#73714
Related to cockroachdb#62168

Release note: None

pkg/kv/kvserver/raft.proto

@@ -207,6 +207,7 @@ message SnapshotResponse {
     ACCEPTED = 1;
     APPLIED = 2;
     ERROR = 3;
+    DECLINED = 5;
     reserved 4;
   }
   Status status = 1;

pkg/kv/kvserver/store_pool.go

@@ -51,6 +51,16 @@ var FailedReservationsTimeout = settings.RegisterDurationSetting(
 	settings.NonNegativeDuration,
 )
 
+// DeclinedSnapshotTimeout specifies a duration during which the local replicate
+// queue will not consider stores which have declined a snapshot a viable
+// target.
+var DeclinedSnapshotTimeout = settings.RegisterDurationSetting(
+	"server.declined_snapshot_timeout",
+	"the amount of time to consider the store throttled for up-replication after a declined rebalance snapshot",
+	60*time.Second,
+	settings.NonNegativeDuration,
+)
+
 const timeAfterStoreSuspectSettingName = "server.time_after_store_suspect"
 
 // TimeAfterStoreSuspect measures how long we consider a store suspect since
@@ -914,6 +924,7 @@ type throttleReason int
 const (
 	_ throttleReason = iota
 	throttleFailed
+	throttleDeclined
 )
 
 // throttle informs the store pool that the given remote store declined a
@@ -939,6 +950,14 @@ func (sp *StorePool) throttle(reason throttleReason, why string, storeID roachpb
 			log.Infof(ctx, "snapshot failed (%s), s%d will be throttled for %s until %s",
 				why, storeID, timeout, detail.throttledUntil)
 		}
+	case throttleDeclined:
+		timeout := DeclinedSnapshotTimeout.Get(&sp.st.SV)
+		detail.throttledUntil = sp.clock.PhysicalTime().Add(timeout)
+		if log.V(2) {
+			ctx := sp.AnnotateCtx(context.TODO())
+			log.Infof(ctx, "snapshot declined (%s), s%d will be throttled for %s until %s",
+				why, storeID, timeout, detail.throttledUntil)
+		}
 	default:
 		log.Warningf(sp.AnnotateCtx(context.TODO()), "unknown throttle reason %v", reason)
 	}