storage: Continue trying replica scatter when stores are throttled

Release note: None

pkg/storage/allocator.go

@@ -150,6 +150,21 @@ func (*allocatorError) purgatoryErrorMarker() {}
 
 var _ purgatoryError = &allocatorError{}
 
+type throttledStoresError struct {
+	numThrottledStores int
+}
+
+func (t *throttledStoresError) Error() string {
+	return fmt.Sprintf("%d matching stores are currently throttled", t.numThrottledStores)
+}
+
+// IsThrottledStoresError returns true iff the error indicates the allocator
+// couldn't take action because too many stores were throttled.
+func IsThrottledStoresError(err error) bool {
+	_, ok := err.(*throttledStoresError)
+	return ok
+}
+
 // allocatorRand pairs a rand.Rand with a mutex.
 // NOTE: Allocator is typically only accessed from a single thread (the
 // replication queue), but this assumption is broken in tests which force
@@ -361,7 +376,7 @@ func (a *Allocator) AllocateTarget(
 	// When there are throttled stores that do match, we shouldn't send
 	// the replica to purgatory.
 	if throttledStoreCount > 0 {
-		return nil, "", errors.Errorf("%d matching stores are currently throttled", throttledStoreCount)
+		return nil, "", &throttledStoresError{numThrottledStores: throttledStoreCount}
 	}
 	return nil, "", &allocatorError{
 		constraints:     zone.Constraints,

pkg/storage/replica_command.go

@@ -2129,12 +2129,12 @@ func (r *Replica) adminScatter(
 		for re := retry.StartWithCtx(ctx, retryOpts); re.Next() && replicasAdded < replicasToAdd; {
 			requeue, err := rq.processOneChange(ctx, r, sysCfg, func() bool { return false } /* canTransferLease */, false /* dryRun */, true /* disableStatsBasedRebalancing */, true /* forceAddReplica */)
 			if err != nil {
-				if IsSnapshotError(err) {
+				if IsSnapshotError(err) || IsThrottledStoresError(err) {
 					// TODO: remove
-					log.Infof(ctx, "encountered snapshot error when adding replica for scatter: %s", err)
+					log.Infof(ctx, "encountered retriable error when adding replica for scatter: %s", err)
 					continue
 				}
-				log.Warningf(ctx, "encountered error when adding replica for scatter: %s", err)
+				log.Warningf(ctx, "encountered non-retriable error when adding replica for scatter: %s", err)
 				break
 			}
 			replicasAdded++
@@ -2160,7 +2160,7 @@ func (r *Replica) adminScatter(
 				log.Infof(ctx, "encountered snapshot error from replicate queue during scatter: %s", err)
 				continue
 			}
-			log.Warningf(ctx, "encountered error from replicate queue during scatter: %s", err)
+			log.Warningf(ctx, "encountered non-retriable error from replicate queue during scatter: %s", err)
 			break
 		}
 		if !requeue {