Limit number of the transformations threads

imageproxy.go

@@ -20,6 +20,7 @@ import (
 	"net/http"
 	"net/url"
 	"path"
+	"runtime"
 	"strings"
 	"time"
 
@@ -125,6 +126,7 @@ func NewProxy(transport http.RoundTripper, cache Cache) *Proxy {
 		Transport: &TransformingTransport{
 			Transport:     transport,
 			CachingClient: client,
+			limiter:       make(chan struct{}, runtime.NumCPU()),
 			log: func(format string, v ...any) {
 				if proxy.Verbose {
 					proxy.logf(format, v...)
@@ -221,7 +223,12 @@ func (p *Proxy) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 
 	timer := prometheus.NewTimer(metricRequestDuration)
-	defer timer.ObserveDuration()
+	metricRequestsInFlight.Inc()
+	defer func() {
+		timer.ObserveDuration()
+		metricRequestsInFlight.Dec()
+	}()
+
 	h.ServeHTTP(w, r)
 }
 
@@ -557,6 +564,9 @@ type TransformingTransport struct {
 	// responses are properly cached.
 	CachingClient *http.Client
 
+	// limiter limits the number of concurrent transformations being processed.
+	limiter chan struct{}
+
 	log func(format string, v ...any)
 
 	updateCacheHeaders func(hdr http.Header)
@@ -598,6 +608,15 @@ func (t *TransformingTransport) RoundTrip(req *http.Request) (*http.Response, er
 		}, nil
 	}
 
+	// enforce limiter after we've checked if we can early return a 304 response,
+	// but before we read the response body and perform transformations.
+	if t.limiter != nil {
+		t.limiter <- struct{}{}
+		defer func() {
+			<-t.limiter
+		}()
+	}
+
 	b, err := io.ReadAll(resp.Body)
 	if err != nil {
 		return nil, err

imageproxy_test.go

@@ -735,6 +735,7 @@ func TestTransformingTransport(t *testing.T) {
 	tr := &TransformingTransport{
 		Transport:     &testTransport{},
 		CachingClient: client,
+		limiter:       make(chan struct{}, 1),
 	}
 	client.Transport = tr
 

metrics.go

@@ -29,11 +29,17 @@ var (
 		Name:      "request_duration_seconds",
 		Help:      "Request response times",
 	})
+	metricRequestsInFlight = prometheus.NewGauge(prometheus.GaugeOpts{
+		Namespace: "http",
+		Name:      "requests_in_flight",
+		Help:      "Number of requests in flight",
+	})
 )
 
 func init() {
 	prometheus.MustRegister(metricTransformationDuration)
 	prometheus.MustRegister(metricServedFromCache)
 	prometheus.MustRegister(metricRemoteErrors)
 	prometheus.MustRegister(metricRequestDuration)
+	prometheus.MustRegister(metricRequestsInFlight)
 }