2024-01-17 12:14:26 +00:00
|
|
|
package layer
|
|
|
|
|
|
|
|
import (
|
|
|
|
"context"
|
|
|
|
"errors"
|
|
|
|
"fmt"
|
|
|
|
"io"
|
|
|
|
"sort"
|
|
|
|
"strings"
|
|
|
|
"sync"
|
|
|
|
|
|
|
|
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/cache"
|
|
|
|
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/data"
|
|
|
|
s3errors "git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/errors"
|
|
|
|
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/middleware"
|
|
|
|
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/internal/logs"
|
|
|
|
"github.com/panjf2000/ants/v2"
|
|
|
|
"go.uber.org/zap"
|
|
|
|
)
|
|
|
|
|
|
|
|
type (
|
|
|
|
// ListObjectsParamsCommon contains common parameters for ListObjectsV1 and ListObjectsV2.
|
|
|
|
ListObjectsParamsCommon struct {
|
|
|
|
BktInfo *data.BucketInfo
|
|
|
|
Delimiter string
|
|
|
|
Encode string
|
|
|
|
MaxKeys int
|
|
|
|
Prefix string
|
|
|
|
}
|
|
|
|
|
|
|
|
// ListObjectsParamsV1 contains params for ListObjectsV1.
|
|
|
|
ListObjectsParamsV1 struct {
|
|
|
|
ListObjectsParamsCommon
|
|
|
|
Marker string
|
|
|
|
}
|
|
|
|
|
|
|
|
// ListObjectsParamsV2 contains params for ListObjectsV2.
|
|
|
|
ListObjectsParamsV2 struct {
|
|
|
|
ListObjectsParamsCommon
|
|
|
|
ContinuationToken string
|
|
|
|
StartAfter string
|
|
|
|
FetchOwner bool
|
|
|
|
}
|
|
|
|
|
|
|
|
// ListObjectsInfo contains common fields of data for ListObjectsV1 and ListObjectsV2.
|
|
|
|
ListObjectsInfo struct {
|
|
|
|
Prefixes []string
|
2024-01-17 14:26:02 +00:00
|
|
|
Objects []*data.NodeVersion
|
2024-01-17 12:14:26 +00:00
|
|
|
IsTruncated bool
|
|
|
|
}
|
|
|
|
|
|
|
|
// ListObjectsInfoV1 holds data which ListObjectsV1 returns.
|
|
|
|
ListObjectsInfoV1 struct {
|
|
|
|
ListObjectsInfo
|
|
|
|
NextMarker string
|
|
|
|
}
|
|
|
|
|
|
|
|
// ListObjectsInfoV2 holds data which ListObjectsV2 returns.
|
|
|
|
ListObjectsInfoV2 struct {
|
|
|
|
ListObjectsInfo
|
|
|
|
NextContinuationToken string
|
|
|
|
}
|
|
|
|
|
|
|
|
// ListObjectVersionsInfo stores info and list of objects versions.
|
|
|
|
ListObjectVersionsInfo struct {
|
|
|
|
CommonPrefixes []string
|
|
|
|
IsTruncated bool
|
|
|
|
KeyMarker string
|
|
|
|
NextKeyMarker string
|
|
|
|
NextVersionIDMarker string
|
2024-01-17 14:26:02 +00:00
|
|
|
Version []*data.ExtendedNodeVersion
|
|
|
|
DeleteMarker []*data.ExtendedNodeVersion
|
2024-01-17 12:14:26 +00:00
|
|
|
VersionIDMarker string
|
|
|
|
}
|
|
|
|
|
|
|
|
allObjectParams struct {
|
|
|
|
Bucket *data.BucketInfo
|
|
|
|
Delimiter string
|
|
|
|
Prefix string
|
|
|
|
MaxKeys int
|
|
|
|
Marker string
|
|
|
|
ContinuationToken string
|
|
|
|
}
|
|
|
|
)
|
|
|
|
|
|
|
|
// ListObjectsV1 returns objects in a bucket for requests of Version 1.
|
|
|
|
func (n *layer) ListObjectsV1(ctx context.Context, p *ListObjectsParamsV1) (*ListObjectsInfoV1, error) {
|
|
|
|
var result ListObjectsInfoV1
|
|
|
|
|
|
|
|
prm := allObjectParams{
|
|
|
|
Bucket: p.BktInfo,
|
|
|
|
Delimiter: p.Delimiter,
|
|
|
|
Prefix: p.Prefix,
|
|
|
|
MaxKeys: p.MaxKeys,
|
|
|
|
Marker: p.Marker,
|
|
|
|
}
|
|
|
|
|
|
|
|
objects, next, err := n.getLatestObjectsVersions(ctx, prm)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
if next != nil {
|
|
|
|
result.IsTruncated = true
|
2024-01-17 14:26:02 +00:00
|
|
|
result.NextMarker = objects[len(objects)-1].FilePath
|
2024-01-17 12:14:26 +00:00
|
|
|
}
|
|
|
|
|
2024-01-17 14:26:02 +00:00
|
|
|
result.Prefixes, result.Objects = triageObjects(objects, p.Prefix, p.Delimiter)
|
2024-01-17 12:14:26 +00:00
|
|
|
|
|
|
|
return &result, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// ListObjectsV2 returns objects in a bucket for requests of Version 2.
|
|
|
|
func (n *layer) ListObjectsV2(ctx context.Context, p *ListObjectsParamsV2) (*ListObjectsInfoV2, error) {
|
|
|
|
var result ListObjectsInfoV2
|
|
|
|
|
|
|
|
prm := allObjectParams{
|
|
|
|
Bucket: p.BktInfo,
|
|
|
|
Delimiter: p.Delimiter,
|
|
|
|
Prefix: p.Prefix,
|
|
|
|
MaxKeys: p.MaxKeys,
|
|
|
|
Marker: p.StartAfter,
|
|
|
|
ContinuationToken: p.ContinuationToken,
|
|
|
|
}
|
|
|
|
|
2024-01-17 14:26:02 +00:00
|
|
|
objects, next, err := n.getLatestObjectsVersions(ctx, prm)
|
2024-01-17 12:14:26 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
if next != nil {
|
|
|
|
result.IsTruncated = true
|
2024-01-17 14:26:02 +00:00
|
|
|
result.NextContinuationToken = next.OID.EncodeToString()
|
2024-01-17 12:14:26 +00:00
|
|
|
}
|
|
|
|
|
2024-01-17 14:26:02 +00:00
|
|
|
result.Prefixes, result.Objects = triageObjects(objects, p.Prefix, p.Delimiter)
|
2024-01-17 12:14:26 +00:00
|
|
|
|
|
|
|
return &result, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (n *layer) ListObjectVersions(ctx context.Context, p *ListObjectVersionsParams) (*ListObjectVersionsInfo, error) {
|
2024-01-19 06:26:58 +00:00
|
|
|
var err error
|
|
|
|
owner := n.BearerOwner(ctx)
|
|
|
|
|
|
|
|
versions := make(map[string][]*data.ExtendedNodeVersion, p.MaxKeys)
|
|
|
|
|
|
|
|
cacheKey := cache.CreateListSessionCacheKey(p.BktInfo.CID, p.Prefix, p.VersionIDMarker)
|
|
|
|
session := n.cache.GetListSession(owner, cacheKey)
|
|
|
|
if session != nil {
|
|
|
|
// after reading next object from stream in session
|
|
|
|
// the current cache value already doesn't match with next token in cache key
|
|
|
|
n.cache.DeleteListSession(owner, cacheKey)
|
|
|
|
} else {
|
|
|
|
session = &data.ListSession{NamesMap: make(map[string]struct{})}
|
|
|
|
session.Context, session.Cancel = context.WithCancel(context.Background())
|
|
|
|
|
|
|
|
if bd, err := middleware.GetBoxData(ctx); err == nil {
|
|
|
|
session.Context = middleware.SetBoxData(session.Context, bd)
|
|
|
|
}
|
|
|
|
|
|
|
|
session.Stream, err = n.treeService.GetAllVersionsByPrefixStream(session.Context, p.BktInfo, p.Prefix)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
poolCtx, cancel := context.WithCancel(ctx)
|
|
|
|
defer cancel()
|
|
|
|
|
|
|
|
pp := allObjectParams{
|
|
|
|
Bucket: p.BktInfo,
|
|
|
|
Delimiter: p.Delimiter,
|
|
|
|
Prefix: p.Prefix,
|
|
|
|
Marker: p.KeyMarker,
|
|
|
|
ContinuationToken: p.VersionIDMarker,
|
|
|
|
MaxKeys: p.MaxKeys,
|
|
|
|
}
|
|
|
|
|
|
|
|
generator, errorCh := nodesGeneratorVersions(poolCtx, pp, session)
|
|
|
|
objOutCh, err := n.initWorkerPoolVersions(poolCtx, 2, pp, generator)
|
2024-01-17 12:14:26 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
2024-01-19 06:26:58 +00:00
|
|
|
for eoi := range objOutCh {
|
|
|
|
name := eoi.NodeVersion.FilePath
|
|
|
|
|
|
|
|
dirName := tryDirectoryName(eoi.NodeVersion, p.Prefix, p.Delimiter)
|
|
|
|
if dirName != "" {
|
|
|
|
name = dirName
|
|
|
|
}
|
|
|
|
|
|
|
|
objVersions, ok := versions[name]
|
|
|
|
if !ok {
|
|
|
|
objVersions = []*data.ExtendedNodeVersion{eoi}
|
|
|
|
} else if dirName == "" {
|
|
|
|
objVersions = append(objVersions, eoi)
|
|
|
|
}
|
|
|
|
|
|
|
|
versions[name] = objVersions
|
|
|
|
}
|
|
|
|
|
|
|
|
if err = <-errorCh; err != nil {
|
|
|
|
return nil, fmt.Errorf("failed to get next object from stream: %w", err)
|
|
|
|
}
|
|
|
|
|
2024-01-17 12:14:26 +00:00
|
|
|
sortedNames := make([]string, 0, len(versions))
|
|
|
|
for k := range versions {
|
|
|
|
sortedNames = append(sortedNames, k)
|
|
|
|
}
|
|
|
|
sort.Strings(sortedNames)
|
|
|
|
|
2024-01-17 14:26:02 +00:00
|
|
|
allObjects := make([]*data.ExtendedNodeVersion, 0, p.MaxKeys)
|
2024-01-17 12:14:26 +00:00
|
|
|
|
|
|
|
for _, name := range sortedNames {
|
|
|
|
sortedVersions := versions[name]
|
|
|
|
sort.Slice(sortedVersions, func(i, j int) bool {
|
|
|
|
return sortedVersions[j].NodeVersion.Timestamp < sortedVersions[i].NodeVersion.Timestamp // sort in reverse order
|
|
|
|
})
|
|
|
|
|
|
|
|
for i, version := range sortedVersions {
|
|
|
|
version.IsLatest = i == 0
|
|
|
|
allObjects = append(allObjects, version)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2024-01-19 06:26:58 +00:00
|
|
|
//if allObjects, err = filterVersionsByMarker(allObjects, p); err != nil {
|
|
|
|
// return nil, err
|
|
|
|
//}
|
2024-01-17 12:14:26 +00:00
|
|
|
|
|
|
|
res := &ListObjectVersionsInfo{
|
|
|
|
KeyMarker: p.KeyMarker,
|
|
|
|
VersionIDMarker: p.VersionIDMarker,
|
|
|
|
}
|
|
|
|
|
2024-01-17 14:26:02 +00:00
|
|
|
res.CommonPrefixes, allObjects = triageExtendedObjects(allObjects, p.Prefix, p.Delimiter)
|
2024-01-17 12:14:26 +00:00
|
|
|
|
|
|
|
if len(allObjects) > p.MaxKeys {
|
|
|
|
res.IsTruncated = true
|
2024-01-17 14:26:02 +00:00
|
|
|
res.NextKeyMarker = allObjects[p.MaxKeys-1].NodeVersion.FilePath
|
|
|
|
res.NextVersionIDMarker = allObjects[p.MaxKeys-1].NodeVersion.OID.EncodeToString()
|
2024-01-17 12:14:26 +00:00
|
|
|
|
2024-01-19 06:26:58 +00:00
|
|
|
session.Next = []*data.NodeVersion{allObjects[p.MaxKeys-1].NodeVersion, allObjects[p.MaxKeys].NodeVersion}
|
|
|
|
n.cache.PutListSession(owner, cache.CreateListSessionCacheKey(p.BktInfo.CID, p.Prefix, res.NextVersionIDMarker), session)
|
|
|
|
|
2024-01-17 12:14:26 +00:00
|
|
|
allObjects = allObjects[:p.MaxKeys]
|
|
|
|
}
|
|
|
|
|
|
|
|
res.Version, res.DeleteMarker = triageVersions(allObjects)
|
|
|
|
return res, nil
|
|
|
|
}
|
|
|
|
|
2024-01-17 14:26:02 +00:00
|
|
|
func (n *layer) getLatestObjectsVersions(ctx context.Context, p allObjectParams) (objects []*data.NodeVersion, next *data.NodeVersion, err error) {
|
2024-01-17 12:14:26 +00:00
|
|
|
if p.MaxKeys == 0 {
|
|
|
|
return nil, nil, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
owner := n.BearerOwner(ctx)
|
2024-01-19 06:26:58 +00:00
|
|
|
cacheKey := cache.CreateListSessionCacheKey(p.Bucket.CID, p.Prefix, p.ContinuationToken) // todo redo for listv1
|
2024-01-17 12:14:26 +00:00
|
|
|
session := n.cache.GetListSession(owner, cacheKey)
|
|
|
|
if session != nil {
|
|
|
|
// after reading next object from stream in session
|
|
|
|
// the current cache value already doesn't match with next token in cache key
|
|
|
|
n.cache.DeleteListSession(owner, cacheKey)
|
|
|
|
} else {
|
|
|
|
session = &data.ListSession{NamesMap: make(map[string]struct{})}
|
|
|
|
session.Context, session.Cancel = context.WithCancel(context.Background())
|
|
|
|
|
|
|
|
if bd, err := middleware.GetBoxData(ctx); err == nil {
|
|
|
|
session.Context = middleware.SetBoxData(session.Context, bd)
|
|
|
|
}
|
|
|
|
|
|
|
|
session.Stream, err = n.treeService.GetLatestVersionsByPrefixStream(session.Context, p.Bucket, p.Prefix)
|
|
|
|
if err != nil {
|
|
|
|
return nil, nil, err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
poolCtx, cancel := context.WithCancel(ctx)
|
|
|
|
defer cancel()
|
|
|
|
|
|
|
|
generator, errorCh := nodesGeneratorStream(poolCtx, p, session)
|
|
|
|
objOutCh, err := n.initWorkerPoolStream(poolCtx, 2, p, generator)
|
|
|
|
if err != nil {
|
|
|
|
return nil, nil, fmt.Errorf("failed to init worker pool: %w", err)
|
|
|
|
}
|
|
|
|
|
2024-01-17 14:26:02 +00:00
|
|
|
objects = make([]*data.NodeVersion, 0, p.MaxKeys+1)
|
2024-01-19 06:26:58 +00:00
|
|
|
objects = append(objects, session.Next...)
|
2024-01-17 12:14:26 +00:00
|
|
|
|
|
|
|
for obj := range objOutCh {
|
|
|
|
objects = append(objects, obj)
|
|
|
|
}
|
|
|
|
|
|
|
|
if err = <-errorCh; err != nil {
|
|
|
|
return nil, nil, fmt.Errorf("failed to get next object from stream: %w", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
if len(objects) > p.MaxKeys {
|
|
|
|
next = objects[p.MaxKeys]
|
|
|
|
objects = objects[:p.MaxKeys]
|
|
|
|
}
|
|
|
|
|
|
|
|
if next != nil {
|
2024-01-19 06:26:58 +00:00
|
|
|
session.Next = []*data.NodeVersion{next}
|
2024-01-17 14:26:02 +00:00
|
|
|
n.cache.PutListSession(owner, cache.CreateListSessionCacheKey(p.Bucket.CID, p.Prefix, next.OID.EncodeToString()), session)
|
2024-01-17 12:14:26 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2024-01-19 06:26:58 +00:00
|
|
|
func (n *layer) getAllObjectsVersions(ctx context.Context, p *ListObjectVersionsParams) (versions map[string][]*data.ExtendedNodeVersion, err error) {
|
|
|
|
owner := n.BearerOwner(ctx)
|
2024-01-17 12:14:26 +00:00
|
|
|
|
2024-01-19 06:26:58 +00:00
|
|
|
versions = make(map[string][]*data.ExtendedNodeVersion, p.MaxKeys)
|
2024-01-17 12:14:26 +00:00
|
|
|
|
2024-01-19 06:26:58 +00:00
|
|
|
cacheKey := cache.CreateListSessionCacheKey(p.BktInfo.CID, p.Prefix, p.VersionIDMarker)
|
|
|
|
session := n.cache.GetListSession(owner, cacheKey)
|
|
|
|
if session != nil {
|
|
|
|
// after reading next object from stream in session
|
|
|
|
// the current cache value already doesn't match with next token in cache key
|
|
|
|
n.cache.DeleteListSession(owner, cacheKey)
|
|
|
|
} else {
|
|
|
|
session = &data.ListSession{NamesMap: make(map[string]struct{})}
|
|
|
|
session.Context, session.Cancel = context.WithCancel(context.Background())
|
|
|
|
|
|
|
|
if bd, err := middleware.GetBoxData(ctx); err == nil {
|
|
|
|
session.Context = middleware.SetBoxData(session.Context, bd)
|
|
|
|
}
|
|
|
|
|
|
|
|
session.Stream, err = n.treeService.GetAllVersionsByPrefixStream(session.Context, p.BktInfo, p.Prefix)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
}
|
2024-01-17 12:14:26 +00:00
|
|
|
|
|
|
|
poolCtx, cancel := context.WithCancel(ctx)
|
|
|
|
defer cancel()
|
|
|
|
|
|
|
|
pp := allObjectParams{
|
|
|
|
Bucket: p.BktInfo,
|
|
|
|
Delimiter: p.Delimiter,
|
|
|
|
Prefix: p.Prefix,
|
|
|
|
Marker: p.KeyMarker,
|
|
|
|
ContinuationToken: p.VersionIDMarker,
|
|
|
|
MaxKeys: p.MaxKeys,
|
|
|
|
}
|
|
|
|
|
2024-01-19 06:26:58 +00:00
|
|
|
generator, errorCh := nodesGeneratorVersions(poolCtx, pp, session)
|
|
|
|
objOutCh, err := n.initWorkerPoolVersions(poolCtx, 2, pp, generator)
|
2024-01-17 12:14:26 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
2024-01-19 06:26:58 +00:00
|
|
|
//if session.Next != nil {
|
|
|
|
// name := session.Next.FilePath
|
|
|
|
// dirName := tryDirectoryName(session.Next, p.Prefix, p.Delimiter)
|
|
|
|
// if dirName != "" {
|
|
|
|
// name = dirName
|
|
|
|
// }
|
|
|
|
//
|
|
|
|
// versions[name] = []*data.ExtendedNodeVersion{{NodeVersion: session.Next}}
|
|
|
|
//}
|
|
|
|
|
2024-01-17 12:14:26 +00:00
|
|
|
for eoi := range objOutCh {
|
2024-01-19 06:26:58 +00:00
|
|
|
name := eoi.NodeVersion.FilePath
|
|
|
|
|
|
|
|
dirName := tryDirectoryName(eoi.NodeVersion, p.Prefix, p.Delimiter)
|
|
|
|
if dirName != "" {
|
|
|
|
name = dirName
|
|
|
|
}
|
|
|
|
|
|
|
|
objVersions, ok := versions[name]
|
2024-01-17 12:14:26 +00:00
|
|
|
if !ok {
|
2024-01-17 14:26:02 +00:00
|
|
|
objVersions = []*data.ExtendedNodeVersion{eoi}
|
2024-01-19 06:26:58 +00:00
|
|
|
} else if dirName == "" {
|
2024-01-17 12:14:26 +00:00
|
|
|
objVersions = append(objVersions, eoi)
|
|
|
|
}
|
2024-01-19 06:26:58 +00:00
|
|
|
|
|
|
|
versions[name] = objVersions
|
|
|
|
}
|
|
|
|
|
|
|
|
if err = <-errorCh; err != nil {
|
|
|
|
return nil, fmt.Errorf("failed to get next object from stream: %w", err)
|
2024-01-17 12:14:26 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
return versions, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func nodesGeneratorStream(ctx context.Context, p allObjectParams, stream *data.ListSession) (<-chan *data.NodeVersion, <-chan error) {
|
|
|
|
nodeCh := make(chan *data.NodeVersion, 1000)
|
|
|
|
errCh := make(chan error, 1)
|
|
|
|
//existed := make(map[string]struct{}, p.MaxKeys) // to squash the same directories
|
|
|
|
existed := stream.NamesMap
|
|
|
|
|
2024-01-19 06:26:58 +00:00
|
|
|
if len(stream.Next) != 0 {
|
2024-01-17 12:14:26 +00:00
|
|
|
existed[continuationToken] = struct{}{}
|
|
|
|
}
|
|
|
|
|
|
|
|
limit := p.MaxKeys
|
2024-01-19 06:26:58 +00:00
|
|
|
if len(stream.Next) == 0 {
|
2024-01-17 12:14:26 +00:00
|
|
|
limit++
|
|
|
|
}
|
|
|
|
|
|
|
|
go func() {
|
|
|
|
var generated int
|
|
|
|
var err error
|
|
|
|
|
|
|
|
LOOP:
|
|
|
|
for err == nil {
|
|
|
|
node, err := stream.Stream.Next(ctx)
|
|
|
|
if err != nil {
|
|
|
|
if !errors.Is(err, io.EOF) {
|
|
|
|
errCh <- fmt.Errorf("stream next: %w", err)
|
|
|
|
}
|
|
|
|
break LOOP
|
|
|
|
}
|
|
|
|
|
|
|
|
if shouldSkip(node, p, existed) {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
select {
|
|
|
|
case <-ctx.Done():
|
|
|
|
break LOOP
|
|
|
|
case nodeCh <- node:
|
|
|
|
generated++
|
|
|
|
|
|
|
|
if generated == limit { // we use maxKeys+1 to be able to know nextMarker/nextContinuationToken
|
|
|
|
break LOOP
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
close(nodeCh)
|
|
|
|
close(errCh)
|
|
|
|
}()
|
|
|
|
|
|
|
|
return nodeCh, errCh
|
|
|
|
}
|
2024-01-19 06:26:58 +00:00
|
|
|
|
|
|
|
func nodesGeneratorVersions(ctx context.Context, p allObjectParams, stream *data.ListSession) (<-chan *data.NodeVersion, <-chan error) {
|
|
|
|
nodeCh := make(chan *data.NodeVersion, 1000)
|
|
|
|
errCh := make(chan error, 1)
|
|
|
|
//existed := make(map[string]struct{}, p.MaxKeys) // to squash the same directories
|
|
|
|
existed := stream.NamesMap
|
|
|
|
|
|
|
|
delete(existed, continuationToken)
|
|
|
|
|
|
|
|
//if len(stream.Next) != 0 {
|
|
|
|
// existed[continuationToken] = struct{}{}
|
|
|
|
//}
|
|
|
|
|
|
|
|
limit := p.MaxKeys + 1
|
|
|
|
//if len(stream.Next) == 0 {
|
|
|
|
// limit++
|
|
|
|
//}
|
2024-01-17 12:14:26 +00:00
|
|
|
|
|
|
|
go func() {
|
2024-01-17 14:26:02 +00:00
|
|
|
var generated int
|
2024-01-19 06:26:58 +00:00
|
|
|
var err error
|
|
|
|
|
|
|
|
ind := 0
|
|
|
|
|
2024-01-17 12:14:26 +00:00
|
|
|
LOOP:
|
2024-01-19 06:26:58 +00:00
|
|
|
for err == nil {
|
|
|
|
var node *data.NodeVersion
|
|
|
|
if ind < len(stream.Next) {
|
|
|
|
node = stream.Next[ind]
|
|
|
|
ind++
|
|
|
|
} else {
|
|
|
|
node, err = stream.Stream.Next(ctx)
|
|
|
|
if err != nil {
|
|
|
|
if !errors.Is(err, io.EOF) {
|
|
|
|
errCh <- fmt.Errorf("stream next: %w", err)
|
|
|
|
}
|
|
|
|
break LOOP
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2024-01-17 14:26:02 +00:00
|
|
|
if shouldSkipVersions(node, p, existed) {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
2024-01-17 12:14:26 +00:00
|
|
|
select {
|
|
|
|
case <-ctx.Done():
|
|
|
|
break LOOP
|
2024-01-17 14:26:02 +00:00
|
|
|
case nodeCh <- node:
|
|
|
|
generated++
|
2024-01-19 06:26:58 +00:00
|
|
|
|
|
|
|
if generated == limit { // we use maxKeys+1 to be able to know nextMarker/nextContinuationToken
|
2024-01-17 14:26:02 +00:00
|
|
|
break LOOP
|
2024-01-17 12:14:26 +00:00
|
|
|
}
|
2024-01-17 14:26:02 +00:00
|
|
|
}
|
2024-01-17 12:14:26 +00:00
|
|
|
}
|
2024-01-17 14:26:02 +00:00
|
|
|
close(nodeCh)
|
2024-01-19 06:26:58 +00:00
|
|
|
close(errCh)
|
2024-01-17 12:14:26 +00:00
|
|
|
}()
|
|
|
|
|
2024-01-19 06:26:58 +00:00
|
|
|
return nodeCh, errCh
|
2024-01-17 12:14:26 +00:00
|
|
|
}
|
|
|
|
|
2024-01-17 14:26:02 +00:00
|
|
|
func (n *layer) initWorkerPoolStream(ctx context.Context, size int, p allObjectParams, input <-chan *data.NodeVersion) (<-chan *data.NodeVersion, error) {
|
2024-01-17 12:14:26 +00:00
|
|
|
reqLog := n.reqLogger(ctx)
|
|
|
|
pool, err := ants.NewPool(size, ants.WithLogger(&logWrapper{reqLog}))
|
|
|
|
if err != nil {
|
|
|
|
return nil, fmt.Errorf("coudln't init go pool for listing: %w", err)
|
|
|
|
}
|
2024-01-17 14:26:02 +00:00
|
|
|
objCh := make(chan *data.NodeVersion, size)
|
2024-01-17 12:14:26 +00:00
|
|
|
|
|
|
|
go func() {
|
|
|
|
var wg sync.WaitGroup
|
|
|
|
|
|
|
|
LOOP:
|
|
|
|
for node := range input {
|
|
|
|
select {
|
|
|
|
case <-ctx.Done():
|
|
|
|
break LOOP
|
|
|
|
default:
|
|
|
|
}
|
|
|
|
|
2024-01-19 12:15:15 +00:00
|
|
|
if dirName := tryDirectoryName(node, p.Prefix, p.Delimiter); dirName != "" || node.IsFilledExtra() { // todo think to not compute twice
|
|
|
|
if dirName != "" {
|
|
|
|
node.FilePath = dirName
|
|
|
|
}
|
2024-01-17 14:26:02 +00:00
|
|
|
select {
|
|
|
|
case <-ctx.Done():
|
|
|
|
case objCh <- node:
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
// We have to make a copy of pointer to data.NodeVersion
|
|
|
|
// to get correct value in submitted task function.
|
|
|
|
func(node *data.NodeVersion) {
|
|
|
|
wg.Add(1)
|
|
|
|
err = pool.Submit(func() {
|
|
|
|
defer wg.Done()
|
|
|
|
oi := n.objectInfoFromObjectsCacheOrFrostFS(ctx, p.Bucket, node)
|
2024-01-17 12:14:26 +00:00
|
|
|
if oi == nil {
|
|
|
|
// try to get object again
|
2024-01-17 14:26:02 +00:00
|
|
|
if oi = n.objectInfoFromObjectsCacheOrFrostFS(ctx, p.Bucket, node); oi == nil {
|
2024-01-17 12:14:26 +00:00
|
|
|
// do not process object which are definitely missing in object service
|
|
|
|
return
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2024-01-17 14:26:02 +00:00
|
|
|
node.FillExtra(oi)
|
2024-01-17 12:14:26 +00:00
|
|
|
|
2024-01-17 14:26:02 +00:00
|
|
|
select {
|
|
|
|
case <-ctx.Done():
|
|
|
|
case objCh <- node:
|
|
|
|
}
|
|
|
|
})
|
|
|
|
if err != nil {
|
|
|
|
wg.Done()
|
|
|
|
reqLog.Warn(logs.FailedToSubmitTaskToPool, zap.Error(err))
|
2024-01-17 12:14:26 +00:00
|
|
|
}
|
2024-01-17 14:26:02 +00:00
|
|
|
}(node)
|
|
|
|
}
|
2024-01-17 12:14:26 +00:00
|
|
|
}
|
|
|
|
wg.Wait()
|
|
|
|
close(objCh)
|
|
|
|
pool.Release()
|
|
|
|
}()
|
|
|
|
|
|
|
|
return objCh, nil
|
|
|
|
}
|
|
|
|
|
2024-01-17 14:26:02 +00:00
|
|
|
func (n *layer) initWorkerPoolVersions(ctx context.Context, size int, p allObjectParams, input <-chan *data.NodeVersion) (<-chan *data.ExtendedNodeVersion, error) {
|
2024-01-17 12:14:26 +00:00
|
|
|
reqLog := n.reqLogger(ctx)
|
|
|
|
pool, err := ants.NewPool(size, ants.WithLogger(&logWrapper{reqLog}))
|
|
|
|
if err != nil {
|
|
|
|
return nil, fmt.Errorf("coudln't init go pool for listing: %w", err)
|
|
|
|
}
|
2024-01-17 14:26:02 +00:00
|
|
|
objCh := make(chan *data.ExtendedNodeVersion)
|
2024-01-17 12:14:26 +00:00
|
|
|
|
|
|
|
go func() {
|
|
|
|
var wg sync.WaitGroup
|
|
|
|
|
|
|
|
LOOP:
|
|
|
|
for node := range input {
|
|
|
|
select {
|
|
|
|
case <-ctx.Done():
|
|
|
|
break LOOP
|
|
|
|
default:
|
|
|
|
}
|
|
|
|
|
2024-01-17 14:26:02 +00:00
|
|
|
if node.IsFilledExtra() {
|
|
|
|
select {
|
|
|
|
case <-ctx.Done():
|
|
|
|
case objCh <- &data.ExtendedNodeVersion{NodeVersion: node}:
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
// We have to make a copy of pointer to data.NodeVersion
|
|
|
|
// to get correct value in submitted task function.
|
|
|
|
func(node *data.NodeVersion) {
|
|
|
|
wg.Add(1)
|
|
|
|
err = pool.Submit(func() {
|
|
|
|
defer wg.Done()
|
|
|
|
|
|
|
|
oi := n.objectInfoFromObjectsCacheOrFrostFS(ctx, p.Bucket, node)
|
|
|
|
if oi == nil {
|
|
|
|
// try to get object again
|
|
|
|
if oi = n.objectInfoFromObjectsCacheOrFrostFS(ctx, p.Bucket, node); oi == nil {
|
|
|
|
// do not process object which are definitely missing in object service
|
|
|
|
return
|
|
|
|
}
|
2024-01-17 12:14:26 +00:00
|
|
|
}
|
2024-01-17 14:26:02 +00:00
|
|
|
|
|
|
|
node.FillExtra(oi)
|
|
|
|
|
|
|
|
select {
|
|
|
|
case <-ctx.Done():
|
|
|
|
case objCh <- &data.ExtendedNodeVersion{NodeVersion: node}:
|
|
|
|
}
|
|
|
|
})
|
|
|
|
if err != nil {
|
|
|
|
wg.Done()
|
|
|
|
reqLog.Warn(logs.FailedToSubmitTaskToPool, zap.Error(err))
|
2024-01-17 12:14:26 +00:00
|
|
|
}
|
2024-01-17 14:26:02 +00:00
|
|
|
}(node)
|
|
|
|
}
|
2024-01-17 12:14:26 +00:00
|
|
|
}
|
|
|
|
wg.Wait()
|
|
|
|
close(objCh)
|
|
|
|
pool.Release()
|
|
|
|
}()
|
|
|
|
|
|
|
|
return objCh, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (n *layer) bucketNodeVersions(ctx context.Context, bkt *data.BucketInfo, prefix string) ([]*data.NodeVersion, error) {
|
|
|
|
var err error
|
|
|
|
|
|
|
|
owner := n.BearerOwner(ctx)
|
|
|
|
cacheKey := cache.CreateObjectsListCacheKey(bkt.CID, prefix, false)
|
|
|
|
nodeVersions := n.cache.GetList(owner, cacheKey)
|
|
|
|
|
|
|
|
if nodeVersions == nil {
|
|
|
|
nodeVersions, err = n.treeService.GetAllVersionsByPrefix(ctx, bkt, prefix)
|
|
|
|
if err != nil {
|
|
|
|
return nil, fmt.Errorf("get all versions from tree service: %w", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
n.cache.PutList(owner, cacheKey, nodeVersions)
|
|
|
|
}
|
|
|
|
|
|
|
|
return nodeVersions, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func shouldSkip(node *data.NodeVersion, p allObjectParams, existed map[string]struct{}) bool {
|
2024-01-17 14:26:02 +00:00
|
|
|
if node.IsDeleteMarker {
|
2024-01-17 12:14:26 +00:00
|
|
|
return true
|
|
|
|
}
|
|
|
|
|
|
|
|
filePath := node.FilePath
|
|
|
|
if dirName := tryDirectoryName(node, p.Prefix, p.Delimiter); len(dirName) != 0 {
|
|
|
|
filePath = dirName
|
|
|
|
}
|
|
|
|
if _, ok := existed[filePath]; ok {
|
|
|
|
return true
|
|
|
|
}
|
|
|
|
|
|
|
|
if filePath <= p.Marker {
|
|
|
|
return true
|
|
|
|
}
|
|
|
|
|
|
|
|
if p.ContinuationToken != "" {
|
|
|
|
if _, ok := existed[continuationToken]; !ok {
|
|
|
|
if p.ContinuationToken != node.OID.EncodeToString() {
|
|
|
|
return true
|
|
|
|
}
|
|
|
|
existed[continuationToken] = struct{}{}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
existed[filePath] = struct{}{}
|
|
|
|
return false
|
|
|
|
}
|
|
|
|
|
|
|
|
func shouldSkipVersions(node *data.NodeVersion, p allObjectParams, existed map[string]struct{}) bool {
|
|
|
|
filePath := node.FilePath
|
|
|
|
if dirName := tryDirectoryName(node, p.Prefix, p.Delimiter); len(dirName) != 0 {
|
|
|
|
filePath = dirName
|
|
|
|
if _, ok := existed[filePath]; ok {
|
|
|
|
return true
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
if filePath < p.Marker {
|
|
|
|
return true
|
|
|
|
}
|
|
|
|
|
|
|
|
if p.ContinuationToken != "" {
|
|
|
|
if _, ok := existed[continuationToken]; !ok {
|
|
|
|
if p.ContinuationToken != node.OID.EncodeToString() {
|
|
|
|
return true
|
|
|
|
}
|
|
|
|
existed[continuationToken] = struct{}{}
|
2024-01-19 06:26:58 +00:00
|
|
|
return true
|
2024-01-17 12:14:26 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
existed[filePath] = struct{}{}
|
|
|
|
return false
|
|
|
|
}
|
|
|
|
|
2024-01-17 14:26:02 +00:00
|
|
|
func triageObjects(allObjects []*data.NodeVersion, prefix, delimiter string) (prefixes []string, objects []*data.NodeVersion) {
|
|
|
|
objects = make([]*data.NodeVersion, 0, len(allObjects))
|
2024-01-17 12:14:26 +00:00
|
|
|
for _, ov := range allObjects {
|
2024-01-17 14:26:02 +00:00
|
|
|
if dirName := tryDirectoryName(ov, prefix, delimiter); dirName != "" {
|
|
|
|
prefixes = append(prefixes, dirName)
|
2024-01-17 12:14:26 +00:00
|
|
|
} else {
|
|
|
|
objects = append(objects, ov)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2024-01-17 14:26:02 +00:00
|
|
|
func triageExtendedObjects(allObjects []*data.ExtendedNodeVersion, prefix, delimiter string) (prefixes []string, objects []*data.ExtendedNodeVersion) {
|
2024-01-17 12:14:26 +00:00
|
|
|
for _, ov := range allObjects {
|
2024-01-17 14:26:02 +00:00
|
|
|
if dirName := tryDirectoryName(ov.NodeVersion, prefix, delimiter); dirName != "" {
|
|
|
|
prefixes = append(prefixes, dirName)
|
2024-01-17 12:14:26 +00:00
|
|
|
} else {
|
|
|
|
objects = append(objects, ov)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2024-01-17 14:26:02 +00:00
|
|
|
func (n *layer) objectInfoFromObjectsCacheOrFrostFS(ctx context.Context, bktInfo *data.BucketInfo, node *data.NodeVersion) (oi *data.ObjectInfo) {
|
2024-01-17 12:14:26 +00:00
|
|
|
owner := n.BearerOwner(ctx)
|
|
|
|
if extInfo := n.cache.GetObject(owner, newAddress(bktInfo.CID, node.OID)); extInfo != nil {
|
|
|
|
return extInfo.ObjectInfo
|
|
|
|
}
|
|
|
|
|
|
|
|
meta, err := n.objectHead(ctx, bktInfo, node.OID)
|
|
|
|
if err != nil {
|
|
|
|
n.reqLogger(ctx).Warn(logs.CouldNotFetchObjectMeta, zap.Error(err))
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
oi = objectInfoFromMeta(bktInfo, meta)
|
|
|
|
oi.MD5Sum = node.MD5
|
|
|
|
n.cache.PutObject(owner, &data.ExtendedObjectInfo{ObjectInfo: oi, NodeVersion: node})
|
|
|
|
|
|
|
|
return oi
|
|
|
|
}
|
|
|
|
|
|
|
|
// tryDirectoryName forms directory name by prefix and delimiter.
|
|
|
|
// If node isn't a directory empty string is returned.
|
|
|
|
// This function doesn't check if node has a prefix. It must do a caller.
|
|
|
|
func tryDirectoryName(node *data.NodeVersion, prefix, delimiter string) string {
|
|
|
|
if len(delimiter) == 0 {
|
|
|
|
return ""
|
|
|
|
}
|
|
|
|
|
|
|
|
tail := strings.TrimPrefix(node.FilePath, prefix)
|
|
|
|
index := strings.Index(tail, delimiter)
|
|
|
|
if index >= 0 {
|
|
|
|
return prefix + tail[:index+1]
|
|
|
|
}
|
|
|
|
|
|
|
|
return ""
|
|
|
|
}
|
|
|
|
|
2024-01-17 14:26:02 +00:00
|
|
|
func filterVersionsByMarker(objects []*data.ExtendedNodeVersion, p *ListObjectVersionsParams) ([]*data.ExtendedNodeVersion, error) {
|
2024-01-17 12:14:26 +00:00
|
|
|
if p.KeyMarker == "" {
|
|
|
|
return objects, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
for i, obj := range objects {
|
2024-01-17 14:26:02 +00:00
|
|
|
if obj.NodeVersion.FilePath == p.KeyMarker {
|
2024-01-17 12:14:26 +00:00
|
|
|
for j := i; j < len(objects); j++ {
|
2024-01-17 14:26:02 +00:00
|
|
|
if objects[j].NodeVersion.FilePath != obj.NodeVersion.FilePath {
|
2024-01-17 12:14:26 +00:00
|
|
|
if p.VersionIDMarker == "" {
|
|
|
|
return objects[j:], nil
|
|
|
|
}
|
|
|
|
break
|
|
|
|
}
|
2024-01-17 14:26:02 +00:00
|
|
|
if objects[j].NodeVersion.OID.EncodeToString() == p.VersionIDMarker {
|
2024-01-17 12:14:26 +00:00
|
|
|
return objects[j+1:], nil
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return nil, s3errors.GetAPIError(s3errors.ErrInvalidVersion)
|
2024-01-17 14:26:02 +00:00
|
|
|
} else if obj.NodeVersion.FilePath > p.KeyMarker {
|
2024-01-17 12:14:26 +00:00
|
|
|
if p.VersionIDMarker != "" {
|
|
|
|
return nil, s3errors.GetAPIError(s3errors.ErrInvalidVersion)
|
|
|
|
}
|
|
|
|
return objects[i:], nil
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// don't use nil as empty slice to be consistent with `return objects[j+1:], nil` above
|
|
|
|
// that can be empty
|
2024-01-17 14:26:02 +00:00
|
|
|
return []*data.ExtendedNodeVersion{}, nil
|
2024-01-17 12:14:26 +00:00
|
|
|
}
|
|
|
|
|
2024-01-17 14:26:02 +00:00
|
|
|
func triageVersions(objVersions []*data.ExtendedNodeVersion) ([]*data.ExtendedNodeVersion, []*data.ExtendedNodeVersion) {
|
2024-01-17 12:14:26 +00:00
|
|
|
if len(objVersions) == 0 {
|
|
|
|
return nil, nil
|
|
|
|
}
|
|
|
|
|
2024-01-17 14:26:02 +00:00
|
|
|
var resVersion []*data.ExtendedNodeVersion
|
|
|
|
var resDelMarkVersions []*data.ExtendedNodeVersion
|
2024-01-17 12:14:26 +00:00
|
|
|
|
|
|
|
for _, version := range objVersions {
|
2024-01-17 14:26:02 +00:00
|
|
|
if version.NodeVersion.IsDeleteMarker {
|
2024-01-17 12:14:26 +00:00
|
|
|
resDelMarkVersions = append(resDelMarkVersions, version)
|
|
|
|
} else {
|
|
|
|
resVersion = append(resVersion, version)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return resVersion, resDelMarkVersions
|
|
|
|
}
|