diff --git a/cmd/http-gw/app.go b/cmd/http-gw/app.go index 84379d4..261f65a 100644 --- a/cmd/http-gw/app.go +++ b/cmd/http-gw/app.go @@ -40,6 +40,7 @@ import ( "github.com/nspcc-dev/neo-go/pkg/crypto/keys" "github.com/nspcc-dev/neo-go/pkg/util" "github.com/nspcc-dev/neo-go/pkg/wallet" + "github.com/panjf2000/ants/v2" "github.com/spf13/viper" "github.com/valyala/fasthttp" "go.opentelemetry.io/otel/trace" @@ -96,6 +97,7 @@ type ( clientCut bool returnIndexPage bool indexPageTemplate string + workerPoolSize int bufferMaxSizeForPut uint64 namespaceHeader string defaultNamespaces []string @@ -194,6 +196,7 @@ func (s *appSettings) update(v *viper.Viper, l *zap.Logger) { returnIndexPage := v.GetBool(cfgIndexPageEnabled) clientCut := v.GetBool(cfgClientCut) bufferMaxSizeForPut := v.GetUint64(cfgBufferMaxSizeForPut) + workerPoolSize := v.GetInt(cfgWorkerPoolSize) namespaceHeader := v.GetString(cfgResolveNamespaceHeader) defaultNamespaces := fetchDefaultNamespaces(v) indexPage, indexEnabled := fetchIndexPageTemplate(v, l) @@ -216,6 +219,7 @@ func (s *appSettings) update(v *viper.Viper, l *zap.Logger) { s.defaultNamespaces = defaultNamespaces s.returnIndexPage = indexEnabled s.indexPageTemplate = indexPage + s.workerPoolSize = workerPoolSize s.corsAllowOrigin = corsAllowOrigin s.corsAllowMethods = corsAllowMethods s.corsAllowHeaders = corsAllowHeaders @@ -242,6 +246,12 @@ func (s *appSettings) IndexPageEnabled() bool { return s.returnIndexPage } +func (s *appSettings) WorkerPoolSize() int { + s.mu.RLock() + defer s.mu.RUnlock() + return s.workerPoolSize +} + func (s *appSettings) IndexPageTemplate() string { s.mu.RLock() defer s.mu.RUnlock() @@ -490,7 +500,13 @@ func (a *app) setHealthStatus() { } func (a *app) Serve() { - handler := handler.New(a.AppParams(), a.settings, tree.NewTree(frostfs.NewPoolWrapper(a.treePool))) + workerPool := a.initWorkerPool() + defer func() { + workerPool.Release() + close(a.webDone) + }() + + handler := handler.New(a.AppParams(), a.settings, tree.NewTree(frostfs.NewPoolWrapper(a.treePool)), workerPool) // Configure router. a.configureRouter(handler) @@ -532,8 +548,14 @@ LOOP: a.metrics.Shutdown() a.stopServices() a.shutdownTracing() +} - close(a.webDone) +func (a *app) initWorkerPool() *ants.Pool { + workerPool, err := ants.NewPool(a.settings.WorkerPoolSize()) + if err != nil { + a.log.Fatal(logs.FailedToCreateWorkerPool, zap.Error(err)) + } + return workerPool } func (a *app) shutdownTracing() { @@ -609,6 +631,7 @@ func (a *app) stopServices() { svc.ShutDown(ctx) } } + func (a *app) configureRouter(handler *handler.Handler) { r := router.New() r.RedirectTrailingSlash = true diff --git a/cmd/http-gw/settings.go b/cmd/http-gw/settings.go index f464fbc..bc36077 100644 --- a/cmd/http-gw/settings.go +++ b/cmd/http-gw/settings.go @@ -70,6 +70,7 @@ const ( cfgIndexPageEnabled = "index_page.enabled" cfgIndexPageTemplatePath = "index_page.template_path" + cfgWorkerPoolSize = "worker_pool_size" // Web. cfgWebReadBufferSize = "web.read_buffer_size" @@ -227,9 +228,6 @@ func settings() *viper.Viper { // pool: v.SetDefault(cfgPoolErrorThreshold, defaultPoolErrorThreshold) - v.SetDefault(cfgIndexPageEnabled, false) - v.SetDefault(cfgIndexPageTemplatePath, "") - // frostfs: v.SetDefault(cfgBufferMaxSizeForPut, defaultBufferMaxSizeForPut) @@ -241,6 +239,7 @@ func settings() *viper.Viper { v.SetDefault(cfgWebStreamRequestBody, true) v.SetDefault(cfgWebMaxRequestBodySize, fasthttp.DefaultMaxRequestBodySize) + v.SetDefault(cfgWorkerPoolSize, 1000) // upload header v.SetDefault(cfgUploaderHeaderEnableDefaultTimestamp, false) diff --git a/docs/gate-configuration.md b/docs/gate-configuration.md index be4b30b..fce5e2c 100644 --- a/docs/gate-configuration.md +++ b/docs/gate-configuration.md @@ -75,18 +75,21 @@ request_timeout: 5s rebalance_timer: 30s pool_error_threshold: 100 reconnect_interval: 1m +worker_pool_size: 1000 + ``` -| Parameter | Type | SIGHUP reload | Default value | Description | -|------------------------|------------|---------------|---------------|-------------------------------------------------------------------------------------------------| -| `rpc_endpoint` | `string` | yes | | The address of the RPC host to which the gateway connects to resolve bucket names. | -| `resolve_order` | `[]string` | yes | `[nns, dns]` | Order of bucket name resolvers to use. | -| `connect_timeout` | `duration` | | `10s` | Timeout to connect to a node. | -| `stream_timeout` | `duration` | | `10s` | Timeout for individual operations in streaming RPC. | -| `request_timeout` | `duration` | | `15s` | Timeout to check node health during rebalance. | -| `rebalance_timer` | `duration` | | `60s` | Interval to check node health. | -| `pool_error_threshold` | `uint32` | | `100` | The number of errors on connection after which node is considered as unhealthy. | -| `reconnect_interval` | `duration` | no | `1m` | Listeners reconnection interval. | +| Parameter | Type | SIGHUP reload | Default value | Description | +|------------------------|------------|---------------|---------------|------------------------------------------------------------------------------------| +| `rpc_endpoint` | `string` | yes | | The address of the RPC host to which the gateway connects to resolve bucket names. | +| `resolve_order` | `[]string` | yes | `[nns, dns]` | Order of bucket name resolvers to use. | +| `connect_timeout` | `duration` | | `10s` | Timeout to connect to a node. | +| `stream_timeout` | `duration` | | `10s` | Timeout for individual operations in streaming RPC. | +| `request_timeout` | `duration` | | `15s` | Timeout to check node health during rebalance. | +| `rebalance_timer` | `duration` | | `60s` | Interval to check node health. | +| `pool_error_threshold` | `uint32` | | `100` | The number of errors on connection after which node is considered as unhealthy. | +| `reconnect_interval` | `duration` | no | `1m` | Listeners reconnection interval. | +| `worker_pool_size` | `int` | | | | # `wallet` section @@ -352,7 +355,12 @@ resolve_bucket: # `index_page` section -Parameters for index HTML-page output with S3-bucket or S3-subdir content for `Get object` request +Parameters for index HTML-page output. Activates if `GetObject` request returns `not found`. Two +index page modes available: + +* `s3` mode uses tree service for listing objects, +* `native` sends requests to nodes via native protocol. + If request pass S3-bucket name instead of CID, `s3` mode will be used, otherwise `native`. ```yaml index_page: diff --git a/go.mod b/go.mod index efb79eb..add46be 100644 --- a/go.mod +++ b/go.mod @@ -12,6 +12,7 @@ require ( github.com/docker/go-units v0.4.0 github.com/fasthttp/router v1.4.1 github.com/nspcc-dev/neo-go v0.106.2 + github.com/panjf2000/ants/v2 v2.5.0 github.com/prometheus/client_golang v1.19.0 github.com/prometheus/client_model v0.5.0 github.com/spf13/pflag v1.0.5 diff --git a/go.sum b/go.sum index ce02ee6..922e4a1 100644 --- a/go.sum +++ b/go.sum @@ -682,6 +682,8 @@ github.com/opencontainers/selinux v1.6.0/go.mod h1:VVGKuOLlE7v4PJyT6h7mNWvq1rzqi github.com/opencontainers/selinux v1.8.0/go.mod h1:RScLhm78qiWa2gbVCcGkC7tCGdgk3ogry1nUQF8Evvo= github.com/opencontainers/selinux v1.8.2/go.mod h1:MUIHuUEvKB1wtJjQdOyYRgOnLD2xAPP8dBsCoU0KuF8= github.com/opencontainers/selinux v1.10.0/go.mod h1:2i0OySw99QjzBBQByd1Gr9gSjvuho1lHsJxIJ3gGbJI= +github.com/panjf2000/ants/v2 v2.5.0 h1:1rWGWSnxCsQBga+nQbA4/iY6VMeNoOIAM0ZWh9u3q2Q= +github.com/panjf2000/ants/v2 v2.5.0/go.mod h1:cU93usDlihJZ5CfRGNDYsiBYvoilLvBF5Qp/BT2GNRE= github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic= github.com/pelletier/go-toml v1.8.1/go.mod h1:T2/BmBdy8dvIRq1a/8aqjN41wvWlN4lrapLU/GW4pbc= github.com/pelletier/go-toml/v2 v2.0.6 h1:nrzqCb7j9cDFj2coyLNLaZuJTLjWjlaz6nvTvIwycIU= diff --git a/internal/handler/browse.go b/internal/handler/browse.go index c89d8c5..4971f59 100644 --- a/internal/handler/browse.go +++ b/internal/handler/browse.go @@ -1,15 +1,21 @@ package handler import ( + "context" "html/template" "net/url" "sort" "strconv" "strings" + "sync" "time" "git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/data" + "git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/logs" "git.frostfs.info/TrueCloudLab/frostfs-http-gw/utils" + cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id" + "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object" + oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id" "github.com/docker/go-units" "github.com/valyala/fasthttp" "go.uber.org/zap" @@ -21,23 +27,73 @@ const ( attrCreated = "Created" attrFileName = "FileName" attrSize = "Size" + + defaultPoolSize = 1000 ) type ( BrowsePageData struct { - BucketName, - Prefix string - Objects []ResponseObject + Container string + Prefix string + Protocol string + Objects []ResponseObject } ResponseObject struct { OID string Created string FileName string + FilePath string Size string IsDir bool + GetURL string } ) +func newListObjectsResponseS3(attrs map[string]string) ResponseObject { + return ResponseObject{ + Created: formatTimestamp(attrs[attrCreated]), + OID: attrs[attrOID], + FileName: attrs[attrFileName], + Size: attrs[attrSize], + IsDir: attrs[attrOID] == "", + } +} + +func newListObjectsResponseNative(attrs map[string]string) ResponseObject { + filename := lastPathElement(attrs[object.AttributeFilePath]) + if filename == "" { + filename = attrs[attrFileName] + } + return ResponseObject{ + OID: attrs[attrOID], + Created: formatTimestamp(attrs[object.AttributeTimestamp] + "000"), + FileName: filename, + FilePath: attrs[object.AttributeFilePath], + Size: attrs[attrSize], + IsDir: false, + } +} + +func getNextDir(filepath, prefix string) string { + restPath := strings.Replace(filepath, prefix, "", 1) + index := strings.Index(restPath, "/") + if index == -1 { + return "" + } + return restPath[:index] +} + +func lastPathElement(path string) string { + if path == "" { + return path + } + index := strings.LastIndex(path, "/") + if index == len(path)-1 { + index = strings.LastIndex(path[:index], "/") + } + return path[index+1:] +} + func parseTimestamp(tstamp string) (time.Time, error) { millis, err := strconv.ParseInt(tstamp, 10, 64) if err != nil { @@ -47,16 +103,6 @@ func parseTimestamp(tstamp string) (time.Time, error) { return time.UnixMilli(millis), nil } -func NewResponseObject(nodes map[string]string) ResponseObject { - return ResponseObject{ - OID: nodes[attrOID], - Created: nodes[attrCreated], - FileName: nodes[attrFileName], - Size: nodes[attrSize], - IsDir: nodes[attrOID] == "", - } -} - func formatTimestamp(strdate string) string { date, err := parseTimestamp(strdate) if err != nil || date.IsZero() { @@ -94,12 +140,9 @@ func trimPrefix(encPrefix string) string { return prefix[:slashIndex] } -func urlencode(prefix, filename string) string { +func urlencode(path string) string { var res strings.Builder - path := filename - if prefix != "" { - path = strings.Join([]string{prefix, filename}, "/") - } + prefixParts := strings.Split(path, "/") for _, prefixPart := range prefixParts { prefixPart = "/" + url.PathEscape(prefixPart) @@ -112,46 +155,207 @@ func urlencode(prefix, filename string) string { return res.String() } -func (h *Handler) browseObjects(c *fasthttp.RequestCtx, bucketInfo *data.BucketInfo, prefix string) { +func (h *Handler) getDirObjectsS3(ctx context.Context, bucketInfo *data.BucketInfo, prefix string) ([]ResponseObject, error) { + nodes, _, err := h.tree.GetSubTreeByPrefix(ctx, bucketInfo, prefix, true) + if err != nil { + return nil, err + } + + var objects = make([]ResponseObject, 0, len(nodes)) + for _, node := range nodes { + meta := node.GetMeta() + if meta == nil { + continue + } + var attrs = make(map[string]string, len(meta)) + for _, m := range meta { + attrs[m.GetKey()] = string(m.GetValue()) + } + obj := newListObjectsResponseS3(attrs) + obj.FilePath = prefix + obj.FileName + obj.GetURL = "/get/" + bucketInfo.Name + urlencode(obj.FilePath) + objects = append(objects, obj) + } + + return objects, nil +} + +func (h *Handler) getDirObjectsNative(ctx context.Context, bucketInfo *data.BucketInfo, prefix string) ([]ResponseObject, error) { + var basePath string + if ind := strings.LastIndex(prefix, "/"); ind != -1 { + basePath = prefix[:ind+1] + } + + filters := object.NewSearchFilters() + filters.AddRootFilter() + if prefix != "" { + filters.AddFilter(object.AttributeFilePath, prefix, object.MatchCommonPrefix) + } + + prm := PrmObjectSearch{ + PrmAuth: PrmAuth{ + BearerToken: bearerToken(ctx), + }, + Container: bucketInfo.CID, + Filters: filters, + } + objectIDs, err := h.frostfs.SearchObjects(ctx, prm) + if err != nil { + return nil, err + } + defer objectIDs.Close() + + ctx, cancel := context.WithCancel(ctx) + defer cancel() + resp, err := h.headDirObjects(ctx, bucketInfo.CID, objectIDs, basePath) + if err != nil { + return nil, err + } + + log := utils.GetReqLogOrDefault(ctx, h.log) + dirs := make(map[string]struct{}) + objects := make([]ResponseObject, 0, 100) + for objExt := range resp { + if objExt.Error != nil { + log.Error(logs.FailedToHeadObject, zap.Error(objExt.Error)) + cancel() + continue + } + if objExt.Object.IsDir { + if _, ok := dirs[objExt.Object.FileName]; ok { + continue + } + objExt.Object.GetURL = "/get/" + bucketInfo.CID.EncodeToString() + urlencode(objExt.Object.FilePath) + dirs[objExt.Object.FileName] = struct{}{} + } else { + objExt.Object.GetURL = "/get/" + bucketInfo.CID.EncodeToString() + "/" + objExt.Object.OID + } + objects = append(objects, objExt.Object) + } + return objects, nil +} + +type ResponseObjectExtended struct { + Object ResponseObject + Error error +} + +func (h *Handler) headDirObjects(ctx context.Context, cnrID cid.ID, objectIDs ResObjectSearch, basePath string) (chan ResponseObjectExtended, error) { + res := make(chan ResponseObjectExtended) + + go func() { + defer close(res) + log := utils.GetReqLogOrDefault(ctx, h.log).With( + zap.String("cid", cnrID.EncodeToString()), + zap.String("path", basePath), + ) + var wg sync.WaitGroup + err := objectIDs.Iterate(func(id oid.ID) bool { + wg.Add(1) + err := h.workerPool.Submit(func() { + defer wg.Done() + var obj ResponseObjectExtended + obj.Object, obj.Error = h.headDirObject(ctx, cnrID, id, basePath) + res <- obj + }) + if err != nil { + wg.Done() + log.Warn(logs.FailedToSumbitTaskToPool, zap.Error(err)) + } + select { + case <-ctx.Done(): + return true + default: + return false + } + }) + if err != nil { + log.Error(logs.FailedToIterateOverResponse, zap.Error(err)) + } + wg.Wait() + }() + + return res, nil +} + +func (h *Handler) headDirObject(ctx context.Context, cnrID cid.ID, objID oid.ID, basePath string) (ResponseObject, error) { + addr := newAddress(cnrID, objID) + obj, err := h.frostfs.HeadObject(ctx, PrmObjectHead{ + PrmAuth: PrmAuth{BearerToken: bearerToken(ctx)}, + Address: addr, + }) + if err != nil { + return ResponseObject{}, err + } + + attrs := loadAttributes(obj.Attributes()) + attrs[attrOID] = objID.EncodeToString() + attrs[attrSize] = strconv.FormatUint(obj.PayloadSize(), 10) + + dirname := getNextDir(attrs[object.AttributeFilePath], basePath) + if dirname == "" { + return newListObjectsResponseNative(attrs), nil + } + + return ResponseObject{ + FileName: dirname, + FilePath: basePath + dirname, + IsDir: true, + }, nil +} + +type browseParams struct { + bucketInfo *data.BucketInfo + prefix string + isNative bool + listObjects func(ctx context.Context, bucketName *data.BucketInfo, prefix string) ([]ResponseObject, error) +} + +func (h *Handler) browseObjects(c *fasthttp.RequestCtx, p browseParams) { + const S3Protocol = "s3" + const FrostfsProtocol = "frostfs" + ctx := utils.GetContextFromRequest(c) reqLog := utils.GetReqLogOrDefault(ctx, h.log) - log := reqLog.With(zap.String("bucket", bucketInfo.Name)) - - nodes, err := h.listObjects(ctx, bucketInfo, prefix) + log := reqLog.With( + zap.String("bucket", p.bucketInfo.Name), + zap.String("container", p.bucketInfo.CID.EncodeToString()), + zap.String("prefix", p.prefix), + ) + objects, err := p.listObjects(ctx, p.bucketInfo, p.prefix) if err != nil { logAndSendBucketError(c, log, err) return } - respObjects := make([]ResponseObject, len(nodes)) - - for i, node := range nodes { - respObjects[i] = NewResponseObject(node) - } - - sort.Slice(respObjects, func(i, j int) bool { - if respObjects[i].IsDir == respObjects[j].IsDir { - return respObjects[i].FileName < respObjects[j].FileName + sort.Slice(objects, func(i, j int) bool { + if objects[i].IsDir == objects[j].IsDir { + return objects[i].FileName < objects[j].FileName } - return respObjects[i].IsDir + return objects[i].IsDir }) - indexTemplate := h.config.IndexPageTemplate() tmpl, err := template.New("index").Funcs(template.FuncMap{ - "formatTimestamp": formatTimestamp, - "formatSize": formatSize, - "trimPrefix": trimPrefix, - "urlencode": urlencode, - "parentDir": parentDir, - }).Parse(indexTemplate) + "formatSize": formatSize, + "trimPrefix": trimPrefix, + "urlencode": urlencode, + "parentDir": parentDir, + }).Parse(h.config.IndexPageTemplate()) if err != nil { logAndSendBucketError(c, log, err) return } + bucketName := p.bucketInfo.Name + protocol := S3Protocol + if p.isNative { + bucketName = p.bucketInfo.CID.EncodeToString() + protocol = FrostfsProtocol + } if err = tmpl.Execute(c, &BrowsePageData{ - BucketName: bucketInfo.Name, - Prefix: prefix, - Objects: respObjects, + Container: bucketName, + Prefix: p.prefix, + Objects: objects, + Protocol: protocol, }); err != nil { logAndSendBucketError(c, log, err) return diff --git a/internal/handler/download.go b/internal/handler/download.go index 19380d4..92e3083 100644 --- a/internal/handler/download.go +++ b/internal/handler/download.go @@ -23,10 +23,9 @@ import ( // DownloadByAddressOrBucketName handles download requests using simple cid/oid or bucketname/key format. func (h *Handler) DownloadByAddressOrBucketName(c *fasthttp.RequestCtx) { - test, _ := c.UserValue("oid").(string) - var id oid.ID - err := id.DecodeString(test) - if err != nil { + cnrIDStr, _ := c.UserValue("cid").(string) + var cnrID cid.ID + if err := cnrID.DecodeString(cnrIDStr); err != nil { h.byObjectName(c, h.receiveFile) } else { h.byAddress(c, h.receiveFile) @@ -45,7 +44,7 @@ func (h *Handler) DownloadByAttribute(c *fasthttp.RequestCtx) { h.byAttribute(c, h.receiveFile) } -func (h *Handler) search(ctx context.Context, cnrID *cid.ID, key, val string, op object.SearchMatchType) (ResObjectSearch, error) { +func (h *Handler) search(ctx context.Context, cnrID cid.ID, key, val string, op object.SearchMatchType) (ResObjectSearch, error) { filters := object.NewSearchFilters() filters.AddRootFilter() filters.AddFilter(key, val, op) @@ -54,7 +53,7 @@ func (h *Handler) search(ctx context.Context, cnrID *cid.ID, key, val string, op PrmAuth: PrmAuth{ BearerToken: bearerToken(ctx), }, - Container: *cnrID, + Container: cnrID, Filters: filters, } @@ -102,7 +101,7 @@ func (h *Handler) DownloadZipped(c *fasthttp.RequestCtx) { return } - resSearch, err := h.search(ctx, &bktInfo.CID, object.AttributeFilePath, prefix, object.MatchCommonPrefix) + resSearch, err := h.search(ctx, bktInfo.CID, object.AttributeFilePath, prefix, object.MatchCommonPrefix) if err != nil { log.Error(logs.CouldNotSearchForObjects, zap.Error(err)) response.Error(c, "could not search for objects: "+err.Error(), fasthttp.StatusBadRequest) diff --git a/internal/handler/handler.go b/internal/handler/handler.go index 62d0897..847b8cd 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -22,6 +22,7 @@ import ( "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object" oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id" "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/user" + "github.com/panjf2000/ants/v2" "github.com/valyala/fasthttp" "go.uber.org/zap" ) @@ -31,6 +32,7 @@ type Config interface { ZipCompression() bool ClientCut() bool IndexPageEnabled() bool + WorkerPoolSize() int IndexPageTemplate() string BufferMaxSizeForPut() uint64 NamespaceHeader() string @@ -165,6 +167,7 @@ type Handler struct { containerResolver ContainerResolver tree *tree.Tree cache *cache.BucketCache + workerPool *ants.Pool } type AppParams struct { @@ -175,7 +178,7 @@ type AppParams struct { Cache *cache.BucketCache } -func New(params *AppParams, config Config, tree *tree.Tree) *Handler { +func New(params *AppParams, config Config, tree *tree.Tree, workerPool *ants.Pool) *Handler { return &Handler{ log: params.Logger, frostfs: params.FrostFS, @@ -184,6 +187,7 @@ func New(params *AppParams, config Config, tree *tree.Tree) *Handler { containerResolver: params.Resolver, tree: tree, cache: params.Cache, + workerPool: workerPool, } } @@ -191,7 +195,7 @@ func New(params *AppParams, config Config, tree *tree.Tree) *Handler { // prepares request and object address to it. func (h *Handler) byAddress(c *fasthttp.RequestCtx, f func(context.Context, request, oid.Address)) { idCnr, _ := c.UserValue("cid").(string) - idObj, _ := c.UserValue("oid").(string) + idObj, _ := url.PathUnescape(c.UserValue("oid").(string)) ctx := utils.GetContextFromRequest(c) reqLog := utils.GetReqLogOrDefault(ctx, h.log) @@ -205,6 +209,16 @@ func (h *Handler) byAddress(c *fasthttp.RequestCtx, f func(context.Context, requ objID := new(oid.ID) if err = objID.DecodeString(idObj); err != nil { + if h.config.IndexPageEnabled() { + c.SetStatusCode(fasthttp.StatusNotFound) + h.browseObjects(c, browseParams{ + bucketInfo: bktInfo, + prefix: idObj, + listObjects: h.getDirObjectsNative, + isNative: true, + }) + return + } log.Error(logs.WrongObjectID, zap.Error(err)) response.Error(c, "wrong object id", fasthttp.StatusBadRequest) return @@ -243,7 +257,12 @@ func (h *Handler) byObjectName(c *fasthttp.RequestCtx, f func(context.Context, r if isDir(unescapedKey) || isContainerRoot(unescapedKey) { if code := checkErrorType(err); code == fasthttp.StatusNotFound || code == fasthttp.StatusOK { c.SetStatusCode(code) - h.browseObjects(c, bktInfo, unescapedKey) + h.browseObjects(c, browseParams{ + bucketInfo: bktInfo, + prefix: unescapedKey, + listObjects: h.getDirObjectsS3, + isNative: false, + }) return } } @@ -298,7 +317,7 @@ func (h *Handler) byAttribute(c *fasthttp.RequestCtx, f func(context.Context, re return } - res, err := h.search(ctx, &bktInfo.CID, key, val, object.MatchStringEqual) + res, err := h.search(ctx, bktInfo.CID, key, val, object.MatchStringEqual) if err != nil { log.Error(logs.CouldNotSearchForObjects, zap.Error(err)) response.Error(c, "could not search for objects: "+err.Error(), fasthttp.StatusBadRequest) @@ -394,25 +413,3 @@ func (h *Handler) readContainer(ctx context.Context, cnrID cid.ID) (*data.Bucket return bktInfo, err } - -func (h *Handler) listObjects(ctx context.Context, bucketInfo *data.BucketInfo, prefix string) ([]map[string]string, error) { - nodes, _, err := h.tree.GetSubTreeByPrefix(ctx, bucketInfo, prefix, true) - if err != nil { - return nil, err - } - - var objects = make([]map[string]string, 0, len(nodes)) - for _, node := range nodes { - meta := node.GetMeta() - if meta == nil { - continue - } - var obj = make(map[string]string, len(meta)) - for _, m := range meta { - obj[m.GetKey()] = string(m.GetValue()) - } - objects = append(objects, obj) - } - - return objects, nil -} diff --git a/internal/handler/handler_test.go b/internal/handler/handler_test.go index 4fe9153..c7eb37c 100644 --- a/internal/handler/handler_test.go +++ b/internal/handler/handler_test.go @@ -26,6 +26,7 @@ import ( oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id" "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/user" "github.com/nspcc-dev/neo-go/pkg/crypto/keys" + "github.com/panjf2000/ants/v2" "github.com/stretchr/testify/require" "github.com/valyala/fasthttp" "go.uber.org/zap" @@ -45,6 +46,10 @@ func (t *treeClientMock) GetSubTree(context.Context, *data.BucketInfo, string, [ type configMock struct { } +func (c *configMock) WorkerPoolSize() int { + return defaultPoolSize +} + func (c *configMock) DefaultTimestamp() bool { return false } @@ -57,10 +62,11 @@ func (c *configMock) IndexPageEnabled() bool { return false } -func (c *configMock) IndexPageTemplatePath() string { +func (c *configMock) IndexPageTemplate() string { return "" } -func (c *configMock) IndexPageTemplate() string { + +func (c *configMock) IndexPageNativeTemplate() string { return "" } @@ -126,7 +132,11 @@ func prepareHandlerContext() (*handlerContext, error) { treeMock := &treeClientMock{} cfgMock := &configMock{} - handler := New(params, cfgMock, tree.NewTree(treeMock)) + workerPool, err := ants.NewPool(defaultPoolSize) + if err != nil { + return nil, err + } + handler := New(params, cfgMock, tree.NewTree(treeMock), workerPool) return &handlerContext{ key: key, diff --git a/internal/handler/utils.go b/internal/handler/utils.go index a944b67..42665a9 100644 --- a/internal/handler/utils.go +++ b/internal/handler/utils.go @@ -13,6 +13,7 @@ import ( "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/bearer" "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client" cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id" + "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object" oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id" "github.com/valyala/fasthttp" "go.uber.org/zap" @@ -61,6 +62,14 @@ func checkErrorType(err error) int { } } +func loadAttributes(attrs []object.Attribute) map[string]string { + result := make(map[string]string) + for _, attr := range attrs { + result[attr.Key()] = attr.Value() + } + return result +} + func isValidToken(s string) bool { for _, c := range s { if c <= ' ' || c > 127 { diff --git a/internal/logs/logs.go b/internal/logs/logs.go index 409f87d..4dfa21f 100644 --- a/internal/logs/logs.go +++ b/internal/logs/logs.go @@ -31,7 +31,8 @@ const ( CouldNotStoreFileInFrostfs = "could not store file in frostfs" // Error in ../../uploader/upload.go AddAttributeToResultObject = "add attribute to result object" // Debug in ../../uploader/filter.go FailedToCreateResolver = "failed to create resolver" // Fatal in ../../app.go - FailedToReadIndexPageTemplate = "failed to read index page template, set default" // Warn in ../../app.go + FailedToCreateWorkerPool = "failed to create worker pool" // Fatal in ../../app.go + FailedToReadIndexPageTemplate = "failed to read index page template" // Error in ../../app.go SetCustomIndexPageTemplate = "set custom index page template" // Info in ../../app.go ContainerResolverWillBeDisabledBecauseOfResolversResolverOrderIsEmpty = "container resolver will be disabled because of resolvers 'resolver_order' is empty" // Info in ../../app.go MetricsAreDisabled = "metrics are disabled" // Warn in ../../app.go @@ -71,6 +72,9 @@ const ( AddedStoragePeer = "added storage peer" // Info in ../../settings.go CouldntGetBucket = "could not get bucket" // Error in ../handler/utils.go CouldntPutBucketIntoCache = "couldn't put bucket info into cache" // Warn in ../handler/handler.go + FailedToSumbitTaskToPool = "failed to submit task to pool" // Error in ../handler/browse.go + FailedToHeadObject = "failed to head object" // Error in ../handler/browse.go + FailedToIterateOverResponse = "failed to iterate over search response" // Error in ../handler/browse.go InvalidCacheEntryType = "invalid cache entry type" // Warn in ../cache/buckets.go InvalidLifetimeUsingDefaultValue = "invalid lifetime, using default value (in seconds)" // Error in ../../cmd/http-gw/settings.go InvalidCacheSizeUsingDefaultValue = "invalid cache size, using default value" // Error in ../../cmd/http-gw/settings.go diff --git a/internal/templates/index.gotmpl b/internal/templates/index.gotmpl index ea66a62..52d7443 100644 --- a/internal/templates/index.gotmpl +++ b/internal/templates/index.gotmpl @@ -1,10 +1,11 @@ -{{$bucketName := .BucketName}} +{{$container := .Container}} {{ $prefix := trimPrefix .Prefix }} - Index of s3://{{$bucketName}}/{{if $prefix}}/{{$prefix}}/{{end}} + Index of {{.Protocol}}://{{$container}} + /{{if $prefix}}/{{$prefix}}/{{end}} -

Index of s3://{{$bucketName}}/{{if $prefix}}{{$prefix}}/{{end}}

+

Index of {{.Protocol}}://{{$container}}/{{if $prefix}}{{$prefix}}/{{end}}

+ @@ -42,20 +48,22 @@ {{if $trimmedPrefix }} + {{else}} + {{end}} {{range .Objects}} @@ -63,21 +71,22 @@ - - + + +
FilenameOID Size Created Download
- ⮐.. + ⮐..
- ⮐.. + ⮐..
{{if .IsDir}} 🗀 - + {{.FileName}}/ {{else}} 🗎 - + {{.FileName}} {{end}} {{if not .IsDir}}{{ formatSize .Size }}{{end}}{{if not .IsDir}}{{ formatTimestamp .Created }}{{end}}{{.OID}}{{if .Size}}{{ formatSize .Size }}{{end}}{{ .Created }} - {{ if not .IsDir }} - + {{ if .OID }} + Link {{ end }}