feat: limit concurrent processing of thumbnail requests

ocis-pkg/middleware/throttle.go

@@ -0,0 +1,19 @@
+package middleware
+
+import (
+	"net/http"
+
+	"github.com/go-chi/chi/v5/middleware"
+)
+
+// Throttle limits the number of concurrent requests.
+func Throttle(limit int) func(http.Handler) http.Handler {
+	if limit > 0 {
+		return middleware.Throttle(limit)
+	}
+	return func(next http.Handler) http.Handler {
+		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			next.ServeHTTP(w, r)
+		})
+	}
+}

services/thumbnails/pkg/command/server.go

@@ -98,6 +98,7 @@ func Server(cfg *config.Config) *cli.Command {
 				http.Metrics(m),
 				http.Namespace(cfg.HTTP.Namespace),
 				http.TraceProvider(traceProvider),
+				http.MaxConcurrentRequests(cfg.HTTP.MaxConcurrentRequests),
 			)
 			if err != nil {
 				logger.Info().

services/thumbnails/pkg/config/defaults/defaultconfig.go

@@ -32,9 +32,10 @@ func DefaultConfig() *config.Config {
 			Namespace: "com.owncloud.api",
 		},
 		HTTP: config.HTTP{
-			Addr:      "127.0.0.1:9186",
-			Root:      "/thumbnails",
-			Namespace: "com.owncloud.web",
+			Addr:                  "127.0.0.1:9186",
+			Root:                  "/thumbnails",
+			Namespace:             "com.owncloud.web",
+			MaxConcurrentRequests: 0,
 		},
 		Service: config.Service{
 			Name: "thumbnails",

services/thumbnails/pkg/config/http.go

@@ -4,8 +4,9 @@ import "github.com/owncloud/ocis/v2/ocis-pkg/shared"
 
 // HTTP defines the available http configuration.
 type HTTP struct {
-	Addr      string                `yaml:"addr" env:"THUMBNAILS_HTTP_ADDR" desc:"The bind address of the HTTP service." introductionVersion:"pre5.0"`
-	TLS       shared.HTTPServiceTLS `yaml:"tls"`
-	Root      string                `yaml:"root" env:"THUMBNAILS_HTTP_ROOT" desc:"Subdirectory that serves as the root for this HTTP service." introductionVersion:"pre5.0"`
-	Namespace string                `yaml:"-"`
+	Addr                  string                `yaml:"addr" env:"THUMBNAILS_HTTP_ADDR" desc:"The bind address of the HTTP service." introductionVersion:"pre5.0"`
+	TLS                   shared.HTTPServiceTLS `yaml:"tls"`
+	Root                  string                `yaml:"root" env:"THUMBNAILS_HTTP_ROOT" desc:"Subdirectory that serves as the root for this HTTP service." introductionVersion:"pre5.0"`
+	Namespace             string                `yaml:"-"`
+	MaxConcurrentRequests int                   `yaml:"max_concurrent_requests" env:"THUMBNAILS_MAX_CONCURRENT_REQUESTS" desc:"Number of maximum concurrent thumbnail requests. Default is 0 which is unlimited." introductionVersion:"6.0"`
 }

services/thumbnails/pkg/server/http/option.go

@@ -16,13 +16,14 @@ type Option func(o *Options)
 
 // Options defines the available options for this package.
 type Options struct {
-	Namespace     string
-	Logger        log.Logger
-	Context       context.Context
-	Config        *config.Config
-	Metrics       *metrics.Metrics
-	Flags         []cli.Flag
-	TraceProvider trace.TracerProvider
+	Namespace             string
+	Logger                log.Logger
+	Context               context.Context
+	Config                *config.Config
+	Metrics               *metrics.Metrics
+	Flags                 []cli.Flag
+	TraceProvider         trace.TracerProvider
+	MaxConcurrentRequests int
 }
 
 // newOptions initializes the available default options.
@@ -81,3 +82,10 @@ func TraceProvider(traceProvider trace.TracerProvider) Option {
 		}
 	}
 }
+
+// MaxConcurrentRequests provides a function to set the MaxConcurrentRequests option.
+func MaxConcurrentRequests(val int) Option {
+	return func(o *Options) {
+		o.MaxConcurrentRequests = val
+	}
+}

services/thumbnails/pkg/server/http/server.go

@@ -39,6 +39,7 @@ func Server(opts ...Option) (http.Service, error) {
 		svc.Middleware(
 			middleware.RealIP,
 			middleware.RequestID,
+			ocismiddleware.Throttle(options.MaxConcurrentRequests),
 			ocismiddleware.Version(
 				options.Config.Service.Name,
 				version.GetString(),

services/thumbnails/pkg/service/http/v0/service.go

@@ -3,12 +3,11 @@ package svc
 import (
 	"context"
 	"fmt"
-	"net/http"
-	"strconv"
-
 	"github.com/go-chi/chi/v5"
 	"github.com/golang-jwt/jwt/v4"
 	"github.com/riandyrn/otelchi"
+	"net/http"
+	"strconv"
 
 	"github.com/owncloud/ocis/v2/ocis-pkg/log"
 	"github.com/owncloud/ocis/v2/ocis-pkg/tracing"

services/webdav/pkg/service/v0/search.go

@@ -47,7 +47,7 @@ func (g Webdav) Search(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	t := r.Header.Get(TokenHeader)
+	t := r.Header.Get(revactx.TokenHeader)
 	ctx := revactx.ContextSetToken(r.Context(), t)
 	ctx = metadata.Set(ctx, revactx.TokenHeader, t)
 

services/webdav/pkg/service/v0/service.go

@@ -13,6 +13,7 @@ import (
 	gatewayv1beta1 "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
 	userv1beta1 "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
 	rpcv1beta1 "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1"
+	revactx "github.com/cs3org/reva/v2/pkg/ctx"
 	"github.com/cs3org/reva/v2/pkg/rgrpc/todo/pool"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/templates"
 	"github.com/go-chi/chi/v5"
@@ -37,10 +38,6 @@ func init() {
 	chi.RegisterMethod("REPORT")
 }
 
-const (
-	TokenHeader = "X-Access-Token"
-)
-
 var (
 	codesEnum = map[int]string{
 		http.StatusBadRequest:       "Sabre\\DAV\\Exception\\BadRequest",
@@ -52,8 +49,8 @@ var (
 
 // Service defines the extension handlers.
 type Service interface {
-	ServeHTTP(http.ResponseWriter, *http.Request)
-	Thumbnail(http.ResponseWriter, *http.Request)
+	ServeHTTP(w http.ResponseWriter, r *http.Request)
+	Thumbnail(w http.ResponseWriter, r *http.Request)
 }
 
 // NewService returns a service implementation for Service.
@@ -235,7 +232,7 @@ func (g Webdav) SpacesThumbnail(w http.ResponseWriter, r *http.Request) {
 		renderError(w, r, errBadRequest(err.Error()))
 		return
 	}
-	t := r.Header.Get(TokenHeader)
+	t := r.Header.Get(revactx.TokenHeader)
 
 	fullPath := filepath.Join(tr.Identifier, tr.Filepath)
 	rsp, err := g.thumbnailsClient.GetThumbnail(r.Context(), &thumbnailssvc.GetThumbnailRequest{
@@ -284,7 +281,7 @@ func (g Webdav) Thumbnail(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	t := r.Header.Get(TokenHeader)
+	t := r.Header.Get(revactx.TokenHeader)
 
 	gatewayClient, err := g.gatewaySelector.Next()
 	if err != nil {
@@ -312,7 +309,7 @@ func (g Webdav) Thumbnail(w http.ResponseWriter, r *http.Request) {
 		user = userRes.GetUser()
 	} else {
 		// look up user from URL via GetUserByClaim
-		ctx := grpcmetadata.AppendToOutgoingContext(r.Context(), TokenHeader, t)
+		ctx := grpcmetadata.AppendToOutgoingContext(r.Context(), revactx.TokenHeader, t)
 		userRes, err := gatewayClient.GetUserByClaim(ctx, &userv1beta1.GetUserByClaimRequest{
 			Claim: "username",
 			Value: tr.Identifier,
@@ -475,11 +472,11 @@ func (g Webdav) sendThumbnailResponse(rsp *thumbnailssvc.GetThumbnailResponse, w
 
 	if dlRsp.StatusCode != http.StatusOK {
 		logger.Debug().
-			Str("transfer_token", rsp.TransferToken).
-			Str("data_endpoint", rsp.DataEndpoint).
+			Str("transfer_token", rsp.GetTransferToken()).
+			Str("data_endpoint", rsp.GetDataEndpoint()).
 			Str("response_status", dlRsp.Status).
 			Msg("could not download thumbnail")
-		renderError(w, r, errInternalError("could not download thumbnail"))
+		renderError(w, r, newErrResponse(dlRsp.StatusCode, "could not download thumbnail"))
 		return
 	}