diff --git a/imageproxy.go b/imageproxy.go index 929089078..3da57a252 100644 --- a/imageproxy.go +++ b/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 diff --git a/imageproxy_test.go b/imageproxy_test.go index 57670d1c7..150d7c429 100644 --- a/imageproxy_test.go +++ b/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 diff --git a/metrics.go b/metrics.go index 0466ac0c9..13512087c 100644 --- a/metrics.go +++ b/metrics.go @@ -29,6 +29,11 @@ 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() { @@ -36,4 +41,5 @@ func init() { prometheus.MustRegister(metricServedFromCache) prometheus.MustRegister(metricRemoteErrors) prometheus.MustRegister(metricRequestDuration) + prometheus.MustRegister(metricRequestsInFlight) }