[#420] Using tree service to list object versions

Signed-off-by: Denis Kirillov <denis@nspcc.ru>
This commit is contained in:
Denis Kirillov 2022-05-20 18:02:00 +03:00 committed by Alex Vanin
parent 55c38e73e6
commit 9c74cca9af
11 changed files with 374 additions and 135 deletions

View file

@ -6,6 +6,7 @@ import (
"time" "time"
"github.com/bluele/gcache" "github.com/bluele/gcache"
"github.com/nspcc-dev/neofs-s3-gw/api/data"
cid "github.com/nspcc-dev/neofs-sdk-go/container/id" cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
oid "github.com/nspcc-dev/neofs-sdk-go/object/id" oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
"go.uber.org/zap" "go.uber.org/zap"
@ -78,6 +79,21 @@ func (l *ObjectsListCache) Get(key ObjectsListKey) []oid.ID {
return result return result
} }
// GetVersions returns a list of ObjectInfo.
func (l *ObjectsListCache) GetVersions(key ObjectsListKey) []*data.NodeVersion {
entry, err := l.cache.Get(key)
if err != nil {
return nil
}
result, ok := entry.([]*data.NodeVersion)
if !ok {
return nil
}
return result
}
// Put puts a list of objects to cache. // Put puts a list of objects to cache.
func (l *ObjectsListCache) Put(key ObjectsListKey, oids []oid.ID) error { func (l *ObjectsListCache) Put(key ObjectsListKey, oids []oid.ID) error {
if len(oids) == 0 { if len(oids) == 0 {
@ -87,6 +103,15 @@ func (l *ObjectsListCache) Put(key ObjectsListKey, oids []oid.ID) error {
return l.cache.Set(key, oids) return l.cache.Set(key, oids)
} }
// PutVersions puts a list of object versions to cache.
func (l *ObjectsListCache) PutVersions(key ObjectsListKey, versions []*data.NodeVersion) error {
if len(versions) == 0 {
return fmt.Errorf("list versions is empty, cid: %s, prefix: %s", key.cid, key.prefix)
}
return l.cache.Set(key, versions)
}
// CleanCacheEntriesContainingObject deletes entries containing specified object. // CleanCacheEntriesContainingObject deletes entries containing specified object.
func (l *ObjectsListCache) CleanCacheEntriesContainingObject(objectName string, cnr cid.ID) { func (l *ObjectsListCache) CleanCacheEntriesContainingObject(objectName string, cnr cid.ID) {
cidStr := cnr.EncodeToString() cidStr := cnr.EncodeToString()

View file

@ -28,9 +28,10 @@ type (
// ObjectInfo holds S3 object data. // ObjectInfo holds S3 object data.
ObjectInfo struct { ObjectInfo struct {
ID oid.ID ID oid.ID
CID cid.ID CID cid.ID
IsDir bool IsDir bool
IsDeleteMarker bool
Bucket string Bucket string
Name string Name string

37
api/data/tree.go Normal file
View file

@ -0,0 +1,37 @@
package data
import (
"time"
oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
"github.com/nspcc-dev/neofs-sdk-go/user"
)
// NodeVersion represent node from tree service.
type NodeVersion struct {
BaseNodeVersion
DeleteMarker *DeleteMarkerInfo
IsUnversioned bool
}
// DeleteMarkerInfo is used to save object info if node in the tree service is delete marker.
// We need this information because the "delete marker" object is no longer stored in NeoFS.
type DeleteMarkerInfo struct {
FilePath string
Created time.Time
Owner user.ID
}
// ExtendedObjectInfo contains additional node info to be able to sort versions by timestamp.
type ExtendedObjectInfo struct {
ObjectInfo *ObjectInfo
NodeVersion *NodeVersion
}
// BaseNodeVersion is minimal node info from tree service.
// Basically used for "system" object.
type BaseNodeVersion struct {
ID uint64
OID oid.ID
Timestamp uint64
}

View file

@ -8,6 +8,7 @@ import (
"io" "io"
"net/url" "net/url"
"strings" "strings"
"time"
"github.com/nats-io/nats.go" "github.com/nats-io/nats.go"
"github.com/nspcc-dev/neo-go/pkg/crypto/keys" "github.com/nspcc-dev/neo-go/pkg/crypto/keys"
@ -562,12 +563,16 @@ func (n *layer) deleteObject(ctx context.Context, bkt *data.BucketInfo, settings
} }
obj.DeleteMarkVersion = unversionedObjectVersionID obj.DeleteMarkVersion = unversionedObjectVersionID
newVersion := &NodeVersion{ newVersion := &data.NodeVersion{
BaseNodeVersion: BaseNodeVersion{ BaseNodeVersion: data.BaseNodeVersion{
OID: *randOID, OID: *randOID,
}, },
IsDeleteMarker: true, DeleteMarker: &data.DeleteMarkerInfo{
IsUnversioned: true, FilePath: obj.Name,
Created: time.Now(),
Owner: n.Owner(ctx),
},
IsUnversioned: true,
} }
if len(obj.VersionID) == 0 && settings.VersioningEnabled { if len(obj.VersionID) == 0 && settings.VersioningEnabled {
@ -595,14 +600,14 @@ func (n *layer) deleteObject(ctx context.Context, bkt *data.BucketInfo, settings
return obj return obj
} }
func (n *layer) removeVersionIfFound(ctx context.Context, bkt *data.BucketInfo, versions []*NodeVersion, obj *VersionedObject) (string, error) { func (n *layer) removeVersionIfFound(ctx context.Context, bkt *data.BucketInfo, versions []*data.NodeVersion, obj *VersionedObject) (string, error) {
for _, version := range versions { for _, version := range versions {
if version.OID.EncodeToString() != obj.VersionID { if version.OID.EncodeToString() != obj.VersionID {
continue continue
} }
var deleteMarkVersion string var deleteMarkVersion string
if version.IsDeleteMarker { if version.DeleteMarker != nil {
deleteMarkVersion = obj.VersionID deleteMarkVersion = obj.VersionID
} }

View file

@ -171,7 +171,7 @@ func (n *layer) PutObject(ctx context.Context, p *PutObjectParams) (*data.Object
own := n.Owner(ctx) own := n.Owner(ctx)
versioningEnabled := n.isVersioningEnabled(ctx, p.BktInfo) versioningEnabled := n.isVersioningEnabled(ctx, p.BktInfo)
newVersion := &NodeVersion{IsUnversioned: !versioningEnabled} newVersion := &data.NodeVersion{IsUnversioned: !versioningEnabled}
r := p.Reader r := p.Reader
if r != nil { if r != nil {
@ -283,7 +283,7 @@ func (n *layer) headLastVersionIfNotDeleted(ctx context.Context, bkt *data.Bucke
return nil, err return nil, err
} }
if node.IsDeleteMarker { if node.DeleteMarker != nil {
return nil, apiErrors.GetAPIError(apiErrors.ErrNoSuchKey) return nil, apiErrors.GetAPIError(apiErrors.ErrNoSuchKey)
} }
@ -365,7 +365,7 @@ func (n *layer) headVersion(ctx context.Context, bkt *data.BucketInfo, p *HeadOb
return nil, fmt.Errorf("couldn't get versions: %w", err) return nil, fmt.Errorf("couldn't get versions: %w", err)
} }
var foundVersion *NodeVersion var foundVersion *data.NodeVersion
for _, version := range versions { for _, version := range versions {
if version.OID.EncodeToString() == p.VersionID { if version.OID.EncodeToString() == p.VersionID {
foundVersion = version foundVersion = version
@ -554,41 +554,56 @@ func (n *layer) getLatestObjectsVersions(ctx context.Context, bkt *data.BucketIn
return objects, nil return objects, nil
} }
func (n *layer) getAllObjectsVersions(ctx context.Context, bkt *data.BucketInfo, prefix, delimiter string) (map[string]*objectVersions, error) { func (n *layer) getAllObjectsVersions(ctx context.Context, bkt *data.BucketInfo, prefix, delimiter string) (map[string][]*data.ExtendedObjectInfo, error) {
var err error var err error
cacheKey := cache.CreateObjectsListCacheKey(&bkt.CID, prefix, false) cacheKey := cache.CreateObjectsListCacheKey(&bkt.CID, prefix, false)
ids := n.listsCache.Get(cacheKey) nodeVersions := n.listsCache.GetVersions(cacheKey)
if ids == nil { if nodeVersions == nil {
ids, err = n.objectSearch(ctx, &findParams{bkt: bkt, prefix: prefix}) nodeVersions, err = n.treeService.GetAllVersionsByPrefix(ctx, &bkt.CID, prefix)
if err != nil { if err != nil {
return nil, err return nil, err
} }
if err = n.listsCache.Put(cacheKey, ids); err != nil { if err = n.listsCache.PutVersions(cacheKey, nodeVersions); err != nil {
n.log.Error("couldn't cache list of objects", zap.Error(err)) n.log.Error("couldn't cache list of objects", zap.Error(err))
} }
} }
versions := make(map[string]*objectVersions, len(ids)/2) versions := make(map[string][]*data.ExtendedObjectInfo, len(nodeVersions))
for i := 0; i < len(ids); i++ { for _, nodeVersion := range nodeVersions {
obj := n.objectFromObjectsCacheOrNeoFS(ctx, bkt, ids[i]) oi := &data.ObjectInfo{}
if obj == nil {
continue if nodeVersion.DeleteMarker != nil { // delete marker does not match any object in NeoFS
} oi.ID = nodeVersion.OID
if oi := objectInfoFromMeta(bkt, obj, prefix, delimiter); oi != nil { oi.Name = nodeVersion.DeleteMarker.FilePath
if isSystem(oi) { oi.Owner = nodeVersion.DeleteMarker.Owner
oi.Created = nodeVersion.DeleteMarker.Created
oi.IsDeleteMarker = true
} else {
obj := n.objectFromObjectsCacheOrNeoFS(ctx, bkt, nodeVersion.OID)
if obj == nil {
continue continue
} }
oi = objectInfoFromMeta(bkt, obj, prefix, delimiter)
objVersions, ok := versions[oi.Name] if oi == nil {
if !ok { continue
objVersions = newObjectVersions(oi.Name)
} }
objVersions.appendVersion(oi)
versions[oi.Name] = objVersions
} }
eoi := &data.ExtendedObjectInfo{
ObjectInfo: oi,
NodeVersion: nodeVersion,
}
objVersions, ok := versions[oi.Name]
if !ok {
objVersions = []*data.ExtendedObjectInfo{eoi}
} else if !oi.IsDir {
objVersions = append(objVersions, eoi)
}
versions[oi.Name] = objVersions
} }
return versions, nil return versions, nil

View file

@ -122,7 +122,7 @@ func (n *layer) putSystemObjectIntoNeoFS(ctx context.Context, p *PutSystemObject
return nil, err return nil, err
} }
newVersion := &BaseNodeVersion{OID: *id} newVersion := &data.BaseNodeVersion{OID: *id}
if err = n.treeService.AddSystemVersion(ctx, &p.BktInfo.CID, p.ObjName, newVersion); err != nil { if err = n.treeService.AddSystemVersion(ctx, &p.BktInfo.CID, p.ObjName, newVersion); err != nil {
return nil, fmt.Errorf("couldn't add new verion to tree service: %w", err) return nil, fmt.Errorf("couldn't add new verion to tree service: %w", err)
} }

View file

@ -30,28 +30,18 @@ type TreeService interface {
// DeleteBucketCORS removes a node from a system tree and returns objID which must be deleted in NeoFS // DeleteBucketCORS removes a node from a system tree and returns objID which must be deleted in NeoFS
DeleteBucketCORS(ctx context.Context, cnrID *cid.ID) (*oid.ID, error) DeleteBucketCORS(ctx context.Context, cnrID *cid.ID) (*oid.ID, error)
GetVersions(ctx context.Context, cnrID *cid.ID, objectName string) ([]*NodeVersion, error) GetVersions(ctx context.Context, cnrID *cid.ID, objectName string) ([]*data.NodeVersion, error)
GetLatestVersion(ctx context.Context, cnrID *cid.ID, objectName string) (*NodeVersion, error) GetLatestVersion(ctx context.Context, cnrID *cid.ID, objectName string) (*data.NodeVersion, error)
GetLatestVersionsByPrefix(ctx context.Context, cnrID *cid.ID, prefix string) ([]oid.ID, error) GetLatestVersionsByPrefix(ctx context.Context, cnrID *cid.ID, prefix string) ([]oid.ID, error)
GetUnversioned(ctx context.Context, cnrID *cid.ID, objectName string) (*NodeVersion, error) GetAllVersionsByPrefix(ctx context.Context, cnrID *cid.ID, prefix string) ([]*data.NodeVersion, error)
AddVersion(ctx context.Context, cnrID *cid.ID, objectName string, newVersion *NodeVersion) error GetUnversioned(ctx context.Context, cnrID *cid.ID, objectName string) (*data.NodeVersion, error)
AddVersion(ctx context.Context, cnrID *cid.ID, objectName string, newVersion *data.NodeVersion) error
RemoveVersion(ctx context.Context, cnrID *cid.ID, nodeID uint64) error RemoveVersion(ctx context.Context, cnrID *cid.ID, nodeID uint64) error
AddSystemVersion(ctx context.Context, cnrID *cid.ID, objectName string, newVersion *BaseNodeVersion) error AddSystemVersion(ctx context.Context, cnrID *cid.ID, objectName string, newVersion *data.BaseNodeVersion) error
GetSystemVersion(ctx context.Context, cnrID *cid.ID, objectName string) (*BaseNodeVersion, error) GetSystemVersion(ctx context.Context, cnrID *cid.ID, objectName string) (*data.BaseNodeVersion, error)
RemoveSystemVersion(ctx context.Context, cnrID *cid.ID, nodeID uint64) error RemoveSystemVersion(ctx context.Context, cnrID *cid.ID, nodeID uint64) error
} }
type NodeVersion struct {
BaseNodeVersion
IsDeleteMarker bool
IsUnversioned bool
}
type BaseNodeVersion struct {
ID uint64
OID oid.ID
}
// ErrNodeNotFound is returned from Tree service in case of not found error. // ErrNodeNotFound is returned from Tree service in case of not found error.
var ErrNodeNotFound = errors.New("not found") var ErrNodeNotFound = errors.New("not found")

View file

@ -267,7 +267,6 @@ func (n *layer) ListObjectVersions(ctx context.Context, p *ListObjectVersionsPar
var ( var (
allObjects = make([]*data.ObjectInfo, 0, p.MaxKeys) allObjects = make([]*data.ObjectInfo, 0, p.MaxKeys)
res = &ListObjectVersionsInfo{} res = &ListObjectVersionsInfo{}
reverse = true
) )
versions, err := n.getAllObjectsVersions(ctx, p.BktInfo, p.Prefix, p.Delimiter) versions, err := n.getAllObjectsVersions(ctx, p.BktInfo, p.Prefix, p.Delimiter)
@ -282,7 +281,14 @@ func (n *layer) ListObjectVersions(ctx context.Context, p *ListObjectVersionsPar
sort.Strings(sortedNames) sort.Strings(sortedNames)
for _, name := range sortedNames { for _, name := range sortedNames {
allObjects = append(allObjects, versions[name].getFiltered(reverse)...) 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 _, version := range sortedVersions {
allObjects = append(allObjects, version.ObjectInfo)
}
} }
for i, obj := range allObjects { for i, obj := range allObjects {
@ -325,7 +331,7 @@ func triageVersions(objVersions []*ObjectVersionInfo) ([]*ObjectVersionInfo, []*
var resDelMarkVersions []*ObjectVersionInfo var resDelMarkVersions []*ObjectVersionInfo
for _, version := range objVersions { for _, version := range objVersions {
if version.Object.Headers[VersionsDeleteMarkAttr] == DelMarkFullObject { if version.Object.IsDeleteMarker {
resDelMarkVersions = append(resDelMarkVersions, version) resDelMarkVersions = append(resDelMarkVersions, version)
} else { } else {
resVersion = append(resVersion, version) resVersion = append(resVersion, version)

View file

@ -11,6 +11,7 @@ import (
"github.com/nspcc-dev/neofs-s3-gw/api" "github.com/nspcc-dev/neofs-s3-gw/api"
"github.com/nspcc-dev/neofs-s3-gw/api/data" "github.com/nspcc-dev/neofs-s3-gw/api/data"
"github.com/nspcc-dev/neofs-s3-gw/creds/accessbox" "github.com/nspcc-dev/neofs-s3-gw/creds/accessbox"
treetest "github.com/nspcc-dev/neofs-s3-gw/internal/neofstest/tree"
bearertest "github.com/nspcc-dev/neofs-sdk-go/bearer/test" bearertest "github.com/nspcc-dev/neofs-sdk-go/bearer/test"
"github.com/nspcc-dev/neofs-sdk-go/object" "github.com/nspcc-dev/neofs-sdk-go/object"
oid "github.com/nspcc-dev/neofs-sdk-go/object/id" oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
@ -164,8 +165,9 @@ func prepareContext(t *testing.T, cachesConfig ...*CachesConfig) *testContext {
} }
layerCfg := &Config{ layerCfg := &Config{
Caches: config, Caches: config,
AnonKey: AnonymousKey{Key: key}, AnonKey: AnonymousKey{Key: key},
TreeService: treetest.NewTreeService(),
} }
return &testContext{ return &testContext{
@ -196,9 +198,8 @@ func TestSimpleVersioning(t *testing.T) {
obj1Content2 := []byte("content obj1 v2") obj1Content2 := []byte("content obj1 v2")
obj1v2 := tc.putObject(obj1Content2) obj1v2 := tc.putObject(obj1Content2)
objv2, buffer2 := tc.getObject(tc.obj, "", false) _, buffer2 := tc.getObject(tc.obj, "", false)
require.Equal(t, obj1Content2, buffer2) require.Equal(t, obj1Content2, buffer2)
require.Contains(t, objv2.Headers[versionsAddAttr], obj1v1.ID.EncodeToString())
_, buffer1 := tc.getObject(tc.obj, obj1v1.ID.EncodeToString(), false) _, buffer1 := tc.getObject(tc.obj, obj1v1.ID.EncodeToString(), false)
require.Equal(t, obj1Content1, buffer1) require.Equal(t, obj1Content1, buffer1)
@ -215,9 +216,8 @@ func TestSimpleNoVersioning(t *testing.T) {
obj1Content2 := []byte("content obj1 v2") obj1Content2 := []byte("content obj1 v2")
obj1v2 := tc.putObject(obj1Content2) obj1v2 := tc.putObject(obj1Content2)
objv2, buffer2 := tc.getObject(tc.obj, "", false) _, buffer2 := tc.getObject(tc.obj, "", false)
require.Equal(t, obj1Content2, buffer2) require.Equal(t, obj1Content2, buffer2)
require.Contains(t, objv2.Headers[versionsDelAttr], obj1v1.ID.EncodeToString())
tc.getObject(tc.obj, obj1v1.ID.EncodeToString(), true) tc.getObject(tc.obj, obj1v1.ID.EncodeToString(), true)
tc.checkListObjects(obj1v2.ID) tc.checkListObjects(obj1v2.ID)

View file

@ -7,6 +7,7 @@ import (
"io" "io"
"strconv" "strconv"
"strings" "strings"
"time"
"github.com/nspcc-dev/neo-go/pkg/crypto/keys" "github.com/nspcc-dev/neo-go/pkg/crypto/keys"
"github.com/nspcc-dev/neofs-s3-gw/api" "github.com/nspcc-dev/neofs-s3-gw/api"
@ -16,6 +17,7 @@ import (
"github.com/nspcc-dev/neofs-s3-gw/internal/neofs/services/tree" "github.com/nspcc-dev/neofs-s3-gw/internal/neofs/services/tree"
cid "github.com/nspcc-dev/neofs-sdk-go/container/id" cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
oid "github.com/nspcc-dev/neofs-sdk-go/object/id" oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
"github.com/nspcc-dev/neofs-sdk-go/user"
"google.golang.org/grpc" "google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/credentials/insecure"
) )
@ -42,7 +44,12 @@ const (
fileNameKV = "FileName" fileNameKV = "FileName"
systemNameKV = "SystemName" systemNameKV = "SystemName"
isUnversionedKV = "IsUnversioned" isUnversionedKV = "IsUnversioned"
isDeleteMarkerKV = "IdDeleteMarker"
// keys for delete marker nodes
isDeleteMarkerKV = "IdDeleteMarker"
filePathKV = "FilePath"
ownerKV = "Owner"
createdKV = "Created"
settingsFileName = "bucket-settings" settingsFileName = "bucket-settings"
notifConfFileName = "bucket-notifications" notifConfFileName = "bucket-notifications"
@ -111,23 +118,50 @@ func (n *TreeNode) Get(key string) (string, bool) {
return value, ok return value, ok
} }
func newNodeVersion(node NodeResponse) (*layer.NodeVersion, error) { func newNodeVersion(node NodeResponse) (*data.NodeVersion, error) {
treeNode, err := newTreeNode(node) treeNode, err := newTreeNode(node)
if err != nil { if err != nil {
return nil, fmt.Errorf("invalid tree node: %w", err) return nil, fmt.Errorf("invalid tree node: %w", err)
} }
return newNodeVersionFromTreeNode(treeNode), nil
}
func newNodeVersionFromTreeNode(treeNode *TreeNode) *data.NodeVersion {
_, isUnversioned := treeNode.Get(isUnversionedKV) _, isUnversioned := treeNode.Get(isUnversionedKV)
_, isDeleteMarker := treeNode.Get(isDeleteMarkerKV) _, isDeleteMarker := treeNode.Get(isDeleteMarkerKV)
return &layer.NodeVersion{ version := &data.NodeVersion{
BaseNodeVersion: layer.BaseNodeVersion{ BaseNodeVersion: data.BaseNodeVersion{
ID: treeNode.ID, ID: treeNode.ID,
OID: treeNode.ObjID, OID: treeNode.ObjID,
Timestamp: treeNode.TimeStamp,
}, },
IsUnversioned: isUnversioned, IsUnversioned: isUnversioned,
IsDeleteMarker: isDeleteMarker, }
}, nil
if isDeleteMarker {
filePath, _ := treeNode.Get(filePathKV)
var created time.Time
if createdStr, ok := treeNode.Get(createdKV); ok {
if utcMilli, err := strconv.ParseInt(createdStr, 10, 64); err == nil {
created = time.UnixMilli(utcMilli)
}
}
var owner user.ID
if ownerStr, ok := treeNode.Get(ownerKV); ok {
_ = owner.DecodeString(ownerStr)
}
version.DeleteMarker = &data.DeleteMarkerInfo{
FilePath: filePath,
Created: created,
Owner: owner,
}
}
return version
} }
func (c *TreeClient) GetSettingsNode(ctx context.Context, cnrID *cid.ID) (*data.BucketSettings, error) { func (c *TreeClient) GetSettingsNode(ctx context.Context, cnrID *cid.ID) (*data.BucketSettings, error) {
@ -240,11 +274,11 @@ func (c *TreeClient) DeleteBucketCORS(ctx context.Context, cnrID *cid.ID) (*oid.
return nil, nil return nil, nil
} }
func (c *TreeClient) GetVersions(ctx context.Context, cnrID *cid.ID, filepath string) ([]*layer.NodeVersion, error) { func (c *TreeClient) GetVersions(ctx context.Context, cnrID *cid.ID, filepath string) ([]*data.NodeVersion, error) {
return c.getVersions(ctx, cnrID, versionTree, filepath, false) return c.getVersions(ctx, cnrID, versionTree, filepath, false)
} }
func (c *TreeClient) GetLatestVersion(ctx context.Context, cnrID *cid.ID, objectName string) (*layer.NodeVersion, error) { func (c *TreeClient) GetLatestVersion(ctx context.Context, cnrID *cid.ID, objectName string) (*data.NodeVersion, error) {
meta := []string{oidKV, isUnversionedKV, isDeleteMarkerKV} meta := []string{oidKV, isUnversionedKV, isDeleteMarkerKV}
path := pathFromName(objectName) path := pathFromName(objectName)
@ -266,20 +300,14 @@ func (c *TreeClient) GetLatestVersionsByPrefix(ctx context.Context, cnrID *cid.I
tailPrefix := path[len(path)-1] tailPrefix := path[len(path)-1]
if len(path) > 1 { if len(path) > 1 {
meta := []string{fileNameKV} var err error
rootID, err = c.getPrefixNodeID(ctx, cnrID, path[:len(path)-1])
nodes, err := c.getNodes(ctx, cnrID, versionTree, fileNameKV, path[:len(path)-1], meta, true)
if err != nil { if err != nil {
if errors.Is(err, layer.ErrNodeNotFound) {
return nil, nil
}
return nil, err return nil, err
} }
if len(nodes) == 0 {
return nil, nil
}
if len(nodes) != 1 {
return nil, layer.ErrNodeNotFound
}
rootID = nodes[0].NodeId
} }
subTree, err := c.getSubTree(ctx, cnrID, versionTree, rootID, 1) subTree, err := c.getSubTree(ctx, cnrID, versionTree, rootID, 1)
@ -289,18 +317,46 @@ func (c *TreeClient) GetLatestVersionsByPrefix(ctx context.Context, cnrID *cid.I
var result []oid.ID var result []oid.ID
for _, node := range subTree { for _, node := range subTree {
if node.GetNodeId() != 0 && hasPrefix(node, tailPrefix) { if node.GetNodeId() != rootID && hasPrefix(node, tailPrefix) {
latestNodes, err := c.getSubTreeLatestVersions(ctx, cnrID, node.GetNodeId()) latestNodes, err := c.getSubTreeVersions(ctx, cnrID, node.GetNodeId(), true)
if err != nil { if err != nil {
return nil, err return nil, err
} }
result = append(result, latestNodes...)
for _, latest := range latestNodes {
result = append(result, latest.OID)
}
} }
} }
return result, nil return result, nil
} }
func (c *TreeClient) getPrefixNodeID(ctx context.Context, cnrID *cid.ID, prefixPath []string) (uint64, error) {
meta := []string{fileNameKV, oidKV}
nodes, err := c.getNodes(ctx, cnrID, versionTree, fileNameKV, prefixPath, meta, false)
if err != nil {
return 0, err
}
var intermediateNodes []uint64
for _, node := range nodes {
if !hasOID(node) {
intermediateNodes = append(intermediateNodes, node.GetNodeId())
}
}
if len(intermediateNodes) == 0 {
return 0, layer.ErrNodeNotFound
}
if len(intermediateNodes) > 1 {
return 0, fmt.Errorf("found more than one intermediate nodes")
}
return intermediateNodes[0], nil
}
func hasPrefix(node *tree.GetSubTreeResponse_Body, prefix string) bool { func hasPrefix(node *tree.GetSubTreeResponse_Body, prefix string) bool {
for _, kv := range node.GetMeta() { for _, kv := range node.GetMeta() {
if kv.GetKey() == fileNameKV { if kv.GetKey() == fileNameKV {
@ -311,7 +367,17 @@ func hasPrefix(node *tree.GetSubTreeResponse_Body, prefix string) bool {
return false return false
} }
func (c *TreeClient) getSubTreeLatestVersions(ctx context.Context, cnrID *cid.ID, nodeID uint64) ([]oid.ID, error) { func hasOID(node *tree.GetNodeByPathResponse_Info) bool {
for _, kv := range node.GetMeta() {
if kv.GetKey() == oidKV {
return true
}
}
return false
}
func (c *TreeClient) getSubTreeVersions(ctx context.Context, cnrID *cid.ID, nodeID uint64, latestOnly bool) ([]*data.NodeVersion, error) {
subTree, err := c.getSubTree(ctx, cnrID, versionTree, nodeID, maxGetSubTreeDepth) subTree, err := c.getSubTree(ctx, cnrID, versionTree, nodeID, maxGetSubTreeDepth)
if err != nil { if err != nil {
return nil, err return nil, err
@ -319,10 +385,10 @@ func (c *TreeClient) getSubTreeLatestVersions(ctx context.Context, cnrID *cid.ID
var emptyOID oid.ID var emptyOID oid.ID
latestVersions := make(map[string]*TreeNode, len(subTree)) versions := make(map[string][]*data.NodeVersion, len(subTree))
for _, node := range subTree { for _, node := range subTree {
treeNode, err := newTreeNode(node) treeNode, err := newTreeNode(node)
if err != nil || treeNode.ObjID.Equals(emptyOID) { // invalid OID attribute if err != nil || treeNode.ObjID.Equals(emptyOID) { // invalid or empty OID attribute
continue continue
} }
fileName, ok := treeNode.Get(fileNameKV) fileName, ok := treeNode.Get(fileNameKV)
@ -331,18 +397,24 @@ func (c *TreeClient) getSubTreeLatestVersions(ctx context.Context, cnrID *cid.ID
} }
key := formLatestNodeKey(node.GetParentId(), fileName) key := formLatestNodeKey(node.GetParentId(), fileName)
latest, ok := latestVersions[key] versionNodes, ok := versions[key]
if !ok || latest.TimeStamp <= treeNode.TimeStamp { // todo also compare oid if !ok {
latestVersions[key] = treeNode versionNodes = []*data.NodeVersion{newNodeVersionFromTreeNode(treeNode)}
} else if !latestOnly {
versionNodes = append(versionNodes, newNodeVersionFromTreeNode(treeNode))
} else if versionNodes[0].Timestamp <= treeNode.TimeStamp {
versionNodes[0] = newNodeVersionFromTreeNode(treeNode)
} }
versions[key] = versionNodes
} }
result := make([]oid.ID, 0, len(latestVersions)) result := make([]*data.NodeVersion, 0, len(versions)) // consider use len(subTree)
for _, treeNode := range latestVersions { for _, version := range versions {
if _, ok := treeNode.Get(isDeleteMarkerKV); ok { if latestOnly && version[0].DeleteMarker != nil {
continue continue
} }
result = append(result, treeNode.ObjID) result = append(result, version...)
} }
return result, nil return result, nil
@ -352,7 +424,42 @@ func formLatestNodeKey(parentID uint64, fileName string) string {
return strconv.FormatUint(parentID, 10) + fileName return strconv.FormatUint(parentID, 10) + fileName
} }
func (c *TreeClient) GetSystemVersion(ctx context.Context, cnrID *cid.ID, objectName string) (*layer.BaseNodeVersion, error) { func (c *TreeClient) GetAllVersionsByPrefix(ctx context.Context, cnrID *cid.ID, prefix string) ([]*data.NodeVersion, error) {
var rootID uint64
path := strings.Split(prefix, separator)
tailPrefix := path[len(path)-1]
if len(path) > 1 {
var err error
rootID, err = c.getPrefixNodeID(ctx, cnrID, path[:len(path)-1])
if err != nil {
if errors.Is(err, layer.ErrNodeNotFound) {
return nil, nil
}
return nil, err
}
}
subTree, err := c.getSubTree(ctx, cnrID, versionTree, rootID, 1)
if err != nil {
return nil, err
}
var result []*data.NodeVersion
for _, node := range subTree {
if node.GetNodeId() != rootID && hasPrefix(node, tailPrefix) {
versions, err := c.getSubTreeVersions(ctx, cnrID, node.GetNodeId(), false)
if err != nil {
return nil, err
}
result = append(result, versions...)
}
}
return result, nil
}
func (c *TreeClient) GetSystemVersion(ctx context.Context, cnrID *cid.ID, objectName string) (*data.BaseNodeVersion, error) {
meta := []string{oidKV} meta := []string{oidKV}
path := pathFromName(objectName) path := pathFromName(objectName)
@ -363,7 +470,7 @@ func (c *TreeClient) GetSystemVersion(ctx context.Context, cnrID *cid.ID, object
return &node.BaseNodeVersion, nil return &node.BaseNodeVersion, nil
} }
func (c *TreeClient) getLatestVersion(ctx context.Context, cnrID *cid.ID, treeID, attrPath string, path, meta []string) (*layer.NodeVersion, error) { func (c *TreeClient) getLatestVersion(ctx context.Context, cnrID *cid.ID, treeID, attrPath string, path, meta []string) (*data.NodeVersion, error) {
nodes, err := c.getNodes(ctx, cnrID, treeID, attrPath, path, meta, true) nodes, err := c.getNodes(ctx, cnrID, treeID, attrPath, path, meta, true)
if err != nil { if err != nil {
if strings.Contains(err.Error(), "not found") { if strings.Contains(err.Error(), "not found") {
@ -379,11 +486,11 @@ func (c *TreeClient) getLatestVersion(ctx context.Context, cnrID *cid.ID, treeID
return newNodeVersion(nodes[0]) return newNodeVersion(nodes[0])
} }
func (c *TreeClient) GetUnversioned(ctx context.Context, cnrID *cid.ID, filepath string) (*layer.NodeVersion, error) { func (c *TreeClient) GetUnversioned(ctx context.Context, cnrID *cid.ID, filepath string) (*data.NodeVersion, error) {
return c.getUnversioned(ctx, cnrID, versionTree, filepath) return c.getUnversioned(ctx, cnrID, versionTree, filepath)
} }
func (c *TreeClient) getUnversioned(ctx context.Context, cnrID *cid.ID, treeID, filepath string) (*layer.NodeVersion, error) { func (c *TreeClient) getUnversioned(ctx context.Context, cnrID *cid.ID, treeID, filepath string) (*data.NodeVersion, error) {
nodes, err := c.getVersions(ctx, cnrID, treeID, filepath, true) nodes, err := c.getVersions(ctx, cnrID, treeID, filepath, true)
if err != nil { if err != nil {
return nil, err return nil, err
@ -400,12 +507,12 @@ func (c *TreeClient) getUnversioned(ctx context.Context, cnrID *cid.ID, treeID,
return nodes[0], nil return nodes[0], nil
} }
func (c *TreeClient) AddVersion(ctx context.Context, cnrID *cid.ID, filepath string, version *layer.NodeVersion) error { func (c *TreeClient) AddVersion(ctx context.Context, cnrID *cid.ID, filepath string, version *data.NodeVersion) error {
return c.addVersion(ctx, cnrID, versionTree, fileNameKV, filepath, version) return c.addVersion(ctx, cnrID, versionTree, fileNameKV, filepath, version)
} }
func (c *TreeClient) AddSystemVersion(ctx context.Context, cnrID *cid.ID, filepath string, version *layer.BaseNodeVersion) error { func (c *TreeClient) AddSystemVersion(ctx context.Context, cnrID *cid.ID, filepath string, version *data.BaseNodeVersion) error {
newVersion := &layer.NodeVersion{ newVersion := &data.NodeVersion{
BaseNodeVersion: *version, BaseNodeVersion: *version,
IsUnversioned: true, IsUnversioned: true,
} }
@ -428,15 +535,18 @@ func (c *TreeClient) Close() error {
return nil return nil
} }
func (c *TreeClient) addVersion(ctx context.Context, cnrID *cid.ID, treeID, attrPath, filepath string, version *layer.NodeVersion) error { func (c *TreeClient) addVersion(ctx context.Context, cnrID *cid.ID, treeID, attrPath, filepath string, version *data.NodeVersion) error {
path := pathFromName(filepath) path := pathFromName(filepath)
meta := map[string]string{ meta := map[string]string{
oidKV: version.OID.EncodeToString(), oidKV: version.OID.EncodeToString(),
attrPath: path[len(path)-1], attrPath: path[len(path)-1],
} }
if version.IsDeleteMarker { if version.DeleteMarker != nil {
meta[isDeleteMarkerKV] = "true" meta[isDeleteMarkerKV] = "true"
meta[filePathKV] = version.DeleteMarker.FilePath
meta[ownerKV] = version.DeleteMarker.Owner.EncodeToString()
meta[createdKV] = strconv.FormatInt(version.DeleteMarker.Created.UTC().UnixMilli(), 10)
} }
if version.IsUnversioned { if version.IsUnversioned {
@ -460,7 +570,7 @@ func (c *TreeClient) addVersion(ctx context.Context, cnrID *cid.ID, treeID, attr
return c.addNodeByPath(ctx, cnrID, treeID, path[:len(path)-1], meta) return c.addNodeByPath(ctx, cnrID, treeID, path[:len(path)-1], meta)
} }
func (c *TreeClient) getVersions(ctx context.Context, cnrID *cid.ID, treeID, filepath string, onlyUnversioned bool) ([]*layer.NodeVersion, error) { func (c *TreeClient) getVersions(ctx context.Context, cnrID *cid.ID, treeID, filepath string, onlyUnversioned bool) ([]*data.NodeVersion, error) {
keysToReturn := []string{oidKV, isUnversionedKV, isDeleteMarkerKV} keysToReturn := []string{oidKV, isUnversionedKV, isDeleteMarkerKV}
path := pathFromName(filepath) path := pathFromName(filepath)
nodes, err := c.getNodes(ctx, cnrID, treeID, fileNameKV, path, keysToReturn, false) nodes, err := c.getNodes(ctx, cnrID, treeID, fileNameKV, path, keysToReturn, false)
@ -471,7 +581,7 @@ func (c *TreeClient) getVersions(ctx context.Context, cnrID *cid.ID, treeID, fil
return nil, fmt.Errorf("couldn't get nodes: %w", err) return nil, fmt.Errorf("couldn't get nodes: %w", err)
} }
result := make([]*layer.NodeVersion, 0, len(nodes)) result := make([]*data.NodeVersion, 0, len(nodes))
for _, node := range nodes { for _, node := range nodes {
nodeVersion, err := newNodeVersion(node) nodeVersion, err := newNodeVersion(node)
if err != nil { if err != nil {

View file

@ -2,25 +2,28 @@ package tree
import ( import (
"context" "context"
"errors"
"sort" "sort"
"strings"
"github.com/nspcc-dev/neofs-s3-gw/api/data" "github.com/nspcc-dev/neofs-s3-gw/api/data"
"github.com/nspcc-dev/neofs-s3-gw/api/layer"
cid "github.com/nspcc-dev/neofs-sdk-go/container/id" cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
oid "github.com/nspcc-dev/neofs-sdk-go/object/id" oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
) )
type TreeServiceMock struct { type TreeServiceMock struct {
settings map[string]*data.BucketSettings settings map[string]*data.BucketSettings
versions map[string]map[string][]*layer.NodeVersion versions map[string]map[string][]*data.NodeVersion
system map[string]map[string]*layer.BaseNodeVersion system map[string]map[string]*data.BaseNodeVersion
} }
var ErrNodeNotFound = errors.New("not found")
func NewTreeService() *TreeServiceMock { func NewTreeService() *TreeServiceMock {
return &TreeServiceMock{ return &TreeServiceMock{
settings: make(map[string]*data.BucketSettings), settings: make(map[string]*data.BucketSettings),
versions: make(map[string]map[string][]*layer.NodeVersion), versions: make(map[string]map[string][]*data.NodeVersion),
system: make(map[string]map[string]*layer.BaseNodeVersion), system: make(map[string]map[string]*data.BaseNodeVersion),
} }
} }
@ -32,7 +35,7 @@ func (t *TreeServiceMock) PutSettingsNode(_ context.Context, id *cid.ID, setting
func (t *TreeServiceMock) GetSettingsNode(_ context.Context, id *cid.ID) (*data.BucketSettings, error) { func (t *TreeServiceMock) GetSettingsNode(_ context.Context, id *cid.ID) (*data.BucketSettings, error) {
settings, ok := t.settings[id.EncodeToString()] settings, ok := t.settings[id.EncodeToString()]
if !ok { if !ok {
return nil, layer.ErrNodeNotFound return nil, ErrNodeNotFound
} }
return settings, nil return settings, nil
@ -58,19 +61,29 @@ func (t *TreeServiceMock) DeleteBucketCORS(ctx context.Context, cnrID *cid.ID) (
panic("implement me") panic("implement me")
} }
func (t *TreeServiceMock) GetVersions(ctx context.Context, cnrID *cid.ID, objectName string) ([]*layer.NodeVersion, error) { func (t *TreeServiceMock) GetVersions(_ context.Context, cnrID *cid.ID, objectName string) ([]*data.NodeVersion, error) {
panic("implement me")
}
func (t *TreeServiceMock) GetLatestVersion(ctx context.Context, cnrID *cid.ID, objectName string) (*layer.NodeVersion, error) {
cnrVersionsMap, ok := t.versions[cnrID.EncodeToString()] cnrVersionsMap, ok := t.versions[cnrID.EncodeToString()]
if !ok { if !ok {
return nil, layer.ErrNodeNotFound return nil, ErrNodeNotFound
} }
versions, ok := cnrVersionsMap[objectName] versions, ok := cnrVersionsMap[objectName]
if !ok { if !ok {
return nil, layer.ErrNodeNotFound return nil, ErrNodeNotFound
}
return versions, nil
}
func (t *TreeServiceMock) GetLatestVersion(ctx context.Context, cnrID *cid.ID, objectName string) (*data.NodeVersion, error) {
cnrVersionsMap, ok := t.versions[cnrID.EncodeToString()]
if !ok {
return nil, ErrNodeNotFound
}
versions, ok := cnrVersionsMap[objectName]
if !ok {
return nil, ErrNodeNotFound
} }
sort.Slice(versions, func(i, j int) bool { sort.Slice(versions, func(i, j int) bool {
@ -81,21 +94,42 @@ func (t *TreeServiceMock) GetLatestVersion(ctx context.Context, cnrID *cid.ID, o
return versions[len(versions)-1], nil return versions[len(versions)-1], nil
} }
return nil, layer.ErrNodeNotFound return nil, ErrNodeNotFound
} }
func (t *TreeServiceMock) GetLatestVersionsByPrefix(ctx context.Context, cnrID *cid.ID, prefix string) ([]oid.ID, error) { func (t *TreeServiceMock) GetLatestVersionsByPrefix(_ context.Context, cnrID *cid.ID, prefix string) ([]oid.ID, error) {
panic("implement me")
}
func (t *TreeServiceMock) GetUnversioned(ctx context.Context, cnrID *cid.ID, objectName string) (*layer.NodeVersion, error) {
panic("implement me")
}
func (t *TreeServiceMock) AddVersion(_ context.Context, cnrID *cid.ID, objectName string, newVersion *layer.NodeVersion) error {
cnrVersionsMap, ok := t.versions[cnrID.EncodeToString()] cnrVersionsMap, ok := t.versions[cnrID.EncodeToString()]
if !ok { if !ok {
t.versions[cnrID.EncodeToString()] = map[string][]*layer.NodeVersion{ return nil, ErrNodeNotFound
}
var result []oid.ID
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].OID)
}
}
return result, nil
}
func (t *TreeServiceMock) GetUnversioned(ctx context.Context, cnrID *cid.ID, objectName string) (*data.NodeVersion, error) {
panic("implement me")
}
func (t *TreeServiceMock) AddVersion(_ context.Context, cnrID *cid.ID, objectName string, newVersion *data.NodeVersion) error {
cnrVersionsMap, ok := t.versions[cnrID.EncodeToString()]
if !ok {
t.versions[cnrID.EncodeToString()] = map[string][]*data.NodeVersion{
objectName: {newVersion}, objectName: {newVersion},
} }
return nil return nil
@ -103,7 +137,7 @@ func (t *TreeServiceMock) AddVersion(_ context.Context, cnrID *cid.ID, objectNam
versions, ok := cnrVersionsMap[objectName] versions, ok := cnrVersionsMap[objectName]
if !ok { if !ok {
cnrVersionsMap[objectName] = []*layer.NodeVersion{newVersion} cnrVersionsMap[objectName] = []*data.NodeVersion{newVersion}
return nil return nil
} }
@ -115,7 +149,19 @@ func (t *TreeServiceMock) AddVersion(_ context.Context, cnrID *cid.ID, objectNam
newVersion.ID = versions[len(versions)-1].ID + 1 newVersion.ID = versions[len(versions)-1].ID + 1
} }
cnrVersionsMap[objectName] = append(versions, newVersion) result := versions
if newVersion.IsUnversioned {
result = make([]*data.NodeVersion, 0, len(versions))
for _, node := range versions {
if !node.IsUnversioned {
result = append(result, node)
}
}
}
cnrVersionsMap[objectName] = append(result, newVersion)
return nil return nil
} }
@ -124,10 +170,10 @@ func (t *TreeServiceMock) RemoveVersion(ctx context.Context, cnrID *cid.ID, node
panic("implement me") panic("implement me")
} }
func (t *TreeServiceMock) AddSystemVersion(_ context.Context, cnrID *cid.ID, objectName string, newVersion *layer.BaseNodeVersion) error { func (t *TreeServiceMock) AddSystemVersion(_ context.Context, cnrID *cid.ID, objectName string, newVersion *data.BaseNodeVersion) error {
cnrSystemMap, ok := t.system[cnrID.EncodeToString()] cnrSystemMap, ok := t.system[cnrID.EncodeToString()]
if !ok { if !ok {
t.system[cnrID.EncodeToString()] = map[string]*layer.BaseNodeVersion{ t.system[cnrID.EncodeToString()] = map[string]*data.BaseNodeVersion{
objectName: newVersion, objectName: newVersion,
} }
return nil return nil
@ -138,15 +184,15 @@ func (t *TreeServiceMock) AddSystemVersion(_ context.Context, cnrID *cid.ID, obj
return nil return nil
} }
func (t *TreeServiceMock) GetSystemVersion(_ context.Context, cnrID *cid.ID, objectName string) (*layer.BaseNodeVersion, error) { func (t *TreeServiceMock) GetSystemVersion(_ context.Context, cnrID *cid.ID, objectName string) (*data.BaseNodeVersion, error) {
cnrSystemMap, ok := t.system[cnrID.EncodeToString()] cnrSystemMap, ok := t.system[cnrID.EncodeToString()]
if !ok { if !ok {
return nil, layer.ErrNodeNotFound return nil, ErrNodeNotFound
} }
sysVersion, ok := cnrSystemMap[objectName] sysVersion, ok := cnrSystemMap[objectName]
if !ok { if !ok {
return nil, layer.ErrNodeNotFound return nil, ErrNodeNotFound
} }
return sysVersion, nil return sysVersion, nil
@ -155,3 +201,7 @@ func (t *TreeServiceMock) GetSystemVersion(_ context.Context, cnrID *cid.ID, obj
func (t *TreeServiceMock) RemoveSystemVersion(ctx context.Context, cnrID *cid.ID, nodeID uint64) error { func (t *TreeServiceMock) RemoveSystemVersion(ctx context.Context, cnrID *cid.ID, nodeID uint64) error {
panic("implement me") panic("implement me")
} }
func (t *TreeServiceMock) GetAllVersionsByPrefix(ctx context.Context, cnrID *cid.ID, prefix string) ([]*data.NodeVersion, error) {
panic("implement me")
}