Return non-200 HTTP response code if concurrent API requests cross the specified concurrent-request-limit # (#130)

* With Zero backlog and backlog getting timing out immediately with 1ms, all concurrent requests beyond cfg.middleware.concurrentRequestLimit result in immediate non-200 HTTP response

* backlogDuration Object does not require parsing string

* make backlogLimit, backlogDuration configurable for concurrent requests

* update flag type to time.Duration and docs

Author: spaparaju

README.md

@@ -83,6 +83,10 @@ Usage of ./observatorium:
     	The name of the HTTP header containing the tenant ID to forward to the metrics upstreams. (default "THANOS-TENANT")
   -metrics.write.endpoint string
     	The endpoint against which to make write requests for metrics.
+  -middleware.backlog-duration-concurrent-requests duration
+    	The time duration to buffer up concurrent requests. (default 1ms)
+  -middleware.backlog-limit-concurrent-requests int
+    	The number of concurrent requests that can buffered.
   -middleware.concurrent-request-limit int
     	The limit that controls the number of concurrently processed requests across all tenants. (default 10000)
   -middleware.rate-limiter.grpc-address string

main.go

@@ -116,8 +116,10 @@ type logsConfig struct {
 }
 
 type middlewareConfig struct {
-	rateLimiterAddress     string
-	concurrentRequestLimit int
+	rateLimiterAddress                string
+	concurrentRequestLimit            int
+	backLogLimitConcurrentRequests    int
+	backLogDurationConcurrentRequests time.Duration
 }
 
 type internalTracingConfig struct {
@@ -369,7 +371,9 @@ func main() {
 		r.Use(middleware.Recoverer)
 		r.Use(middleware.StripSlashes)
 		r.Use(middleware.Timeout(middlewareTimeout)) // best set per handler.
-		r.Use(middleware.Throttle(cfg.middleware.concurrentRequestLimit))
+		// With default value of zero backlog concurrent requests crossing a rate-limit result in non-200 HTTP response.
+		r.Use(middleware.ThrottleBacklog(cfg.middleware.concurrentRequestLimit,
+			cfg.middleware.backLogLimitConcurrentRequests, cfg.middleware.backLogDurationConcurrentRequests))
 		r.Use(server.Logger(logger))
 
 		ins := signalhttp.NewHandlerInstrumenter(reg, []string{"group", "handler"})
@@ -707,6 +711,11 @@ func parseFlags() (config, error) {
 			" If not specified, local, non-shared rate limiting will be used.")
 	flag.IntVar(&cfg.middleware.concurrentRequestLimit, "middleware.concurrent-request-limit", 10_000,
 		"The limit that controls the number of concurrently processed requests across all tenants.")
+	flag.IntVar(&cfg.middleware.backLogLimitConcurrentRequests, "middleware.backlog-limit-concurrent-requests", 0,
+		"The number of concurrent requests that can buffered.")
+	flag.DurationVar(&cfg.middleware.backLogDurationConcurrentRequests, "middleware.backlog-duration-concurrent-requests", 1*time.Millisecond,
+		"The time duration to buffer up concurrent requests.")
+
 	flag.Parse()
 
 	metricsReadEndpoint, err := url.ParseRequestURI(rawMetricsReadEndpoint)