OCPBUGS-18971: limit number of simultaneous client requests

pkg/client/api.go

@@ -128,11 +128,13 @@ func (c *httpAPIClient) Do(ctx context.Context, verb, endpoint string, query url
 
 // NewGenericAPIClient builds a new generic Prometheus API client for the given base URL and HTTP Client.
 func NewGenericAPIClient(client *http.Client, baseURL *url.URL, headers http.Header) GenericAPIClient {
-	return &httpAPIClient{
+	c := &httpAPIClient{
 		client:  client,
 		baseURL: baseURL,
 		headers: headers,
 	}
+
+	return newRequestLimiter(c)
 }
 
 const (

pkg/client/limiter.go

@@ -0,0 +1,86 @@
+/*
+Copyright 2017 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package client
+
+import (
+	"context"
+	"net/url"
+
+	"k8s.io/component-base/metrics"
+	"k8s.io/component-base/metrics/legacyregistry"
+)
+
+// Without a limit, the adapter could flood the Prometheus API with many
+// requests when there are many pods running in the cluster because a query for
+// getting all pod metrics would translate into (2 x pod number) queries to the
+// Prometheus API.
+// In the worst case, the Prometheus pods can hit their SOMAXCONN limit leading
+// to timed-out requests from other clients. In particular it has been reported
+// that it could fail the Kubelet liveness probes and lead to Prometheus pod
+// restarts.
+// The number has been chosen from empirical data.
+const maxConcurrentRequests = 100
+
+var (
+	inflightRequests = metrics.NewGauge(
+		&metrics.GaugeOpts{
+			Namespace: "prometheus_adapter",
+			Subsystem: "prometheus_client",
+			Name:      "inflight_requests",
+			Help:      "Number of inflight requests to the Prometheus service",
+		})
+
+	maxRequests = metrics.NewGauge(
+		&metrics.GaugeOpts{
+			Namespace: "prometheus_adapter",
+			Subsystem: "prometheus_client",
+			Name:      "max_requests",
+			Help:      "Maximum number of requests to the Prometheus service",
+		})
+)
+
+func init() {
+	legacyregistry.MustRegister(inflightRequests, maxRequests)
+	maxRequests.Set(maxConcurrentRequests)
+}
+
+type requestLimitClient struct {
+	c        GenericAPIClient
+	inflight chan struct{}
+}
+
+func newRequestLimiter(c GenericAPIClient) GenericAPIClient {
+	return &requestLimitClient{
+		c:        c,
+		inflight: make(chan struct{}, maxConcurrentRequests),
+	}
+}
+
+func (c *requestLimitClient) Do(ctx context.Context, verb, endpoint string, query url.Values) (APIResponse, error) {
+	select {
+	case c.inflight <- struct{}{}:
+		inflightRequests.Inc()
+		defer func() {
+			inflightRequests.Dec()
+			<-c.inflight
+		}()
+	case <-ctx.Done():
+		return APIResponse{}, ctx.Err()
+	}
+
+	return c.c.Do(ctx, verb, endpoint, query)
+}

pkg/client/limiter_test.go

@@ -0,0 +1,97 @@
+/*
+Copyright 2017 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package client
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestRequestLimitClient(t *testing.T) {
+	var (
+		ctx     = context.Background()
+		total   atomic.Int64
+		unblock = make(chan struct{})
+	)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		total.Add(1)
+
+		w.Write([]byte("{}"))
+		if r.URL.Path == "/nonblocking" {
+			return
+		}
+
+		// Requests will be blocked until the test closes the unblock channel.
+		<-unblock
+	}))
+	defer srv.Close()
+
+	// Make as many requests as the max allowed number + 1.
+	var (
+		wg      sync.WaitGroup
+		errChan = make(chan error, maxConcurrentRequests+1)
+		u, _    = url.Parse(srv.URL)
+		c       = NewGenericAPIClient(&http.Client{}, u, nil)
+	)
+	for i := 0; i < maxConcurrentRequests+1; i++ {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+
+			_, err := c.Do(ctx, "GET", "/", nil)
+			if err != nil {
+				err = fmt.Errorf("request #%d: %w", i, err)
+			}
+			errChan <- err
+		}(i)
+	}
+
+	// Wait for the unblocked requests to hit the server.
+	for total.Load() != maxConcurrentRequests {
+	}
+
+	// Make one more blocked request which should timeout before hitting the server.
+	ctx2, _ := context.WithTimeout(ctx, time.Second)
+	_, err := c.Do(ctx2, "GET", "/nonblocking", nil)
+	switch {
+	case err == nil:
+		t.Fatalf("expected %dth request to fail", maxConcurrentRequests+2)
+	case ctx2.Err() == nil:
+		t.Fatalf("expected %dth request to timeout", maxConcurrentRequests+2)
+	}
+
+	// Release all inflight requests.
+	close(unblock)
+
+	// Wait for all requests to complete.
+	wg.Wait()
+
+	// Check that no error was returned.
+	close(errChan)
+	for err := range errChan {
+		if err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	}
+}