Limit number of the transformations threads

imageproxy.go

@@ -31,6 +31,7 @@ import (
 	"net/http"
 	"net/url"
 	"path"
+	"runtime"
 	"strings"
 	"time"
 
@@ -113,6 +114,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 ...interface{}) {
 				if proxy.Verbose {
 					proxy.logf(format, v...)
@@ -151,7 +153,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)
 }
 
@@ -427,6 +434,8 @@ type TransformingTransport struct {
 	// responses are properly cached.
 	CachingClient *http.Client
 
+	limiter chan struct{}
+
 	log func(format string, v ...interface{})
 }
 
@@ -440,6 +449,13 @@ func (t *TransformingTransport) RoundTrip(req *http.Request) (*http.Response, er
 		return t.Transport.RoundTrip(req)
 	}
 
+	if t.limiter != nil {
+		t.limiter <- struct{}{}
+		defer func() {
+			<-t.limiter
+		}()
+	}
+
 	f := req.URL.Fragment
 	req.URL.Fragment = ""
 	resp, err := t.CachingClient.Do(req)

imageproxy_test.go

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

metrics.go

@@ -26,11 +26,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)
 }