Merge pull request #36777 from lucy-zhang/backport19.1-36735

release-19.1: storage: limit number of AddSSTable requests per second

docs/generated/settings/settings.html

@@ -28,6 +28,7 @@
 <tr><td><code>kv.allocator.load_based_rebalancing</code></td><td>enumeration</td><td><code>2</code></td><td>whether to rebalance based on the distribution of QPS across stores [off = 0, leases = 1, leases and replicas = 2]</td></tr>
 <tr><td><code>kv.allocator.qps_rebalance_threshold</code></td><td>float</td><td><code>0.25</code></td><td>minimum fraction away from the mean a store's QPS (such as queries per second) can be before it is considered overfull or underfull</td></tr>
 <tr><td><code>kv.allocator.range_rebalance_threshold</code></td><td>float</td><td><code>0.05</code></td><td>minimum fraction away from the mean a store's range count can be before it is considered overfull or underfull</td></tr>
+<tr><td><code>kv.bulk_io_write.addsstable_max_rate</code></td><td>float</td><td><code>1.7976931348623157E+308</code></td><td>maximum number of AddSSTable requests per second for a single store</td></tr>
 <tr><td><code>kv.bulk_io_write.concurrent_addsstable_requests</code></td><td>integer</td><td><code>1</code></td><td>number of AddSSTable requests a store will handle concurrently before queuing</td></tr>
 <tr><td><code>kv.bulk_io_write.concurrent_export_requests</code></td><td>integer</td><td><code>3</code></td><td>number of export requests a store will handle concurrently before queuing</td></tr>
 <tr><td><code>kv.bulk_io_write.concurrent_import_requests</code></td><td>integer</td><td><code>1</code></td><td>number of import requests a store will handle concurrently before queuing</td></tr>

pkg/storage/batcheval/eval_context.go

@@ -37,6 +37,7 @@ type Limiters struct {
 	BulkIOWriteRate              *rate.Limiter
 	ConcurrentImportRequests     limit.ConcurrentRequestLimiter
 	ConcurrentExportRequests     limit.ConcurrentRequestLimiter
+	AddSSTableRequestRate        *rate.Limiter
 	ConcurrentAddSSTableRequests limit.ConcurrentRequestLimiter
 	// concurrentRangefeedIters is a semaphore used to limit the number of
 	// rangefeeds in the "catch-up" state across the store. The "catch-up" state

pkg/storage/store.go

@@ -123,6 +123,15 @@ var importRequestsLimit = settings.RegisterPositiveIntSetting(
 	1,
 )
 
+// addSSTableRequestMaxRate is the maximum number of AddSSTable requests per second.
+var addSSTableRequestMaxRate = settings.RegisterNonNegativeFloatSetting(
+	"kv.bulk_io_write.addsstable_max_rate",
+	"maximum number of AddSSTable requests per second for a single store",
+	float64(rate.Inf),
+)
+
+const addSSTableRequestBurst = 32
+
 // addSSTableRequestLimit limits concurrent AddSSTable requests.
 var addSSTableRequestLimit = settings.RegisterPositiveIntSetting(
 	"kv.bulk_io_write.concurrent_addsstable_requests",
@@ -860,6 +869,16 @@ func NewStore(cfg StoreConfig, eng engine.Engine, nodeDesc *roachpb.NodeDescript
 		}
 		s.limiters.ConcurrentExportRequests.SetLimit(limit)
 	})
+	s.limiters.AddSSTableRequestRate = rate.NewLimiter(
+		rate.Limit(addSSTableRequestMaxRate.Get(&cfg.Settings.SV)), addSSTableRequestBurst)
+	addSSTableRequestMaxRate.SetOnChange(&cfg.Settings.SV, func() {
+		rateLimit := addSSTableRequestMaxRate.Get(&cfg.Settings.SV)
+		if math.IsInf(rateLimit, 0) {
+			// This value causes the burst limit to be ignored
+			rateLimit = float64(rate.Inf)
+		}
+		s.limiters.AddSSTableRequestRate.SetLimit(rate.Limit(rateLimit))
+	})
 	s.limiters.ConcurrentAddSSTableRequests = limit.MakeConcurrentRequestLimiter(
 		"addSSTableRequestLimiter", int(addSSTableRequestLimit.Get(&cfg.Settings.SV)),
 	)
@@ -2759,6 +2778,10 @@ func (s *Store) Send(
 			return nil, roachpb.NewError(err)
 		}
 		defer s.limiters.ConcurrentAddSSTableRequests.Finish()
+
+		if err := s.limiters.AddSSTableRequestRate.Wait(ctx); err != nil {
+			return nil, roachpb.NewError(err)
+		}
 		s.engine.PreIngestDelay(ctx)
 	}