From f2b7263e596831273b5c0743d732ebed4fdece0d Mon Sep 17 00:00:00 2001
From: Michael Butler <butler@cockroachlabs.com>
Date: Thu, 7 Mar 2024 10:58:22 -0700
Subject: [PATCH 1/4] roachtest: add new prefixless backup fixtures

This patch adds two new backupFixture generators that create 400GB and 8TB
non-revision history prefixless backups to be used for online restore scale
testing.

Epic: None

Release note: none
---
 pkg/cmd/roachtest/tests/backup_fixtures.go | 50 +++++++++++++++++++++-
 1 file changed, 48 insertions(+), 2 deletions(-)

diff --git a/pkg/cmd/roachtest/tests/backup_fixtures.go b/pkg/cmd/roachtest/tests/backup_fixtures.go
index 063a8dd03b3d..f1b0f2921abf 100644
--- a/pkg/cmd/roachtest/tests/backup_fixtures.go
+++ b/pkg/cmd/roachtest/tests/backup_fixtures.go
@@ -118,7 +118,7 @@ type backupFixtureSpecs struct {
 }
 
 func (bf *backupFixtureSpecs) initTestName() {
-	bf.testName = "backupFixture/" + bf.scheduledBackupSpecs.workload.String() + "/" + bf.scheduledBackupSpecs.cloud
+	bf.testName = fmt.Sprintf("backupFixture/%s/revision-history=%t/%s", bf.scheduledBackupSpecs.workload.String(), !bf.scheduledBackupSpecs.nonRevisionHistory, bf.scheduledBackupSpecs.cloud)
 }
 
 func makeBackupDriver(t test.Test, c cluster.Cluster, sp backupFixtureSpecs) backupDriver {
@@ -161,7 +161,7 @@ func (bd *backupDriver) prepareCluster(ctx context.Context) {
 	if !bd.sp.scheduledBackupSpecs.ignoreExistingBackups {
 		// This check allows the roachtest to fail fast, instead of when the
 		// scheduled backup cmd is issued.
-		require.False(bd.t, bd.checkForExistingBackupCollection(ctx))
+		require.False(bd.t, bd.checkForExistingBackupCollection(ctx), fmt.Sprintf("existing backup in collection %s", bd.sp.scheduledBackupSpecs.backupCollection()))
 	}
 }
 
@@ -269,6 +269,52 @@ func registerBackupFixtures(r registry.Registry) {
 			skip:   "only for fixture generation",
 			suites: registry.Suites(registry.Nightly),
 		},
+		{
+			// 400GB backup fixture, no revision history, with 48 incremental layers.
+			// This will used by the online restore roachtests. During 24.2
+			// development, we can use it to enable OR of incremental backups.
+			hardware: makeHardwareSpecs(hardwareSpecs{workloadNode: true}),
+			scheduledBackupSpecs: makeBackupFixtureSpecs(scheduledBackupSpecs{
+				backupSpecs: backupSpecs{
+					version:            fixtureFromMasterVersion,
+					nonRevisionHistory: true,
+				},
+			}),
+			timeout: 5 * time.Hour,
+			initWorkloadViaRestore: &restoreSpecs{
+				backup: backupSpecs{
+					version:           fixtureFromMasterVersion,
+					numBackupsInChain: 48,
+				},
+				restoreUptoIncremental: 12,
+			},
+			skip:   "only for fixture generation",
+			suites: registry.Suites(registry.Nightly),
+		},
+		{
+			// 8TB backup fixture, no revision history, with 48 incremental layers.
+			// This will used by the online restore roachtests. During 24.2
+			// development, we can use it to enable OR of incremental backups.
+			hardware: makeHardwareSpecs(hardwareSpecs{nodes: 10, volumeSize: 1500, workloadNode: true}),
+			scheduledBackupSpecs: makeBackupFixtureSpecs(scheduledBackupSpecs{
+				backupSpecs: backupSpecs{
+					version:            fixtureFromMasterVersion,
+					nonRevisionHistory: true,
+					workload:           tpceRestore{customers: 500000},
+				},
+			}),
+			timeout: 23 * time.Hour,
+			initWorkloadViaRestore: &restoreSpecs{
+				backup: backupSpecs{
+					version:            "v23.1.11",
+					numBackupsInChain:  48,
+					nonRevisionHistory: true,
+				},
+				restoreUptoIncremental: 12,
+			},
+			skip:   "only for fixture generation",
+			suites: registry.Suites(registry.Weekly),
+		},
 		{
 			// 15 GB backup fixture with 48 incremental layers. This is used by
 			// restore/tpce/15GB/aws/nodes=4/cpus=8. Runs weekly to catch any

From b3dec445867c2cb8e052c991f1c20e6caf345d26 Mon Sep 17 00:00:00 2001
From: Austen McClernon <austen@cockroachlabs.com>
Date: Mon, 26 Feb 2024 14:30:33 -0500
Subject: [PATCH 2/4] roachpb,kvserver: add max io overload store capacity

This commit introduces `IOThresholdMax` as a field on `StoreCapacity`.
IOThresholdMax holds the store's maximum IO threshold over the last 5
minutes.

Part of: #118866
Epic: none
Release note: None
---
 pkg/kv/kvserver/BUILD.bazel |  1 +
 pkg/kv/kvserver/store.go    | 26 ++++++++++++++++++++++++--
 pkg/roachpb/metadata.proto  |  3 +++
 3 files changed, 28 insertions(+), 2 deletions(-)

diff --git a/pkg/kv/kvserver/BUILD.bazel b/pkg/kv/kvserver/BUILD.bazel
index 7042698a1ccc..dc6a3f1f68fb 100644
--- a/pkg/kv/kvserver/BUILD.bazel
+++ b/pkg/kv/kvserver/BUILD.bazel
@@ -212,6 +212,7 @@ go_library(
         "//pkg/util/quotapool",
         "//pkg/util/retry",
         "//pkg/util/shuffle",
+        "//pkg/util/slidingwindow",
         "//pkg/util/stop",
         "//pkg/util/syncutil",
         "//pkg/util/syncutil/singleflight",
diff --git a/pkg/kv/kvserver/store.go b/pkg/kv/kvserver/store.go
index 8bf465f646b4..58a3ddb73646 100644
--- a/pkg/kv/kvserver/store.go
+++ b/pkg/kv/kvserver/store.go
@@ -84,6 +84,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/quotapool"
 	"github.com/cockroachdb/cockroach/pkg/util/retry"
 	"github.com/cockroachdb/cockroach/pkg/util/shuffle"
+	"github.com/cockroachdb/cockroach/pkg/util/slidingwindow"
 	"github.com/cockroachdb/cockroach/pkg/util/stop"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil/singleflight"
@@ -1061,7 +1062,10 @@ type Store struct {
 
 	ioThreshold struct {
 		syncutil.Mutex
-		t *admissionpb.IOThreshold // never nil
+		t                 *admissionpb.IOThreshold // never nil
+		maxL0NumSubLevels *slidingwindow.Swag
+		maxL0NumFiles     *slidingwindow.Swag
+		maxL0Size         *slidingwindow.Swag
 	}
 
 	counts struct {
@@ -1413,6 +1417,11 @@ func NewStore(
 		rangeFeedSlowClosedTimestampNudge: singleflight.NewGroup("rangfeed-ct-nudge", "range"),
 	}
 	s.ioThreshold.t = &admissionpb.IOThreshold{}
+	// Track the maxScore over the last 5 minutes, in one minute windows.
+	now := cfg.Clock.Now().GoTime()
+	s.ioThreshold.maxL0NumSubLevels = slidingwindow.NewMaxSwag(now, time.Minute, 5)
+	s.ioThreshold.maxL0NumFiles = slidingwindow.NewMaxSwag(now, time.Minute, 5)
+	s.ioThreshold.maxL0Size = slidingwindow.NewMaxSwag(now, time.Minute, 5)
 	var allocatorStorePool storepool.AllocatorStorePool
 	var storePoolIsDeterministic bool
 	if cfg.StorePool != nil {
@@ -2575,11 +2584,16 @@ func (s *Store) GossipStore(ctx context.Context, useCached bool) error {
 	return s.storeGossip.GossipStore(ctx, useCached)
 }
 
-// UpdateIOThreshold updates the IOThreshold reported in the StoreDescriptor.
+// UpdateIOThreshold updates the IOThreshold and IOThresholdMax reported in the
+// StoreDescriptor.
 func (s *Store) UpdateIOThreshold(ioThreshold *admissionpb.IOThreshold) {
+	now := s.Clock().Now().GoTime()
 	s.ioThreshold.Lock()
 	defer s.ioThreshold.Unlock()
 	s.ioThreshold.t = ioThreshold
+	s.ioThreshold.maxL0NumSubLevels.Record(now, float64(ioThreshold.L0NumSubLevels))
+	s.ioThreshold.maxL0NumFiles.Record(now, float64(ioThreshold.L0NumFiles))
+	s.ioThreshold.maxL0Size.Record(now, float64(ioThreshold.L0Size))
 }
 
 // VisitReplicasOption optionally modifies store.VisitReplicas.
@@ -2925,9 +2939,17 @@ func (s *Store) Capacity(ctx context.Context, useCached bool) (roachpb.StoreCapa
 	capacity.CPUPerSecond = totalStoreCPUTimePerSecond
 	capacity.QueriesPerSecond = totalQueriesPerSecond
 	capacity.WritesPerSecond = totalWritesPerSecond
+	goNow := now.ToTimestamp().GoTime()
 	{
 		s.ioThreshold.Lock()
 		capacity.IOThreshold = *s.ioThreshold.t
+		capacity.IOThresholdMax = *s.ioThreshold.t
+		maxL0NumSubLevels, _ := s.ioThreshold.maxL0NumSubLevels.Query(goNow)
+		maxL0NumFiles, _ := s.ioThreshold.maxL0NumFiles.Query(goNow)
+		maxL0Size, _ := s.ioThreshold.maxL0Size.Query(goNow)
+		capacity.IOThresholdMax.L0NumSubLevels = int64(maxL0NumSubLevels)
+		capacity.IOThresholdMax.L0NumFiles = int64(maxL0NumFiles)
+		capacity.IOThresholdMax.L0Size = int64(maxL0Size)
 		s.ioThreshold.Unlock()
 	}
 	capacity.BytesPerReplica = roachpb.PercentilesFromData(bytesPerReplica)
diff --git a/pkg/roachpb/metadata.proto b/pkg/roachpb/metadata.proto
index 8325b45cf826..e96498b4cca7 100644
--- a/pkg/roachpb/metadata.proto
+++ b/pkg/roachpb/metadata.proto
@@ -323,6 +323,9 @@ message StoreCapacity {
   // tracked in replica stats.
   optional double cpu_per_second = 14 [(gogoproto.nullable) = false, (gogoproto.customname) = "CPUPerSecond"];
   optional cockroach.util.admission.admissionpb.IOThreshold io_threshold = 13 [(gogoproto.nullable) = false, (gogoproto.customname) = "IOThreshold" ];
+  // io_threshold_max tracks the maximum io overload values the store has had
+  // over the last 5 minutes.
+  optional cockroach.util.admission.admissionpb.IOThreshold io_threshold_max = 15 [(gogoproto.nullable) = false, (gogoproto.customname) = "IOThresholdMax" ];
   // 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.

From fe5e53a30c80b2ed53d5d1d5b5781d783e3ab364 Mon Sep 17 00:00:00 2001
From: Austen McClernon <austen@cockroachlabs.com>
Date: Mon, 26 Feb 2024 14:35:42 -0500
Subject: [PATCH 3/4] allocator: use max io threshold score instead of
 instantaneous

The allocator uses the IO threshold score to determine replicas to
filter for rebalancing and lease transfers. Previously, this score would
be the last instantaneous value gossiped. As part of #118866, leases
will be shed upon encountering overload, in addition to the existing
behavior of blocking lease transfers. To avoid leases shedding quickly
from one node to another, consider the maximum IO threshold score
instead.

Note that the use of the maximum IO overload score is gated behind a
cluster version gate. Without the version gate it would be possible for
an upgraded node to continually shed all of its leases based off of the
maximum IO overload score, whilst prior version nodes transfer the lease
back based off the instantaneous.

Part of: #118866
Release note: None
---
 .../settings/settings-for-tenants.txt         |  2 +-
 docs/generated/settings/settings.html         |  2 +-
 pkg/clusterversion/cockroach_versions.go      |  6 +++
 .../allocator/allocatorimpl/BUILD.bazel       |  1 +
 .../allocator/allocatorimpl/allocator.go      | 10 ++--
 .../allocatorimpl/allocator_scorer.go         | 47 +++++++++++++++----
 .../allocator/allocatorimpl/allocator_test.go | 44 +++++++++--------
 .../allocator/storepool/store_pool.go         | 18 ++++++-
 .../allocator/storepool/store_pool_test.go    | 20 +++++---
 pkg/kv/kvserver/asim/state/load.go            | 28 +++++------
 .../asim/tests/datadriven_simulation_test.go  |  2 +-
 pkg/kv/kvserver/store_pool_test.go            | 22 ++++++++-
 pkg/kv/kvserver/store_rebalancer_test.go      | 28 ++++-------
 .../admission/admissionpb/io_threshold.go     |  4 ++
 14 files changed, 154 insertions(+), 80 deletions(-)

diff --git a/docs/generated/settings/settings-for-tenants.txt b/docs/generated/settings/settings-for-tenants.txt
index 4d1804526dd7..2e1fd81b41ff 100644
--- a/docs/generated/settings/settings-for-tenants.txt
+++ b/docs/generated/settings/settings-for-tenants.txt
@@ -337,4 +337,4 @@ trace.snapshot.rate	duration	0s	if non-zero, interval at which background trace
 trace.span_registry.enabled	boolean	true	if set, ongoing traces can be seen at https://<ui>/#/debug/tracez	application
 trace.zipkin.collector	string		the address of a Zipkin instance to receive traces, as <host>:<port>. If no port is specified, 9411 will be used.	application
 ui.display_timezone	enumeration	etc/utc	the timezone used to format timestamps in the ui [etc/utc = 0, america/new_york = 1]	application
-version	version	1000023.2-upgrading-to-1000024.1-step-018	set the active cluster version in the format '<major>.<minor>'	application
+version	version	1000023.2-upgrading-to-1000024.1-step-020	set the active cluster version in the format '<major>.<minor>'	application
diff --git a/docs/generated/settings/settings.html b/docs/generated/settings/settings.html
index f7d9c5a2df5a..67e8c88c0e2d 100644
--- a/docs/generated/settings/settings.html
+++ b/docs/generated/settings/settings.html
@@ -291,6 +291,6 @@
 <tr><td><div id="setting-trace-span-registry-enabled" class="anchored"><code>trace.span_registry.enabled</code></div></td><td>boolean</td><td><code>true</code></td><td>if set, ongoing traces can be seen at https://&lt;ui&gt;/#/debug/tracez</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-trace-zipkin-collector" class="anchored"><code>trace.zipkin.collector</code></div></td><td>string</td><td><code></code></td><td>the address of a Zipkin instance to receive traces, as &lt;host&gt;:&lt;port&gt;. If no port is specified, 9411 will be used.</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-ui-display-timezone" class="anchored"><code>ui.display_timezone</code></div></td><td>enumeration</td><td><code>etc/utc</code></td><td>the timezone used to format timestamps in the ui [etc/utc = 0, america/new_york = 1]</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
-<tr><td><div id="setting-version" class="anchored"><code>version</code></div></td><td>version</td><td><code>1000023.2-upgrading-to-1000024.1-step-018</code></td><td>set the active cluster version in the format &#39;&lt;major&gt;.&lt;minor&gt;&#39;</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
+<tr><td><div id="setting-version" class="anchored"><code>version</code></div></td><td>version</td><td><code>1000023.2-upgrading-to-1000024.1-step-020</code></td><td>set the active cluster version in the format &#39;&lt;major&gt;.&lt;minor&gt;&#39;</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
 </tbody>
 </table>
diff --git a/pkg/clusterversion/cockroach_versions.go b/pkg/clusterversion/cockroach_versions.go
index fab1dcab56f7..197836d6324a 100644
--- a/pkg/clusterversion/cockroach_versions.go
+++ b/pkg/clusterversion/cockroach_versions.go
@@ -296,6 +296,11 @@ const (
 	// database to be SURVIVE ZONE.
 	V24_1_SystemDatabaseSurvivability
 
+	// V24_1_GossipMaximumIOOverload is the version at which stores begin
+	// populating the store capacity field IOThresholdMax. The field shouldn't be
+	// used for allocator decisions before then.
+	V24_1_GossipMaximumIOOverload
+
 	numKeys
 )
 
@@ -361,6 +366,7 @@ var versionTable = [numKeys]roachpb.Version{
 	V24_1_SessionBasedLeasingUpgradeDescriptor: {Major: 23, Minor: 2, Internal: 14},
 	V24_1_PebbleFormatSyntheticPrefixSuffix:    {Major: 23, Minor: 2, Internal: 16},
 	V24_1_SystemDatabaseSurvivability:          {Major: 23, Minor: 2, Internal: 18},
+	V24_1_GossipMaximumIOOverload:              {Major: 23, Minor: 2, Internal: 20},
 }
 
 // Latest is always the highest version key. This is the maximum logical cluster
diff --git a/pkg/kv/kvserver/allocator/allocatorimpl/BUILD.bazel b/pkg/kv/kvserver/allocator/allocatorimpl/BUILD.bazel
index 40188ec61a78..59d5989732e7 100644
--- a/pkg/kv/kvserver/allocator/allocatorimpl/BUILD.bazel
+++ b/pkg/kv/kvserver/allocator/allocatorimpl/BUILD.bazel
@@ -11,6 +11,7 @@ go_library(
     importpath = "github.com/cockroachdb/cockroach/pkg/kv/kvserver/allocator/allocatorimpl",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/clusterversion",
         "//pkg/gossip",
         "//pkg/kv/kvpb",
         "//pkg/kv/kvserver/allocator",
diff --git a/pkg/kv/kvserver/allocator/allocatorimpl/allocator.go b/pkg/kv/kvserver/allocator/allocatorimpl/allocator.go
index e685f0f8375b..8624b8acdebd 100644
--- a/pkg/kv/kvserver/allocator/allocatorimpl/allocator.go
+++ b/pkg/kv/kvserver/allocator/allocatorimpl/allocator.go
@@ -18,6 +18,7 @@ import (
 	"math/rand"
 	"time"
 
+	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/allocator"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/allocator/load"
@@ -2105,7 +2106,6 @@ func (a *Allocator) nonIOOverloadedLeaseTargets(
 	}
 
 	sl, _, _ := storePool.GetStoreListFromIDs(replDescsToStoreIDs(existingReplicas), storepool.StoreFilterSuspect)
-	avgIOOverload := sl.CandidateIOOverloadScores.Mean
 
 	for _, replDesc := range existingReplicas {
 		store, ok := sl.FindStoreByID(replDesc.StoreID)
@@ -2120,7 +2120,7 @@ func (a *Allocator) nonIOOverloadedLeaseTargets(
 		// Instead, we create a buffer between the two to avoid leases moving back
 		// and forth.
 		if (replDesc.StoreID == leaseStoreID) &&
-			(!ok || !ioOverloadOptions.existingLeaseCheck(ctx, store, avgIOOverload)) {
+			(!ok || !ioOverloadOptions.existingLeaseCheck(ctx, store, sl)) {
 			continue
 		}
 
@@ -2128,7 +2128,7 @@ func (a *Allocator) nonIOOverloadedLeaseTargets(
 		// if it is filtered out similar to above, or the replica store doesn't
 		// pass the lease transfer IO overload check.
 		if replDesc.StoreID != leaseStoreID &&
-			(!ok || !ioOverloadOptions.transferLeaseToCheck(ctx, store, avgIOOverload)) {
+			(!ok || !ioOverloadOptions.transferLeaseToCheck(ctx, store, sl)) {
 			continue
 		}
 
@@ -2148,7 +2148,6 @@ func (a *Allocator) leaseholderShouldMoveDueToIOOverload(
 	ioOverloadOptions IOOverloadOptions,
 ) bool {
 	sl, _, _ := storePool.GetStoreListFromIDs(replDescsToStoreIDs(existingReplicas), storepool.StoreFilterSuspect)
-	avgIOOverload := sl.CandidateIOOverloadScores.Mean
 
 	// Check the existing replicas for the leaseholder, if it doesn't meet the
 	// check return that the lease should be moved due to IO overload on the
@@ -2157,7 +2156,7 @@ func (a *Allocator) leaseholderShouldMoveDueToIOOverload(
 	// overloaded.
 	for _, replDesc := range existingReplicas {
 		if store, ok := sl.FindStoreByID(replDesc.StoreID); ok && replDesc.StoreID == leaseStoreID {
-			return !ioOverloadOptions.existingLeaseCheck(ctx, store, avgIOOverload)
+			return !ioOverloadOptions.existingLeaseCheck(ctx, store, sl)
 		}
 	}
 
@@ -2232,6 +2231,7 @@ func (a *Allocator) IOOverloadOptions() IOOverloadOptions {
 	return IOOverloadOptions{
 		ReplicaEnforcementLevel:      IOOverloadEnforcementLevel(ReplicaIOOverloadThresholdEnforcement.Get(&a.st.SV)),
 		LeaseEnforcementLevel:        IOOverloadEnforcementLevel(LeaseIOOverloadThresholdEnforcement.Get(&a.st.SV)),
+		UseIOThresholdMax:            a.st.Version.IsActive(context.Background(), clusterversion.V24_1_GossipMaximumIOOverload),
 		ReplicaIOOverloadThreshold:   ReplicaIOOverloadThreshold.Get(&a.st.SV),
 		LeaseIOOverloadThreshold:     LeaseIOOverloadThreshold.Get(&a.st.SV),
 		LeaseIOOverloadShedThreshold: LeaseIOOverloadShedThreshold.Get(&a.st.SV),
diff --git a/pkg/kv/kvserver/allocator/allocatorimpl/allocator_scorer.go b/pkg/kv/kvserver/allocator/allocatorimpl/allocator_scorer.go
index 3f28948bacb1..141abf6f0b91 100644
--- a/pkg/kv/kvserver/allocator/allocatorimpl/allocator_scorer.go
+++ b/pkg/kv/kvserver/allocator/allocatorimpl/allocator_scorer.go
@@ -1111,7 +1111,7 @@ func rankedCandidateListForAllocation(
 			!options.getIOOverloadOptions().allocateReplicaToCheck(
 				ctx,
 				s,
-				candidateStores.CandidateIOOverloadScores.Mean,
+				candidateStores,
 			) {
 			continue
 		}
@@ -1761,7 +1761,7 @@ func rankedCandidateListForRebalancing(
 				s,
 				// We only wish to compare the IO overload to the
 				// comparable stores average and not the cluster.
-				comparable.candidateSL.CandidateIOOverloadScores.Mean,
+				comparable.candidateSL,
 			)
 			cand.balanceScore = options.balanceScore(comparable.candidateSL, s.Capacity)
 			cand.convergesScore = options.rebalanceToConvergesScore(comparable, s)
@@ -2381,6 +2381,11 @@ type IOOverloadOptions struct {
 	ReplicaEnforcementLevel IOOverloadEnforcementLevel
 	LeaseEnforcementLevel   IOOverloadEnforcementLevel
 
+	// TODO(kvoli): Remove this max protection check after 25.1. In mixed version
+	// clusters, the max IO score is not populated on pre v24.1 nodes. Use the
+	// instantaneous value.
+	UseIOThresholdMax bool
+
 	ReplicaIOOverloadThreshold   float64
 	LeaseIOOverloadThreshold     float64
 	LeaseIOOverloadShedThreshold float64
@@ -2411,13 +2416,32 @@ func ioOverloadCheck(
 	return true, ""
 }
 
+func (o IOOverloadOptions) storeScore(store roachpb.StoreDescriptor) float64 {
+	var score float64
+	if o.UseIOThresholdMax {
+		score, _ = store.Capacity.IOThresholdMax.Score()
+	} else {
+		score, _ = store.Capacity.IOThreshold.Score()
+	}
+
+	return score
+}
+
+func (o IOOverloadOptions) storeListAvgScore(storeList storepool.StoreList) float64 {
+	if o.UseIOThresholdMax {
+		return storeList.CandidateMaxIOOverloadScores.Mean
+	}
+	return storeList.CandidateIOOverloadScores.Mean
+}
+
 // allocateReplicaToCheck returns true if the store IO overload does not exceed
 // the cluster threshold and mean, or the enforcement level does not prevent
 // replica allocation to IO overloaded stores.
 func (o IOOverloadOptions) allocateReplicaToCheck(
-	ctx context.Context, store roachpb.StoreDescriptor, avg float64,
+	ctx context.Context, store roachpb.StoreDescriptor, storeList storepool.StoreList,
 ) bool {
-	score, _ := store.Capacity.IOThreshold.Score()
+	score := o.storeScore(store)
+	avg := o.storeListAvgScore(storeList)
 
 	if ok, reason := ioOverloadCheck(score, avg,
 		o.ReplicaIOOverloadThreshold, IOOverloadMeanThreshold,
@@ -2435,9 +2459,10 @@ func (o IOOverloadOptions) allocateReplicaToCheck(
 // exceed the cluster threshold and mean, or the enforcement level does not
 // prevent replica rebalancing to IO overloaded stores.
 func (o IOOverloadOptions) rebalanceReplicaToCheck(
-	ctx context.Context, store roachpb.StoreDescriptor, avg float64,
+	ctx context.Context, store roachpb.StoreDescriptor, storeList storepool.StoreList,
 ) bool {
-	score, _ := store.Capacity.IOThreshold.Score()
+	score := o.storeScore(store)
+	avg := o.storeListAvgScore(storeList)
 
 	if ok, reason := ioOverloadCheck(score, avg,
 		o.ReplicaIOOverloadThreshold, IOOverloadMeanThreshold,
@@ -2454,9 +2479,10 @@ func (o IOOverloadOptions) rebalanceReplicaToCheck(
 // the cluster threshold and mean, or the enforcement level does not prevent
 // lease transfers to IO overloaded stores.
 func (o IOOverloadOptions) transferLeaseToCheck(
-	ctx context.Context, store roachpb.StoreDescriptor, avg float64,
+	ctx context.Context, store roachpb.StoreDescriptor, storeList storepool.StoreList,
 ) bool {
-	score, _ := store.Capacity.IOThreshold.Score()
+	score := o.storeScore(store)
+	avg := o.storeListAvgScore(storeList)
 
 	if ok, reason := ioOverloadCheck(score, avg,
 		o.LeaseIOOverloadThreshold, IOOverloadMeanThreshold,
@@ -2474,9 +2500,10 @@ func (o IOOverloadOptions) transferLeaseToCheck(
 // the cluster threshold and mean, or the enforcement level does not prevent
 // existing stores from holidng leases whilst being IO overloaded.
 func (o IOOverloadOptions) existingLeaseCheck(
-	ctx context.Context, store roachpb.StoreDescriptor, avg float64,
+	ctx context.Context, store roachpb.StoreDescriptor, storeList storepool.StoreList,
 ) bool {
-	score, _ := store.Capacity.IOThreshold.Score()
+	score := o.storeScore(store)
+	avg := o.storeListAvgScore(storeList)
 
 	if ok, reason := ioOverloadCheck(score, avg,
 		o.LeaseIOOverloadShedThreshold, IOOverloadMeanThreshold,
diff --git a/pkg/kv/kvserver/allocator/allocatorimpl/allocator_test.go b/pkg/kv/kvserver/allocator/allocatorimpl/allocator_test.go
index 3a3a7a3c7c0c..947cb6a58926 100644
--- a/pkg/kv/kvserver/allocator/allocatorimpl/allocator_test.go
+++ b/pkg/kv/kvserver/allocator/allocatorimpl/allocator_test.go
@@ -349,22 +349,22 @@ var oneStoreHighIOOverload = []*roachpb.StoreDescriptor{
 	{
 		StoreID:  1,
 		Node:     roachpb.NodeDescriptor{NodeID: 1},
-		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 600, IOThreshold: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold - 5)},
+		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 600, IOThresholdMax: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold - 5)},
 	},
 	{
 		StoreID:  2,
 		Node:     roachpb.NodeDescriptor{NodeID: 2},
-		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 1800, IOThreshold: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold - 5)},
+		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 1800, IOThresholdMax: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold - 5)},
 	},
 	{
 		StoreID:  3,
 		Node:     roachpb.NodeDescriptor{NodeID: 3},
-		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 600, IOThreshold: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold + 5)},
+		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 600, IOThresholdMax: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold + 5)},
 	},
 	{
 		StoreID:  4,
 		Node:     roachpb.NodeDescriptor{NodeID: 4},
-		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 1200, IOThreshold: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold - 5)},
+		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 1200, IOThresholdMax: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold - 5)},
 	},
 }
 
@@ -372,17 +372,17 @@ var allStoresHighIOOverload = []*roachpb.StoreDescriptor{
 	{
 		StoreID:  1,
 		Node:     roachpb.NodeDescriptor{NodeID: 1},
-		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 1200, IOThreshold: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold + 1)},
+		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 1200, IOThresholdMax: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold + 1)},
 	},
 	{
 		StoreID:  2,
 		Node:     roachpb.NodeDescriptor{NodeID: 2},
-		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 800, IOThreshold: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold + 1)},
+		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 800, IOThresholdMax: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold + 1)},
 	},
 	{
 		StoreID:  3,
 		Node:     roachpb.NodeDescriptor{NodeID: 3},
-		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 600, IOThreshold: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold + 1)},
+		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 600, IOThresholdMax: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold + 1)},
 	},
 }
 
@@ -390,17 +390,17 @@ var allStoresHighIOOverloadSkewed = []*roachpb.StoreDescriptor{
 	{
 		StoreID:  1,
 		Node:     roachpb.NodeDescriptor{NodeID: 1},
-		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 1200, IOThreshold: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold + 1)},
+		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 1200, IOThresholdMax: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold + 1)},
 	},
 	{
 		StoreID:  2,
 		Node:     roachpb.NodeDescriptor{NodeID: 2},
-		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 800, IOThreshold: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold + 50)},
+		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 800, IOThresholdMax: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold + 50)},
 	},
 	{
 		StoreID:  3,
 		Node:     roachpb.NodeDescriptor{NodeID: 3},
-		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 600, IOThreshold: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold + 55)},
+		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 600, IOThresholdMax: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold + 55)},
 	},
 }
 
@@ -408,27 +408,27 @@ var threeStoresHighIOOverloadAscRangeCount = []*roachpb.StoreDescriptor{
 	{
 		StoreID:  1,
 		Node:     roachpb.NodeDescriptor{NodeID: 1},
-		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 100, IOThreshold: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold + 10)},
+		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 100, IOThresholdMax: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold + 10)},
 	},
 	{
 		StoreID:  2,
 		Node:     roachpb.NodeDescriptor{NodeID: 2},
-		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 400, IOThreshold: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold + 10)},
+		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 400, IOThresholdMax: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold + 10)},
 	},
 	{
 		StoreID:  3,
 		Node:     roachpb.NodeDescriptor{NodeID: 3},
-		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 1600, IOThreshold: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold + 10)},
+		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 1600, IOThresholdMax: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold + 10)},
 	},
 	{
 		StoreID:  4,
 		Node:     roachpb.NodeDescriptor{NodeID: 4},
-		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 6400, IOThreshold: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold - 10)},
+		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 6400, IOThresholdMax: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold - 10)},
 	},
 	{
 		StoreID:  5,
 		Node:     roachpb.NodeDescriptor{NodeID: 5},
-		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 25000, IOThreshold: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold - 10)},
+		Capacity: roachpb.StoreCapacity{Capacity: 200, Available: 200, RangeCount: 25000, IOThresholdMax: TestingIOThresholdWithScore(DefaultReplicaIOOverloadThreshold - 10)},
 	},
 }
 
@@ -2081,8 +2081,8 @@ func TestAllocatorTransferLeaseTargetIOOverloadCheck(t *testing.T) {
 					StoreID: roachpb.StoreID(i + 1),
 					Node:    roachpb.NodeDescriptor{NodeID: roachpb.NodeID(i + 1)},
 					Capacity: roachpb.StoreCapacity{
-						LeaseCount:  int32(tc.leaseCounts[i]),
-						IOThreshold: TestingIOThresholdWithScore(tc.IOScores[i]),
+						LeaseCount:     int32(tc.leaseCounts[i]),
+						IOThresholdMax: TestingIOThresholdWithScore(tc.IOScores[i]),
 					},
 				}
 			}
@@ -2912,8 +2912,8 @@ func TestAllocatorShouldTransferLeaseIOOverload(t *testing.T) {
 					StoreID: roachpb.StoreID(i + 1),
 					Node:    roachpb.NodeDescriptor{NodeID: roachpb.NodeID(i + 1)},
 					Capacity: roachpb.StoreCapacity{
-						LeaseCount:  int32(tc.leaseCounts[i]),
-						IOThreshold: TestingIOThresholdWithScore(tc.IOScores[i]),
+						LeaseCount:     int32(tc.leaseCounts[i]),
+						IOThresholdMax: TestingIOThresholdWithScore(tc.IOScores[i]),
 					},
 				}
 			}
@@ -4698,7 +4698,11 @@ func TestAllocatorRebalanceIOOverloadCheck(t *testing.T) {
 			sg.GossipStores(test.stores, t)
 			// Enable read disk health checking in candidate exclusion.
 			options := a.ScorerOptions(ctx)
-			options.IOOverloadOptions = IOOverloadOptions{ReplicaEnforcementLevel: test.enforcement, ReplicaIOOverloadThreshold: 1}
+			options.IOOverloadOptions = IOOverloadOptions{
+				ReplicaEnforcementLevel:    test.enforcement,
+				ReplicaIOOverloadThreshold: 1,
+				UseIOThresholdMax:          true,
+			}
 			add, remove, _, ok := a.RebalanceVoter(
 				ctx,
 				sp,
diff --git a/pkg/kv/kvserver/allocator/storepool/store_pool.go b/pkg/kv/kvserver/allocator/storepool/store_pool.go
index 9d15b58263ad..6a816d136f4c 100644
--- a/pkg/kv/kvserver/allocator/storepool/store_pool.go
+++ b/pkg/kv/kvserver/allocator/storepool/store_pool.go
@@ -961,6 +961,10 @@ type StoreList struct {
 	// CandidateIOOverloadScores tracks the IO overload stats for Stores that are
 	// eligible to be rebalance candidates.
 	CandidateIOOverloadScores Stat
+
+	// CandidateMaxIOOverloadScores tracks the max IO overload stats for Stores
+	// that are eligible to be rebalance candidates.
+	CandidateMaxIOOverloadScores Stat
 }
 
 // MakeStoreList constructs a new store list based on the passed in descriptors.
@@ -976,6 +980,8 @@ func MakeStoreList(descriptors []roachpb.StoreDescriptor) StoreList {
 		sl.CandidateCPU.update(desc.Capacity.CPUPerSecond)
 		score, _ := desc.Capacity.IOThreshold.Score()
 		sl.CandidateIOOverloadScores.update(score)
+		maxScore, _ := desc.Capacity.IOThresholdMax.Score()
+		sl.CandidateMaxIOOverloadScores.update(maxScore)
 	}
 	return sl
 }
@@ -988,12 +994,16 @@ func (sl StoreList) String() string {
 func (sl StoreList) SafeFormat(w redact.SafePrinter, _ rune) {
 	var buf redact.StringBuilder
 	buf.Printf(
-		"  candidate: avg-ranges=%.2f avg-leases=%.2f avg-disk-usage=%s avg-queries-per-second=%.2f avg-store-cpu-per-second=%s",
+		"  candidate: avg-ranges=%.2f avg-leases=%.2f avg-disk-usage=%s "+
+			"avg-queries-per-second=%.2f avg-store-cpu-per-second=%s "+
+			"avg-io-overload=%.2f(max=%.2f)",
 		sl.CandidateRanges.Mean,
 		sl.CandidateLeases.Mean,
 		humanizeutil.IBytes(int64(sl.candidateLogicalBytes.Mean)),
 		sl.CandidateQueriesPerSecond.Mean,
 		humanizeutil.Duration(time.Duration(int64(sl.CandidateCPU.Mean))),
+		sl.CandidateIOOverloadScores.Mean,
+		sl.CandidateMaxIOOverloadScores.Mean,
 	)
 	if len(sl.Stores) > 0 {
 		buf.Printf("\n")
@@ -1002,12 +1012,16 @@ func (sl StoreList) SafeFormat(w redact.SafePrinter, _ rune) {
 	}
 	for _, desc := range sl.Stores {
 		ioScore, _ := desc.Capacity.IOThreshold.Score()
-		buf.Printf("  %v: ranges=%d leases=%d disk-usage=%s queries-per-second=%.2f store-cpu-per-second=%s io-overload=%.2f\n",
+		maxIOScore, _ := desc.Capacity.IOThresholdMax.Score()
+		buf.Printf(
+			"  %v: ranges=%d leases=%d disk-usage=%s queries-per-second=%.2f "+
+				"store-cpu-per-second=%s io-overload=%.2f(max=%.2f)\n",
 			desc.StoreID, desc.Capacity.RangeCount,
 			desc.Capacity.LeaseCount, humanizeutil.IBytes(desc.Capacity.LogicalBytes),
 			desc.Capacity.QueriesPerSecond,
 			humanizeutil.Duration(time.Duration(int64(desc.Capacity.CPUPerSecond))),
 			ioScore,
+			maxIOScore,
 		)
 	}
 	w.Print(buf)
diff --git a/pkg/kv/kvserver/allocator/storepool/store_pool_test.go b/pkg/kv/kvserver/allocator/storepool/store_pool_test.go
index 82d471367e81..4abe2e2ae46b 100644
--- a/pkg/kv/kvserver/allocator/storepool/store_pool_test.go
+++ b/pkg/kv/kvserver/allocator/storepool/store_pool_test.go
@@ -943,19 +943,25 @@ func TestStoreListString(t *testing.T) {
 					L0NumFiles:              int64(i),
 					L0NumFilesThreshold:     scale,
 				},
+				IOThresholdMax: admissionpb.IOThreshold{
+					L0NumSubLevels:          10 * int64(i),
+					L0NumSubLevelsThreshold: scale,
+					L0NumFiles:              10 * int64(i),
+					L0NumFilesThreshold:     scale,
+				},
 			},
 		})
 	}
 
 	require.Equal(t,
-		"  candidate: avg-ranges=0.00 avg-leases=0.00 avg-disk-usage=0 B avg-queries-per-second=0.00 avg-store-cpu-per-second=0µs <no candidates>",
+		"  candidate: avg-ranges=0.00 avg-leases=0.00 avg-disk-usage=0 B avg-queries-per-second=0.00 avg-store-cpu-per-second=0µs avg-io-overload=0.00(max=0.00) <no candidates>",
 		MakeStoreList([]roachpb.StoreDescriptor{}).String())
 
-	require.Equal(t, "  candidate: avg-ranges=30.00 avg-leases=30.00 avg-disk-usage=3.0 KiB avg-queries-per-second=30.00 avg-store-cpu-per-second=3µs\n"+
-		"  1: ranges=10 leases=10 disk-usage=1.0 KiB queries-per-second=10.00 store-cpu-per-second=1µs io-overload=0.10\n"+
-		"  2: ranges=20 leases=20 disk-usage=2.0 KiB queries-per-second=20.00 store-cpu-per-second=2µs io-overload=0.20\n"+
-		"  3: ranges=30 leases=30 disk-usage=3.0 KiB queries-per-second=30.00 store-cpu-per-second=3µs io-overload=0.30\n"+
-		"  4: ranges=40 leases=40 disk-usage=4.0 KiB queries-per-second=40.00 store-cpu-per-second=4µs io-overload=0.40\n"+
-		"  5: ranges=50 leases=50 disk-usage=5.0 KiB queries-per-second=50.00 store-cpu-per-second=5µs io-overload=0.50\n",
+	require.Equal(t, "  candidate: avg-ranges=30.00 avg-leases=30.00 avg-disk-usage=3.0 KiB avg-queries-per-second=30.00 avg-store-cpu-per-second=3µs avg-io-overload=0.30(max=3.00)\n"+
+		"  1: ranges=10 leases=10 disk-usage=1.0 KiB queries-per-second=10.00 store-cpu-per-second=1µs io-overload=0.10(max=1.00)\n"+
+		"  2: ranges=20 leases=20 disk-usage=2.0 KiB queries-per-second=20.00 store-cpu-per-second=2µs io-overload=0.20(max=2.00)\n"+
+		"  3: ranges=30 leases=30 disk-usage=3.0 KiB queries-per-second=30.00 store-cpu-per-second=3µs io-overload=0.30(max=3.00)\n"+
+		"  4: ranges=40 leases=40 disk-usage=4.0 KiB queries-per-second=40.00 store-cpu-per-second=4µs io-overload=0.40(max=4.00)\n"+
+		"  5: ranges=50 leases=50 disk-usage=5.0 KiB queries-per-second=50.00 store-cpu-per-second=5µs io-overload=0.50(max=5.00)\n",
 		MakeStoreList(stores).String())
 }
diff --git a/pkg/kv/kvserver/asim/state/load.go b/pkg/kv/kvserver/asim/state/load.go
index 485430e1f65b..7240ad372a5a 100644
--- a/pkg/kv/kvserver/asim/state/load.go
+++ b/pkg/kv/kvserver/asim/state/load.go
@@ -137,7 +137,7 @@ func NewCapacityOverride() CapacityOverride {
 		QueriesPerSecond: capacityOverrideSentinel,
 		WritesPerSecond:  capacityOverrideSentinel,
 		CPUPerSecond:     capacityOverrideSentinel,
-		IOThreshold: admissionpb.IOThreshold{
+		IOThresholdMax: admissionpb.IOThreshold{
 			L0NumSubLevels:           capacityOverrideSentinel,
 			L0NumSubLevelsThreshold:  capacityOverrideSentinel,
 			L0NumFiles:               capacityOverrideSentinel,
@@ -151,7 +151,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_threhold=%v",
+			"queries_per_sec=%.2f, writes_per_sec=%.2f, cpu_per_sec=%.2f, io_threshold_max=%v",
 		co.Capacity,
 		co.Available,
 		co.Used,
@@ -161,7 +161,7 @@ func (co CapacityOverride) String() string {
 		co.QueriesPerSecond,
 		co.WritesPerSecond,
 		co.CPUPerSecond,
-		co.IOThreshold,
+		co.IOThresholdMax,
 	)
 }
 
@@ -196,23 +196,23 @@ func mergeOverride(
 	if override.CPUPerSecond != capacityOverrideSentinel {
 		ret.CPUPerSecond = override.CPUPerSecond
 	}
-	if override.IOThreshold.L0NumFiles != capacityOverrideSentinel {
-		ret.IOThreshold.L0NumFiles = override.IOThreshold.L0NumFiles
+	if override.IOThresholdMax.L0NumFiles != capacityOverrideSentinel {
+		ret.IOThresholdMax.L0NumFiles = override.IOThresholdMax.L0NumFiles
 	}
-	if override.IOThreshold.L0NumFilesThreshold != capacityOverrideSentinel {
-		ret.IOThreshold.L0NumFilesThreshold = override.IOThreshold.L0NumFilesThreshold
+	if override.IOThresholdMax.L0NumFilesThreshold != capacityOverrideSentinel {
+		ret.IOThresholdMax.L0NumFilesThreshold = override.IOThresholdMax.L0NumFilesThreshold
 	}
 	if override.IOThreshold.L0NumSubLevels != capacityOverrideSentinel {
-		ret.IOThreshold.L0NumSubLevels = override.IOThreshold.L0NumSubLevels
+		ret.IOThresholdMax.L0NumSubLevels = override.IOThresholdMax.L0NumSubLevels
 	}
-	if override.IOThreshold.L0NumSubLevelsThreshold != capacityOverrideSentinel {
-		ret.IOThreshold.L0NumSubLevelsThreshold = override.IOThreshold.L0NumSubLevelsThreshold
+	if override.IOThresholdMax.L0NumSubLevelsThreshold != capacityOverrideSentinel {
+		ret.IOThresholdMax.L0NumSubLevelsThreshold = override.IOThresholdMax.L0NumSubLevelsThreshold
 	}
-	if override.IOThreshold.L0Size != capacityOverrideSentinel {
-		ret.IOThreshold.L0Size = override.IOThreshold.L0Size
+	if override.IOThresholdMax.L0Size != capacityOverrideSentinel {
+		ret.IOThresholdMax.L0Size = override.IOThresholdMax.L0Size
 	}
-	if override.IOThreshold.L0MinimumSizePerSubLevel != capacityOverrideSentinel {
-		ret.IOThreshold.L0MinimumSizePerSubLevel = override.IOThreshold.L0MinimumSizePerSubLevel
+	if override.IOThresholdMax.L0MinimumSizePerSubLevel != capacityOverrideSentinel {
+		ret.IOThresholdMax.L0MinimumSizePerSubLevel = override.IOThresholdMax.L0MinimumSizePerSubLevel
 	}
 	return ret
 }
diff --git a/pkg/kv/kvserver/asim/tests/datadriven_simulation_test.go b/pkg/kv/kvserver/asim/tests/datadriven_simulation_test.go
index bd8ee4961082..6f1c1455dec1 100644
--- a/pkg/kv/kvserver/asim/tests/datadriven_simulation_test.go
+++ b/pkg/kv/kvserver/asim/tests/datadriven_simulation_test.go
@@ -319,7 +319,7 @@ func TestDataDriven(t *testing.T) {
 				capacityOverride.Capacity = capacity
 				capacityOverride.Available = available
 				if ioThreshold != -1 {
-					capacityOverride.IOThreshold = allocatorimpl.TestingIOThresholdWithScore(ioThreshold)
+					capacityOverride.IOThresholdMax = allocatorimpl.TestingIOThresholdWithScore(ioThreshold)
 				}
 				eventGen.ScheduleEvent(settingsGen.Settings.StartTime, delay, event.SetCapacityOverrideEvent{
 					StoreID:          state.StoreID(store),
diff --git a/pkg/kv/kvserver/store_pool_test.go b/pkg/kv/kvserver/store_pool_test.go
index 78baa02a3949..d78985ffbd2b 100644
--- a/pkg/kv/kvserver/store_pool_test.go
+++ b/pkg/kv/kvserver/store_pool_test.go
@@ -68,6 +68,12 @@ func TestStorePoolUpdateLocalStore(t *testing.T) {
 					L0NumFiles:              5,
 					L0NumFilesThreshold:     1000,
 				},
+				IOThresholdMax: admissionpb.IOThreshold{
+					L0NumSubLevels:          5,
+					L0NumSubLevelsThreshold: 20,
+					L0NumFiles:              5,
+					L0NumFilesThreshold:     1000,
+				},
 			},
 		},
 		{
@@ -87,6 +93,12 @@ func TestStorePoolUpdateLocalStore(t *testing.T) {
 					L0NumFiles:              10,
 					L0NumFilesThreshold:     1000,
 				},
+				IOThresholdMax: admissionpb.IOThreshold{
+					L0NumSubLevels:          10,
+					L0NumSubLevelsThreshold: 20,
+					L0NumFiles:              10,
+					L0NumFilesThreshold:     1000,
+				},
 			},
 		},
 	}
@@ -143,7 +155,11 @@ func TestStorePoolUpdateLocalStore(t *testing.T) {
 		t.Errorf("expected WritesPerSecond %f, but got %f", expectedWPS, desc.Capacity.WritesPerSecond)
 	}
 	if expectedNumL0Sublevels := int64(5); desc.Capacity.IOThreshold.L0NumSubLevels != expectedNumL0Sublevels {
-		t.Errorf("expected L0 Sub-Levels %d, but got %d", expectedNumL0Sublevels, desc.Capacity.IOThreshold.L0NumFiles)
+		t.Errorf("expected L0 Sub-Levels %d, but got %d", expectedNumL0Sublevels, desc.Capacity.IOThreshold.L0NumSubLevels)
+	}
+	ioScoreMax, _ := desc.Capacity.IOThresholdMax.Score()
+	if expectedIOThresholdScoreMax := 0.25; ioScoreMax != expectedIOThresholdScoreMax {
+		t.Errorf("expected IOThresholdMax score %f, but got %f", expectedIOThresholdScoreMax, ioScoreMax)
 	}
 
 	sp.UpdateLocalStoreAfterRebalance(roachpb.StoreID(2), rangeUsageInfo, roachpb.REMOVE_VOTER)
@@ -166,6 +182,10 @@ func TestStorePoolUpdateLocalStore(t *testing.T) {
 	if expectedNumL0Sublevels := int64(10); desc.Capacity.IOThreshold.L0NumSubLevels != expectedNumL0Sublevels {
 		t.Errorf("expected L0 Sub-Levels %d, but got %d", expectedNumL0Sublevels, desc.Capacity.IOThreshold.L0NumFiles)
 	}
+	ioScoreMax, _ = desc.Capacity.IOThresholdMax.Score()
+	if expectedIOThresholdScoreMax := 0.5; ioScoreMax != expectedIOThresholdScoreMax {
+		t.Errorf("expected IOThresholdMax score %f, but got %f", expectedIOThresholdScoreMax, ioScoreMax)
+	}
 
 	sp.UpdateLocalStoresAfterLeaseTransfer(roachpb.StoreID(1), roachpb.StoreID(2), rangeUsageInfo)
 	desc, ok = sp.GetStoreDescriptor(roachpb.StoreID(1))
diff --git a/pkg/kv/kvserver/store_rebalancer_test.go b/pkg/kv/kvserver/store_rebalancer_test.go
index af6712b359d9..5f908d54dc19 100644
--- a/pkg/kv/kvserver/store_rebalancer_test.go
+++ b/pkg/kv/kvserver/store_rebalancer_test.go
@@ -61,8 +61,7 @@ var (
 			Capacity: roachpb.StoreCapacity{
 				QueriesPerSecond: 3000,
 				CPUPerSecond:     3000 * float64(time.Millisecond),
-				IOThreshold: allocatorimpl.TestingIOThresholdWithScore(
-					allocatorimpl.DefaultReplicaIOOverloadThreshold - 10),
+				IOThresholdMax:   allocatorimpl.TestingIOThresholdWithScore(allocatorimpl.DefaultReplicaIOOverloadThreshold - 10),
 			},
 		},
 		{
@@ -81,8 +80,7 @@ var (
 			Capacity: roachpb.StoreCapacity{
 				QueriesPerSecond: 2800,
 				CPUPerSecond:     2800 * float64(time.Millisecond),
-				IOThreshold: allocatorimpl.TestingIOThresholdWithScore(
-					allocatorimpl.DefaultReplicaIOOverloadThreshold - 5),
+				IOThresholdMax:   allocatorimpl.TestingIOThresholdWithScore(allocatorimpl.DefaultReplicaIOOverloadThreshold - 5),
 			},
 		},
 		{
@@ -101,8 +99,7 @@ var (
 			Capacity: roachpb.StoreCapacity{
 				QueriesPerSecond: 2600,
 				CPUPerSecond:     2600 * float64(time.Millisecond),
-				IOThreshold: allocatorimpl.TestingIOThresholdWithScore(
-					allocatorimpl.DefaultReplicaIOOverloadThreshold + 2),
+				IOThresholdMax:   allocatorimpl.TestingIOThresholdWithScore(allocatorimpl.DefaultReplicaIOOverloadThreshold + 2),
 			},
 		},
 		{
@@ -121,8 +118,7 @@ var (
 			Capacity: roachpb.StoreCapacity{
 				QueriesPerSecond: 2400,
 				CPUPerSecond:     2400 * float64(time.Millisecond),
-				IOThreshold: allocatorimpl.TestingIOThresholdWithScore(
-					allocatorimpl.DefaultReplicaIOOverloadThreshold - 10),
+				IOThresholdMax:   allocatorimpl.TestingIOThresholdWithScore(allocatorimpl.DefaultReplicaIOOverloadThreshold - 10),
 			},
 		},
 		{
@@ -141,8 +137,7 @@ var (
 			Capacity: roachpb.StoreCapacity{
 				QueriesPerSecond: 2200,
 				CPUPerSecond:     2200 * float64(time.Millisecond),
-				IOThreshold: allocatorimpl.TestingIOThresholdWithScore(
-					allocatorimpl.DefaultReplicaIOOverloadThreshold - 3),
+				IOThresholdMax:   allocatorimpl.TestingIOThresholdWithScore(allocatorimpl.DefaultReplicaIOOverloadThreshold - 3),
 			},
 		},
 		{
@@ -161,8 +156,7 @@ var (
 			Capacity: roachpb.StoreCapacity{
 				QueriesPerSecond: 2000,
 				CPUPerSecond:     2000 * float64(time.Millisecond),
-				IOThreshold: allocatorimpl.TestingIOThresholdWithScore(
-					allocatorimpl.DefaultReplicaIOOverloadThreshold + 2),
+				IOThresholdMax:   allocatorimpl.TestingIOThresholdWithScore(allocatorimpl.DefaultReplicaIOOverloadThreshold + 2),
 			},
 		},
 		{
@@ -181,8 +175,7 @@ var (
 			Capacity: roachpb.StoreCapacity{
 				QueriesPerSecond: 1800,
 				CPUPerSecond:     1800 * float64(time.Millisecond),
-				IOThreshold: allocatorimpl.TestingIOThresholdWithScore(
-					allocatorimpl.DefaultReplicaIOOverloadThreshold - 10),
+				IOThresholdMax:   allocatorimpl.TestingIOThresholdWithScore(allocatorimpl.DefaultReplicaIOOverloadThreshold - 10),
 			},
 		},
 		{
@@ -201,8 +194,7 @@ var (
 			Capacity: roachpb.StoreCapacity{
 				QueriesPerSecond: 1600,
 				CPUPerSecond:     1600 * float64(time.Millisecond),
-				IOThreshold: allocatorimpl.TestingIOThresholdWithScore(
-					allocatorimpl.DefaultReplicaIOOverloadThreshold - 5),
+				IOThresholdMax:   allocatorimpl.TestingIOThresholdWithScore(allocatorimpl.DefaultReplicaIOOverloadThreshold - 5),
 			},
 		},
 		{
@@ -221,8 +213,7 @@ var (
 			Capacity: roachpb.StoreCapacity{
 				QueriesPerSecond: 1400,
 				CPUPerSecond:     1400 * float64(time.Millisecond),
-				IOThreshold: allocatorimpl.TestingIOThresholdWithScore(
-					allocatorimpl.DefaultReplicaIOOverloadThreshold + 3),
+				IOThresholdMax:   allocatorimpl.TestingIOThresholdWithScore(allocatorimpl.DefaultReplicaIOOverloadThreshold + 3),
 			},
 		},
 	}
@@ -1284,6 +1275,7 @@ func TestChooseRangeToRebalanceAcrossHeterogeneousZones(t *testing.T) {
 			rctx := sr.NewRebalanceContext(ctx, options, hottestRanges, LBRebalancingLeasesAndReplicas)
 			rctx.options.IOOverloadOptions = allocatorimpl.IOOverloadOptions{
 				ReplicaEnforcementLevel: allocatorimpl.IOOverloadThresholdBlockTransfers,
+				UseIOThresholdMax:       true,
 			}
 			rctx.options.LoadThreshold = allocatorimpl.WithAllDims(0.05)
 
diff --git a/pkg/util/admission/admissionpb/io_threshold.go b/pkg/util/admission/admissionpb/io_threshold.go
index 5e7c50317cb9..defb2e083e85 100644
--- a/pkg/util/admission/admissionpb/io_threshold.go
+++ b/pkg/util/admission/admissionpb/io_threshold.go
@@ -30,6 +30,10 @@ import (
 // to compactions falling behind (though that may change if we increase the
 // max number of compactions). And we will need to incorporate overload due to
 // disk bandwidth bottleneck.
+//
+// NOTE: Future updates to the scoring function should be version gated as the
+// threshold is gossiped and used to determine lease/replica placement via the
+// allocator.
 func (iot *IOThreshold) Score() (float64, bool) {
 	// iot.L0NumFilesThreshold and iot.L0NumSubLevelsThreshold are initialized to
 	// 0 by default, and there appears to be a period of time before we update

From 78b3141eee758c7c7b85946b8843c1322cc5f461 Mon Sep 17 00:00:00 2001
From: Michael Butler <butler@cockroachlabs.com>
Date: Wed, 13 Mar 2024 13:41:49 -0600
Subject: [PATCH 4/4] c2c: prevent NPE in SHOW TENANT WITH REPLICATION STATUS

If the user ran SHOW TENANT ... WITH REPLICATION STATUS before the stream
ingestion job pts was set, a null pointer exception would occur and crash the
node. This patch fixes this bug.

Epic: None

Release note: none
---
 pkg/sql/show_tenant.go | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/pkg/sql/show_tenant.go b/pkg/sql/show_tenant.go
index a3a305187aae..fe287be283a2 100644
--- a/pkg/sql/show_tenant.go
+++ b/pkg/sql/show_tenant.go
@@ -168,8 +168,9 @@ func (n *showTenantNode) getTenantValues(
 						// Protected timestamp might not be set yet, no need to fail.
 						log.Warningf(params.ctx, "protected timestamp unavailable for tenant %q and job %d: %v",
 							tenantInfo.Name, jobId, err)
+					} else {
+						values.protectedTimestamp = record.Timestamp
 					}
-					values.protectedTimestamp = record.Timestamp
 				}
 			}
 		case mtinfopb.DataStateReady, mtinfopb.DataStateDrop: