dkirillov/frostfs-s3-gw
1
0
Fork 0
forked from TrueCloudLab/frostfs-s3-gw
Code Issues Pull requests Projects Releases Packages Wiki Activity

[#165] Refactor list versions

Browse source 
Signed-off-by: Denis Kirillov <d.kirillov@yadro.com>
This commit is contained in:
Denis Kirillov 2024-01-21 01:13:35 +03:00
parent 6d52f46012
commit 2d7973b3f1
5 changed files with 169 additions and 313 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
api/layer/layer.go
View file

@ -809,16 +809,15 @@ func (n *layer) ResolveBucket(ctx context.Context, name string) (cid.ID, error)
} }
func (n *layer) DeleteBucket(ctx context.Context, p *DeleteBucketParams) error { func (n *layer) DeleteBucket(ctx context.Context, p *DeleteBucketParams) error {
res, err := n.ListObjectVersions(ctx, &ListObjectVersionsParams{ res, _, err := n.getAllObjectsVersions(ctx, commonVersionsListingParams{
BktInfo: p.BktInfo, BktInfo: p.BktInfo,
MaxKeys: 1, MaxKeys: 1,
}) })
//todo fix ^
if err != nil { if err != nil {
return err return err
} }
if len(res.DeleteMarker) != 0 || len(res.Version) != 0 { if len(res) != 0 {
return errors.GetAPIError(errors.ErrBucketNotEmpty) return errors.GetAPIError(errors.ErrBucketNotEmpty)
} }

266
api/layer/listing.go
View file

@ -73,29 +73,42 @@ type (
VersionIDMarker string VersionIDMarker string
} }
allObjectListingParams struct { commonVersionsListingParams struct {
BktInfo *data.BucketInfo BktInfo *data.BucketInfo
Delimiter string Delimiter string
Prefix string Prefix string
MaxKeys int MaxKeys int
Marker string Marker string
Bookmark string Bookmark string
VersionAPi string
} }
commonLatestVersionsListingParams struct {
commonVersionsListingParams
ListType ListType
}
)
type ListType int
const (
ListObjectsV1Type ListType = iota + 1
ListObjectsV2Type ListType = iota + 1
) )
// ListObjectsV1 returns objects in a bucket for requests of Version 1. // ListObjectsV1 returns objects in a bucket for requests of Version 1.
func (n *layer) ListObjectsV1(ctx context.Context, p *ListObjectsParamsV1) (*ListObjectsInfoV1, error) { func (n *layer) ListObjectsV1(ctx context.Context, p *ListObjectsParamsV1) (*ListObjectsInfoV1, error) {
var result ListObjectsInfoV1 var result ListObjectsInfoV1
prm := allObjectListingParams{ prm := commonLatestVersionsListingParams{
BktInfo: p.BktInfo, commonVersionsListingParams: commonVersionsListingParams{
Delimiter: p.Delimiter, BktInfo: p.BktInfo,
Prefix: p.Prefix, Delimiter: p.Delimiter,
MaxKeys: p.MaxKeys, Prefix: p.Prefix,
Marker: p.Marker, MaxKeys: p.MaxKeys,
Bookmark: p.Marker, Marker: p.Marker,
VersionAPi: "v1", Bookmark: p.Marker,
},
ListType: ListObjectsV1Type,
} }
objects, next, err := n.getLatestObjectsVersions(ctx, prm) objects, next, err := n.getLatestObjectsVersions(ctx, prm)
@ -117,14 +130,16 @@ func (n *layer) ListObjectsV1(ctx context.Context, p *ListObjectsParamsV1) (*Lis
func (n *layer) ListObjectsV2(ctx context.Context, p *ListObjectsParamsV2) (*ListObjectsInfoV2, error) { func (n *layer) ListObjectsV2(ctx context.Context, p *ListObjectsParamsV2) (*ListObjectsInfoV2, error) {
var result ListObjectsInfoV2 var result ListObjectsInfoV2
prm := allObjectListingParams{ prm := commonLatestVersionsListingParams{
BktInfo: p.BktInfo, commonVersionsListingParams: commonVersionsListingParams{
Delimiter: p.Delimiter, BktInfo: p.BktInfo,
Prefix: p.Prefix, Delimiter: p.Delimiter,
MaxKeys: p.MaxKeys, Prefix: p.Prefix,
Marker: p.StartAfter, MaxKeys: p.MaxKeys,
Bookmark: p.ContinuationToken, Marker: p.StartAfter,
VersionAPi: "v2", Bookmark: p.ContinuationToken,
},
ListType: ListObjectsV2Type,
} }
objects, next, err := n.getLatestObjectsVersions(ctx, prm) objects, next, err := n.getLatestObjectsVersions(ctx, prm)
@ -143,14 +158,13 @@ func (n *layer) ListObjectsV2(ctx context.Context, p *ListObjectsParamsV2) (*Lis
} }
func (n *layer) ListObjectVersions(ctx context.Context, p *ListObjectVersionsParams) (*ListObjectVersionsInfo, error) { func (n *layer) ListObjectVersions(ctx context.Context, p *ListObjectVersionsParams) (*ListObjectVersionsInfo, error) {
prm := allObjectListingParams{ prm := commonVersionsListingParams{
BktInfo: p.BktInfo, BktInfo: p.BktInfo,
Delimiter: p.Delimiter, Delimiter: p.Delimiter,
Prefix: p.Prefix, Prefix: p.Prefix,
MaxKeys: p.MaxKeys, MaxKeys: p.MaxKeys,
Marker: p.KeyMarker, Marker: p.KeyMarker,
Bookmark: p.VersionIDMarker, Bookmark: p.VersionIDMarker,
VersionAPi: "vs",
} }
objects, isTruncated, err := n.getAllObjectsVersions(ctx, prm) objects, isTruncated, err := n.getAllObjectsVersions(ctx, prm)
@ -174,44 +188,24 @@ func (n *layer) ListObjectVersions(ctx context.Context, p *ListObjectVersionsPar
return res, nil return res, nil
} }
func (n *layer) getLatestObjectsVersions(ctx context.Context, p allObjectListingParams) (objects []*data.NodeVersion, next *data.NodeVersion, err error) { func (n *layer) getLatestObjectsVersions(ctx context.Context, p commonLatestVersionsListingParams) (objects []*data.NodeVersion, next *data.NodeVersion, err error) {
if p.MaxKeys == 0 { if p.MaxKeys == 0 {
return nil, nil, nil return nil, nil, nil
} }
owner := n.BearerOwner(ctx) session, err := n.getListLatestVersionsSession(ctx, p)
cacheKey := cache.CreateListSessionCacheKey(p.BktInfo.CID, p.Prefix, p.Bookmark) // todo redo for listv1 if err != nil {
session := n.cache.GetListSession(owner, cacheKey) return nil, nil, err
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.BktInfo, p.Prefix)
if err != nil {
return nil, nil, err
}
} }
poolCtx, cancel := context.WithCancel(ctx) generator, errorCh := nodesGeneratorStream(ctx, p.commonVersionsListingParams, session)
defer cancel() objOutCh, err := n.initWorkerPoolStream(ctx, 2, p.commonVersionsListingParams, generator)
generator, errorCh := nodesGeneratorStream(poolCtx, p, session)
objOutCh, err := n.initWorkerPoolStream(poolCtx, 2, p, generator)
if err != nil { if err != nil {
return nil, nil, fmt.Errorf("failed to init worker pool: %w", err) return nil, nil, fmt.Errorf("failed to init worker pool: %w", err)
} }
objects = make([]*data.NodeVersion, 0, p.MaxKeys+1) objects = make([]*data.NodeVersion, 0, p.MaxKeys+1)
objects = append(objects, session.Next...) objects = append(objects, session.Next...)
for obj := range objOutCh { for obj := range objOutCh {
objects = append(objects, obj) objects = append(objects, obj)
} }
@ -222,27 +216,19 @@ func (n *layer) getLatestObjectsVersions(ctx context.Context, p allObjectListing
if len(objects) > p.MaxKeys { if len(objects) > p.MaxKeys {
next = objects[p.MaxKeys] next = objects[p.MaxKeys]
n.putListLatestVersionsSession(ctx, p, session, objects)
objects = objects[:p.MaxKeys] objects = objects[:p.MaxKeys]
} }
if next != nil {
session.Next = []*data.NodeVersion{next}
if p.VersionAPi == "v1" {
n.cache.PutListSession(owner, cache.CreateListSessionCacheKey(p.BktInfo.CID, p.Prefix, objects[len(objects)-1].FilePath), session)
} else {
n.cache.PutListSession(owner, cache.CreateListSessionCacheKey(p.BktInfo.CID, p.Prefix, next.OID.EncodeToString()), session)
}
}
return return
} }
func (n *layer) getAllObjectsVersions(ctx context.Context, p allObjectListingParams) ([]*data.ExtendedNodeVersion, bool, error) { func (n *layer) getAllObjectsVersions(ctx context.Context, p commonVersionsListingParams) ([]*data.ExtendedNodeVersion, bool, error) {
if p.MaxKeys == 0 { if p.MaxKeys == 0 {
return nil, false, nil return nil, false, nil
} }
session, err := n.getListVersionsSession(ctx, p) session, err := n.getListAllVersionsSession(ctx, p)
if err != nil { if err != nil {
return nil, false, err return nil, false, err
} }
@ -253,8 +239,26 @@ func (n *layer) getAllObjectsVersions(ctx context.Context, p allObjectListingPar
return nil, false, err return nil, false, err
} }
allObjects := handleGeneratedVersions(objOutCh, p, session)
if err = <-errorCh; err != nil {
return nil, false, fmt.Errorf("failed to get next object from stream: %w", err)
}
var isTruncated bool
if len(allObjects) > p.MaxKeys {
isTruncated = true
n.putListAllVersionsSession(ctx, p, session, allObjects)
allObjects = allObjects[:p.MaxKeys]
}
return allObjects, isTruncated, nil
}
func handleGeneratedVersions(objOutCh <-chan *data.ExtendedNodeVersion, p commonVersionsListingParams, session *data.ListSession) []*data.ExtendedNodeVersion {
var lastName string var lastName string
groupedVersions := make([][]*data.ExtendedNodeVersion, 0, p.MaxKeys) var listRowStartIndex int
allObjects := make([]*data.ExtendedNodeVersion, 0, p.MaxKeys)
for eoi := range objOutCh { for eoi := range objOutCh {
name := eoi.NodeVersion.FilePath name := eoi.NodeVersion.FilePath
@ -264,61 +268,53 @@ func (n *layer) getAllObjectsVersions(ctx context.Context, p allObjectListingPar
} }
if lastName != name { if lastName != name {
groupedVersions = append(groupedVersions, []*data.ExtendedNodeVersion{eoi}) formVersionsListRow(allObjects, listRowStartIndex, session)
listRowStartIndex = len(allObjects)
allObjects = append(allObjects, eoi)
} else if dirName == "" { } else if dirName == "" {
groupedVersions[len(groupedVersions)-1] = append(groupedVersions[len(groupedVersions)-1], eoi) allObjects = append(allObjects, eoi)
} }
lastName = name lastName = name
} }
if err = <-errorCh; err != nil { formVersionsListRow(allObjects, listRowStartIndex, session)
return nil, false, fmt.Errorf("failed to get next object from stream: %w", err)
}
allObjects := make([]*data.ExtendedNodeVersion, 0, p.MaxKeys) return allObjects
for _, versions := range groupedVersions {
sort.Slice(versions, func(i, j int) bool {
return versions[j].NodeVersion.Timestamp < versions[i].NodeVersion.Timestamp // sort in reverse order
})
for i, version := range versions {
version.IsLatest = i == 0 && (session.Next == nil || session.Next[0].FilePath != versions[0].NodeVersion.FilePath)
allObjects = append(allObjects, version)
}
}
var isTruncated bool
if len(allObjects) > p.MaxKeys {
isTruncated = true
session.Next = make([]*data.NodeVersion, len(allObjects)-p.MaxKeys+1)
session.Next[0] = allObjects[p.MaxKeys-1].NodeVersion
for i, node := range allObjects[p.MaxKeys:] {
session.Next[i+1] = node.NodeVersion
}
session.Acquired.Store(false)
n.cache.PutListSession(n.BearerOwner(ctx), cache.CreateListSessionCacheKey(p.BktInfo.CID, p.Prefix, session.Next[0].OID.EncodeToString()), session)
allObjects = allObjects[:p.MaxKeys]
}
return allObjects, isTruncated, nil
} }
func (n *layer) getListVersionsSession(ctx context.Context, p allObjectListingParams) (*data.ListSession, error) { func formVersionsListRow(objects []*data.ExtendedNodeVersion, rowStartIndex int, session *data.ListSession) {
if len(objects) == 0 {
return
}
prevVersions := objects[rowStartIndex:]
sort.Slice(prevVersions, func(i, j int) bool {
return prevVersions[j].NodeVersion.Timestamp < prevVersions[i].NodeVersion.Timestamp // sort in reverse order to have last added first
})
objects[rowStartIndex].IsLatest = len(session.Next) == 0 || session.Next[0].FilePath != objects[rowStartIndex].NodeVersion.FilePath
}
func (n *layer) getListLatestVersionsSession(ctx context.Context, p commonLatestVersionsListingParams) (*data.ListSession, error) {
return n.getListVersionsSession(ctx, p.commonVersionsListingParams, true)
}
func (n *layer) getListAllVersionsSession(ctx context.Context, p commonVersionsListingParams) (*data.ListSession, error) {
return n.getListVersionsSession(ctx, p, false)
}
func (n *layer) getListVersionsSession(ctx context.Context, p commonVersionsListingParams, latestOnly bool) (*data.ListSession, error) {
owner := n.BearerOwner(ctx) owner := n.BearerOwner(ctx)
cacheKey := cache.CreateListSessionCacheKey(p.BktInfo.CID, p.Prefix, p.Bookmark) cacheKey := cache.CreateListSessionCacheKey(p.BktInfo.CID, p.Prefix, p.Bookmark)
session := n.cache.GetListSession(owner, cacheKey) session := n.cache.GetListSession(owner, cacheKey)
if session == nil { if session == nil {
return n.initNewListVersionsSession(ctx, p) return n.initNewVersionsByPrefixSession(ctx, p, latestOnly)
} }
if session.Acquired.Swap(true) { if session.Acquired.Swap(true) {
return n.initNewListVersionsSession(ctx, p) return n.initNewVersionsByPrefixSession(ctx, p, latestOnly)
} }
// after reading next object from stream in session // after reading next object from stream in session
@ -328,7 +324,7 @@ func (n *layer) getListVersionsSession(ctx context.Context, p allObjectListingPa
return session, nil return session, nil
} }
func (n *layer) initNewListVersionsSession(ctx context.Context, p allObjectListingParams) (session *data.ListSession, err error) { func (n *layer) initNewVersionsByPrefixSession(ctx context.Context, p commonVersionsListingParams, latestOnly bool) (session *data.ListSession, err error) {
session = &data.ListSession{NamesMap: make(map[string]struct{})} session = &data.ListSession{NamesMap: make(map[string]struct{})}
session.Context, session.Cancel = context.WithCancel(context.Background()) session.Context, session.Cancel = context.WithCancel(context.Background())
@ -336,7 +332,7 @@ func (n *layer) initNewListVersionsSession(ctx context.Context, p allObjectListi
session.Context = middleware.SetBoxData(session.Context, bd) session.Context = middleware.SetBoxData(session.Context, bd)
} }
session.Stream, err = n.treeService.GetAllVersionsByPrefixStream(session.Context, p.BktInfo, p.Prefix) session.Stream, err = n.treeService.InitVersionsByPrefixStream(session.Context, p.BktInfo, p.Prefix, latestOnly)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -344,7 +340,45 @@ func (n *layer) initNewListVersionsSession(ctx context.Context, p allObjectListi
return session, nil return session, nil
} }
func nodesGeneratorStream(ctx context.Context, p allObjectListingParams, stream *data.ListSession) (<-chan *data.NodeVersion, <-chan error) { func (n *layer) putListLatestVersionsSession(ctx context.Context, p commonLatestVersionsListingParams, session *data.ListSession, allObjects []*data.NodeVersion) {
if len(allObjects) <= p.MaxKeys {
return
}
var cacheKey cache.ListSessionKey
switch p.ListType {
case ListObjectsV1Type:
cacheKey = cache.CreateListSessionCacheKey(p.BktInfo.CID, p.Prefix, allObjects[p.MaxKeys-1].FilePath)
case ListObjectsV2Type:
cacheKey = cache.CreateListSessionCacheKey(p.BktInfo.CID, p.Prefix, allObjects[p.MaxKeys].OID.EncodeToString())
default:
// should never happen
panic("invalid list type")
}
session.Acquired.Store(false)
session.Next = []*data.NodeVersion{allObjects[p.MaxKeys]}
n.cache.PutListSession(n.BearerOwner(ctx), cacheKey, session)
}
func (n *layer) putListAllVersionsSession(ctx context.Context, p commonVersionsListingParams, session *data.ListSession, allObjects []*data.ExtendedNodeVersion) {
if len(allObjects) <= p.MaxKeys {
return
}
session.Acquired.Store(false)
session.Next = make([]*data.NodeVersion, len(allObjects)-p.MaxKeys+1)
session.Next[0] = allObjects[p.MaxKeys-1].NodeVersion
for i, node := range allObjects[p.MaxKeys:] {
session.Next[i+1] = node.NodeVersion
}
cacheKey := cache.CreateListSessionCacheKey(p.BktInfo.CID, p.Prefix, session.Next[0].OID.EncodeToString())
n.cache.PutListSession(n.BearerOwner(ctx), cacheKey, session)
}
func nodesGeneratorStream(ctx context.Context, p commonVersionsListingParams, stream *data.ListSession) (<-chan *data.NodeVersion, <-chan error) {
nodeCh := make(chan *data.NodeVersion, 1000) nodeCh := make(chan *data.NodeVersion, 1000)
errCh := make(chan error, 1) errCh := make(chan error, 1)
//existed := make(map[string]struct{}, p.MaxKeys) // to squash the same directories //existed := make(map[string]struct{}, p.MaxKeys) // to squash the same directories
@ -395,7 +429,7 @@ func nodesGeneratorStream(ctx context.Context, p allObjectListingParams, stream
return nodeCh, errCh return nodeCh, errCh
} }
func nodesGeneratorVersions(ctx context.Context, p allObjectListingParams, stream *data.ListSession) (<-chan *data.NodeVersion, <-chan error) { func nodesGeneratorVersions(ctx context.Context, p commonVersionsListingParams, stream *data.ListSession) (<-chan *data.NodeVersion, <-chan error) {
nodeCh := make(chan *data.NodeVersion, 1000) nodeCh := make(chan *data.NodeVersion, 1000)
errCh := make(chan error, 1) errCh := make(chan error, 1)
existed := stream.NamesMap existed := stream.NamesMap
@ -448,7 +482,7 @@ func nodesGeneratorVersions(ctx context.Context, p allObjectListingParams, strea
return nodeCh, errCh return nodeCh, errCh
} }
func (n *layer) initWorkerPoolStream(ctx context.Context, size int, p allObjectListingParams, input <-chan *data.NodeVersion) (<-chan *data.NodeVersion, error) { func (n *layer) initWorkerPoolStream(ctx context.Context, size int, p commonVersionsListingParams, input <-chan *data.NodeVersion) (<-chan *data.NodeVersion, error) {
reqLog := n.reqLogger(ctx) reqLog := n.reqLogger(ctx)
pool, err := ants.NewPool(size, ants.WithLogger(&logWrapper{reqLog})) pool, err := ants.NewPool(size, ants.WithLogger(&logWrapper{reqLog}))
if err != nil { if err != nil {
@ -513,7 +547,7 @@ func (n *layer) initWorkerPoolStream(ctx context.Context, size int, p allObjectL
return objCh, nil return objCh, nil
} }
func (n *layer) initWorkerPoolVersions(ctx context.Context, size int, p allObjectListingParams, input <-chan *data.NodeVersion) (<-chan *data.ExtendedNodeVersion, error) { func (n *layer) initWorkerPoolVersions(ctx context.Context, size int, p commonVersionsListingParams, input <-chan *data.NodeVersion) (<-chan *data.ExtendedNodeVersion, error) {
reqLog := n.reqLogger(ctx) reqLog := n.reqLogger(ctx)
pool, err := ants.NewPool(size, ants.WithLogger(&logWrapper{reqLog})) pool, err := ants.NewPool(size, ants.WithLogger(&logWrapper{reqLog}))
if err != nil { if err != nil {
@ -576,7 +610,7 @@ func (n *layer) initWorkerPoolVersions(ctx context.Context, size int, p allObjec
return objCh, nil return objCh, nil
} }
func shouldSkip(node *data.NodeVersion, p allObjectListingParams, existed map[string]struct{}) bool { func shouldSkip(node *data.NodeVersion, p commonVersionsListingParams, existed map[string]struct{}) bool {
if node.IsDeleteMarker { if node.IsDeleteMarker {
return true return true
} }
@ -606,7 +640,7 @@ func shouldSkip(node *data.NodeVersion, p allObjectListingParams, existed map[st
return false return false
} }
func shouldSkipVersions(node *data.NodeVersion, p allObjectListingParams, existed map[string]struct{}) bool { func shouldSkipVersions(node *data.NodeVersion, p commonVersionsListingParams, existed map[string]struct{}) bool {
filePath := node.FilePath filePath := node.FilePath
if dirName := tryDirectoryName(node, p.Prefix, p.Delimiter); len(dirName) != 0 { if dirName := tryDirectoryName(node, p.Prefix, p.Delimiter); len(dirName) != 0 {
filePath = dirName filePath = dirName

58
api/layer/tree_mock.go
View file

@ -11,12 +11,12 @@ import (
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id" oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
) )
type LatestVersionsByPrefixStreamMock struct { type VersionsByPrefixStreamMock struct {
result []*data.NodeVersion result []*data.NodeVersion
offset int offset int
} }
func (s *LatestVersionsByPrefixStreamMock) Next(context.Context) (*data.NodeVersion, error) { func (s *VersionsByPrefixStreamMock) Next(context.Context) (*data.NodeVersion, error) {
if s.offset > len(s.result)-1 { if s.offset > len(s.result)-1 {
return nil, io.EOF return nil, io.EOF
} }
@ -187,7 +187,7 @@ func (t *TreeServiceMock) GetLatestVersion(_ context.Context, bktInfo *data.Buck
return nil, ErrNodeNotFound return nil, ErrNodeNotFound
} }
func (t *TreeServiceMock) GetLatestVersionsByPrefix(_ context.Context, bktInfo *data.BucketInfo, prefix string) ([]*data.NodeVersion, error) { func (t *TreeServiceMock) InitVersionsByPrefixStream(_ context.Context, bktInfo *data.BucketInfo, prefix string, latestOnly bool) (data.VersionsStream, error) {
cnrVersionsMap, ok := t.versions[bktInfo.CID.EncodeToString()] cnrVersionsMap, ok := t.versions[bktInfo.CID.EncodeToString()]
if !ok { if !ok {
return nil, ErrNodeNotFound return nil, ErrNodeNotFound
@ -200,6 +200,11 @@ func (t *TreeServiceMock) GetLatestVersionsByPrefix(_ context.Context, bktInfo *
continue continue
} }
if !latestOnly {
result = append(result, versions...)
continue
}
sort.Slice(versions, func(i, j int) bool { sort.Slice(versions, func(i, j int) bool {
return versions[i].ID < versions[j].ID return versions[i].ID < versions[j].ID
}) })
@ -209,52 +214,7 @@ func (t *TreeServiceMock) GetLatestVersionsByPrefix(_ context.Context, bktInfo *
} }
} }
return result, nil return &VersionsByPrefixStreamMock{
}
func (t *TreeServiceMock) GetLatestVersionsByPrefixStream(ctx context.Context, bktInfo *data.BucketInfo, prefix string) (data.VersionsStream, error) {
cnrVersionsMap, ok := t.versions[bktInfo.CID.EncodeToString()]
if !ok {
return nil, ErrNodeNotFound
}
var result []*data.NodeVersion
for key, versions := range cnrVersionsMap {
if !strings.HasPrefix(key, prefix) {
continue
}
sort.Slice(versions, func(i, j int) bool {
return versions[i].ID < versions[j].ID
})
if len(versions) != 0 {
result = append(result, versions[len(versions)-1])
}
}
return &LatestVersionsByPrefixStreamMock{
result: result,
}, nil
}
func (t *TreeServiceMock) GetAllVersionsByPrefixStream(ctx context.Context, bktInfo *data.BucketInfo, prefix string) (data.VersionsStream, error) {
cnrVersionsMap, ok := t.versions[bktInfo.CID.EncodeToString()]
if !ok {
return nil, ErrNodeNotFound
}
var result []*data.NodeVersion
for key, versions := range cnrVersionsMap {
if !strings.HasPrefix(key, prefix) {
continue
}
result = append(result, versions...)
}
return &LatestVersionsByPrefixStreamMock{
result: result, result: result,
}, nil }, nil
} }

5
api/layer/tree_service.go
View file

@ -54,10 +54,7 @@ type TreeService interface {
GetVersions(ctx context.Context, bktInfo *data.BucketInfo, objectName string) ([]*data.NodeVersion, error) GetVersions(ctx context.Context, bktInfo *data.BucketInfo, objectName string) ([]*data.NodeVersion, error)
GetLatestVersion(ctx context.Context, bktInfo *data.BucketInfo, objectName string) (*data.NodeVersion, error) GetLatestVersion(ctx context.Context, bktInfo *data.BucketInfo, objectName string) (*data.NodeVersion, error)
GetLatestVersionsByPrefix(ctx context.Context, bktInfo *data.BucketInfo, prefix string) ([]*data.NodeVersion, error) InitVersionsByPrefixStream(ctx context.Context, bktInfo *data.BucketInfo, prefix string, latestOnly bool) (data.VersionsStream, error)
GetLatestVersionsByPrefixStream(ctx context.Context, bktInfo *data.BucketInfo, prefix string) (data.VersionsStream, error)
GetAllVersionsByPrefix(ctx context.Context, bktInfo *data.BucketInfo, prefix string) ([]*data.NodeVersion, error)
GetAllVersionsByPrefixStream(ctx context.Context, bktInfo *data.BucketInfo, prefix string) (data.VersionsStream, error)
GetUnversioned(ctx context.Context, bktInfo *data.BucketInfo, objectName string) (*data.NodeVersion, error) GetUnversioned(ctx context.Context, bktInfo *data.BucketInfo, objectName string) (*data.NodeVersion, error)
AddVersion(ctx context.Context, bktInfo *data.BucketInfo, newVersion *data.NodeVersion) (uint64, error) AddVersion(ctx context.Context, bktInfo *data.BucketInfo, newVersion *data.NodeVersion) (uint64, error)
RemoveVersion(ctx context.Context, bktInfo *data.BucketInfo, nodeID uint64) error RemoveVersion(ctx context.Context, bktInfo *data.BucketInfo, nodeID uint64) error

148
pkg/service/tree/tree.go
View file

@ -638,10 +638,6 @@ func pathFromName(objectName string) []string {
return strings.Split(objectName, separator) return strings.Split(objectName, separator)
} }
func (c *Tree) GetLatestVersionsByPrefix(ctx context.Context, bktInfo *data.BucketInfo, prefix string) ([]*data.NodeVersion, error) {
return c.getVersionsByPrefix(ctx, bktInfo, prefix, true)
}
type DummySubTreeStream struct { type DummySubTreeStream struct {
data NodeResponse data NodeResponse
read bool read bool
@ -656,7 +652,7 @@ func (s *DummySubTreeStream) Next() (NodeResponse, error) {
return s.data, nil return s.data, nil
} }
type LatestVersionsByPrefixStreamImpl struct { type VersionsByPrefixStreamImpl struct {
ctx context.Context ctx context.Context
rootID uint64 rootID uint64
intermediateRootID uint64 intermediateRootID uint64
@ -672,7 +668,8 @@ type LatestVersionsByPrefixStreamImpl struct {
currentLatest *data.NodeVersion currentLatest *data.NodeVersion
} }
func (s *LatestVersionsByPrefixStreamImpl) Next(ctx context.Context) (*data.NodeVersion, error) { // Next todo remove recursion
func (s *VersionsByPrefixStreamImpl) Next(ctx context.Context) (*data.NodeVersion, error) {
if s.ended { if s.ended {
return nil, io.EOF return nil, io.EOF
} }
@ -771,16 +768,16 @@ func (s *LatestVersionsByPrefixStreamImpl) Next(ctx context.Context) (*data.Node
return nodeVersion, nil return nodeVersion, nil
} }
func (c *Tree) GetLatestVersionsByPrefixStream(ctx context.Context, bktInfo *data.BucketInfo, prefix string) (data.VersionsStream, error) { func (c *Tree) InitVersionsByPrefixStream(ctx context.Context, bktInfo *data.BucketInfo, prefix string, latestOnly bool) (data.VersionsStream, error) {
mainStream, tailPrefix, rootID, err := c.getSubTreeByPrefixMainStream(ctx, bktInfo, versionTree, prefix) mainStream, tailPrefix, rootID, err := c.getSubTreeByPrefixMainStream(ctx, bktInfo, versionTree, prefix)
if err != nil { if err != nil {
if errors.Is(err, io.EOF) { if errors.Is(err, io.EOF) {
return &LatestVersionsByPrefixStreamImpl{ended: true}, nil return &VersionsByPrefixStreamImpl{ended: true}, nil
} }
return nil, err return nil, err
} }
return &LatestVersionsByPrefixStreamImpl{ return &VersionsByPrefixStreamImpl{
ctx: ctx, ctx: ctx,
namesMap: map[uint64]string{}, namesMap: map[uint64]string{},
rootID: rootID, rootID: rootID,
@ -789,7 +786,7 @@ func (c *Tree) GetLatestVersionsByPrefixStream(ctx context.Context, bktInfo *dat
mainStream: mainStream, mainStream: mainStream,
headPrefix: strings.TrimSuffix(prefix, tailPrefix), headPrefix: strings.TrimSuffix(prefix, tailPrefix),
tailPrefix: tailPrefix, tailPrefix: tailPrefix,
latestOnly: true, latestOnly: latestOnly,
}, nil }, nil
} }
@ -931,73 +928,6 @@ func isIntermediate(node NodeResponse) bool {
return node.GetMeta()[0].GetKey() == FileNameKey return node.GetMeta()[0].GetKey() == FileNameKey
} }
func (c *Tree) getSubTreeVersionsOld(ctx context.Context, bktInfo *data.BucketInfo, node NodeResponse, parentFilePath string, latestOnly bool) ([]*data.NodeVersion, error) {
return c.getSubTreeVersions(ctx, bktInfo, node, parentFilePath, latestOnly, false)
}
func (c *Tree) getSubTreeVersions(ctx context.Context, bktInfo *data.BucketInfo, node NodeResponse, parentFilePath string, latestOnly, skipLeafs bool) ([]*data.NodeVersion, error) {
var err error
subTree := []NodeResponse{node}
if !skipLeafs || isIntermediate(node) {
subTree, err = c.service.GetSubTree(ctx, bktInfo, versionTree, node.GetNodeID(), maxGetSubTreeDepth)
if err != nil {
return nil, err
}
}
var parentPrefix string
if parentFilePath != "" { // The root of subTree can also have a parent
parentPrefix = strings.TrimSuffix(parentFilePath, separator) + separator // To avoid 'foo//bar'
}
var emptyOID oid.ID
var filepath string
namesMap := make(map[uint64]string, len(subTree))
versions := make(map[string][]*data.NodeVersion, len(subTree))
for i, node := range subTree {
treeNode, fileName, err := parseTreeNode(node)
if err != nil {
continue
}
if i != 0 {
if filepath, err = formFilePath(node, fileName, namesMap); err != nil {
return nil, fmt.Errorf("invalid node order: %w", err)
}
} else {
filepath = parentPrefix + fileName
namesMap[treeNode.ID] = filepath
}
if treeNode.ObjID.Equals(emptyOID) { // The node can be intermediate but we still want to update namesMap
continue
}
key := formLatestNodeKey(node.GetParentID(), fileName)
versionNodes, ok := versions[key]
if !ok {
versionNodes = []*data.NodeVersion{newNodeVersionFromTreeNode(filepath, treeNode)}
} else if !latestOnly {
versionNodes = append(versionNodes, newNodeVersionFromTreeNode(filepath, treeNode))
} else if versionNodes[0].Timestamp <= treeNode.TimeStamp {
versionNodes[0] = newNodeVersionFromTreeNode(filepath, treeNode)
}
versions[key] = versionNodes
}
result := make([]*data.NodeVersion, 0, len(versions)) // consider use len(subTree)
for _, version := range versions {
if latestOnly && version[0].IsDeleteMarker {
continue
}
result = append(result, version...)
}
return result, nil
}
func formFilePath(node NodeResponse, fileName string, namesMap map[uint64]string) (string, error) { func formFilePath(node NodeResponse, fileName string, namesMap map[uint64]string) (string, error) {
parentPath, ok := namesMap[node.GetParentID()] parentPath, ok := namesMap[node.GetParentID()]
if !ok { if !ok {
@ -1028,70 +958,6 @@ func formLatestNodeKey(parentID uint64, fileName string) string {
return strconv.FormatUint(parentID, 10) + "." + fileName return strconv.FormatUint(parentID, 10) + "." + fileName
} }
func (c *Tree) GetAllVersionsByPrefix(ctx context.Context, bktInfo *data.BucketInfo, prefix string) ([]*data.NodeVersion, error) {
return c.getVersionsByPrefixOld(ctx, bktInfo, prefix, false)
}
func (c *Tree) GetAllVersionsByPrefixStream(ctx context.Context, bktInfo *data.BucketInfo, prefix string) (data.VersionsStream, error) {
//return c.getVersionsByPrefixOld(ctx, bktInfo, prefix, false)
mainStream, tailPrefix, rootID, err := c.getSubTreeByPrefixMainStream(ctx, bktInfo, versionTree, prefix)
if err != nil {
if errors.Is(err, io.EOF) {
return &LatestVersionsByPrefixStreamImpl{ended: true}, nil
}
return nil, err
}
return &LatestVersionsByPrefixStreamImpl{
ctx: ctx,
namesMap: map[uint64]string{},
rootID: rootID,
service: c.service,
bktInfo: bktInfo,
mainStream: mainStream,
headPrefix: strings.TrimSuffix(prefix, tailPrefix),
tailPrefix: tailPrefix,
}, nil
}
func (c *Tree) getVersionsByPrefix(ctx context.Context, bktInfo *data.BucketInfo, prefix string, latestOnly bool) ([]*data.NodeVersion, error) {
prefixNodes, headPrefix, err := c.getSubTreeByPrefix(ctx, bktInfo, versionTree, prefix, latestOnly)
if err != nil {
return nil, err
}
var result []*data.NodeVersion
for _, node := range prefixNodes {
versions, err := c.getSubTreeVersions(ctx, bktInfo, node, headPrefix, latestOnly, true)
if err != nil {
return nil, err
}
result = append(result, versions...)
}
return result, nil
}
func (c *Tree) getVersionsByPrefixOld(ctx context.Context, bktInfo *data.BucketInfo, prefix string, latestOnly bool) ([]*data.NodeVersion, error) {
prefixNodes, headPrefix, err := c.getSubTreeByPrefix(ctx, bktInfo, versionTree, prefix, latestOnly)
if err != nil {
return nil, err
}
var result []*data.NodeVersion
for _, node := range prefixNodes {
versions, err := c.getSubTreeVersionsOld(ctx, bktInfo, node, headPrefix, latestOnly)
if err != nil {
return nil, err
}
result = append(result, versions...)
}
return result, nil
}
func (c *Tree) GetUnversioned(ctx context.Context, bktInfo *data.BucketInfo, filepath string) (*data.NodeVersion, error) { func (c *Tree) GetUnversioned(ctx context.Context, bktInfo *data.BucketInfo, filepath string) (*data.NodeVersion, error) {
return c.getUnversioned(ctx, bktInfo, versionTree, filepath) return c.getUnversioned(ctx, bktInfo, versionTree, filepath)
} }