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

[#449] Add tree service for bucket tagging

Browse source 
Signed-off-by: Angira Kekteeva <kira@nspcc.ru>
Signed-off-by: Alex Vanin <alexey@nspcc.ru>
This commit is contained in:
Angira Kekteeva 2022-05-25 05:58:25 +04:00 committed by Alex Vanin
parent 99feb1d936
commit 385a2102f0
8 changed files with 186 additions and 83 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
api/cache/system.go vendored
View file

@ -104,7 +104,8 @@ func (o *SystemCache) GetNotificationConfiguration(key string) *data.Notificatio
return result
}
func (o *SystemCache) GetObjectTagging(key string) map[string]string {
// GetTagging returns tags of a bucket or an object
func (o *SystemCache) GetTagging(key string) map[string]string {
entry, err := o.cache.Get(key)
if err != nil {
return nil
@ -135,7 +136,8 @@ func (o *SystemCache) PutNotificationConfiguration(key string, obj *data.Notific
return o.cache.Set(key, obj)
}
func (o *SystemCache) PutObjectTagging(key string, tagSet map[string]string) error {
// PutTagging puts tags of a bucket or an object
func (o *SystemCache) PutTagging(key string, tagSet map[string]string) error {
return o.cache.Set(key, tagSet)
}

6
api/handler/tagging.go
View file

@ -141,7 +141,7 @@ func (h *handler) PutBucketTaggingHandler(w http.ResponseWriter, r *http.Request
return
}
if err = h.obj.PutBucketTagging(r.Context(), bktInfo, tagSet); err != nil {
if err = h.obj.PutBucketTagging(r.Context(), &bktInfo.CID, tagSet); err != nil {
h.logAndSendError(w, "could not put object tagging", reqInfo, err)
return
}
@ -156,7 +156,7 @@ func (h *handler) GetBucketTaggingHandler(w http.ResponseWriter, r *http.Request
return
}
tagSet, err := h.obj.GetBucketTagging(r.Context(), bktInfo)
tagSet, err := h.obj.GetBucketTagging(r.Context(), &bktInfo.CID)
if err != nil {
h.logAndSendError(w, "could not get object tagging", reqInfo, err)
return
@ -177,7 +177,7 @@ func (h *handler) DeleteBucketTaggingHandler(w http.ResponseWriter, r *http.Requ
return
}
if err = h.obj.DeleteBucketTagging(r.Context(), bktInfo); err != nil {
if err = h.obj.DeleteBucketTagging(r.Context(), &bktInfo.CID); err != nil {
h.logAndSendError(w, "could not delete bucket tagging", reqInfo, err)
return
}

6
api/layer/layer.go
View file

@ -211,9 +211,9 @@ type (
HeadSystemObject(ctx context.Context, bktInfo *data.BucketInfo, name string) (*data.ObjectInfo, error)
GetObjectInfo(ctx context.Context, p *HeadObjectParams) (*data.ObjectInfo, error)
GetBucketTagging(ctx context.Context, bktInfo *data.BucketInfo) (map[string]string, error)
PutBucketTagging(ctx context.Context, bktInfo *data.BucketInfo, tagSet map[string]string) error
DeleteBucketTagging(ctx context.Context, bktInfo *data.BucketInfo) error
GetBucketTagging(ctx context.Context, cnrID *cid.ID) (map[string]string, error)
PutBucketTagging(ctx context.Context, cnrID *cid.ID, tagSet map[string]string) error
DeleteBucketTagging(ctx context.Context, cnrID *cid.ID) error
GetObjectTagging(ctx context.Context, p *data.ObjectTaggingInfo) (map[string]string, error)
PutObjectTagging(ctx context.Context, p *data.ObjectTaggingInfo, tagSet map[string]string) error

74
api/layer/tagging.go
View file

@ -5,10 +5,10 @@ import (
errorsStd "errors"
"strings"
"go.uber.org/zap"
"github.com/nspcc-dev/neofs-s3-gw/api/data"
"github.com/nspcc-dev/neofs-s3-gw/api/errors"
cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
"go.uber.org/zap"
)
func (n *layer) GetObjectTagging(ctx context.Context, p *data.ObjectTaggingInfo) (map[string]string, error) {
@ -16,7 +16,7 @@ func (n *layer) GetObjectTagging(ctx context.Context, p *data.ObjectTaggingInfo)
err error
tags map[string]string
)
tags = n.systemCache.GetObjectTagging(objectTaggingCacheKey(p))
tags = n.systemCache.GetTagging(objectTaggingCacheKey(p))
if tags != nil {
return tags, nil
}
@ -34,7 +34,7 @@ func (n *layer) GetObjectTagging(ctx context.Context, p *data.ObjectTaggingInfo)
return nil, err
}
if err = n.systemCache.PutObjectTagging(objectTaggingCacheKey(p), tags); err != nil {
if err = n.systemCache.PutTagging(objectTaggingCacheKey(p), tags); err != nil {
n.log.Error("couldn't cache system object", zap.Error(err))
}
@ -55,7 +55,7 @@ func (n *layer) PutObjectTagging(ctx context.Context, p *data.ObjectTaggingInfo,
return err
}
if err = n.systemCache.PutObjectTagging(objectTaggingCacheKey(p), tagSet); err != nil {
if err = n.systemCache.PutTagging(objectTaggingCacheKey(p), tagSet); err != nil {
n.log.Error("couldn't cache system object", zap.Error(err))
}
@ -81,53 +81,53 @@ func (n *layer) DeleteObjectTagging(ctx context.Context, p *data.ObjectTaggingIn
return nil
}
func (n *layer) GetBucketTagging(ctx context.Context, bktInfo *data.BucketInfo) (map[string]string, error) {
objInfo, err := n.HeadSystemObject(ctx, bktInfo, formBucketTagObjectName(bktInfo.Name))
if err != nil && !errors.IsS3Error(err, errors.ErrNoSuchKey) {
func (n *layer) GetBucketTagging(ctx context.Context, cnrID *cid.ID) (map[string]string, error) {
var (
err error
tags map[string]string
)
tags = n.systemCache.GetTagging(bucketTaggingCacheKey(cnrID))
if tags != nil {
return tags, nil
}
if tags, err = n.treeService.GetBucketTagging(ctx, cnrID); err != nil && !errorsStd.Is(err, ErrNodeNotFound) {
return nil, err
}
return formTagSet(objInfo), nil
}
func (n *layer) PutBucketTagging(ctx context.Context, bktInfo *data.BucketInfo, tagSet map[string]string) error {
s := &PutSystemObjectParams{
BktInfo: bktInfo,
ObjName: formBucketTagObjectName(bktInfo.Name),
Metadata: tagSet,
Prefix: tagPrefix,
Reader: nil,
if err := n.systemCache.PutTagging(bucketTaggingCacheKey(cnrID), tags); err != nil {
n.log.Error("couldn't cache system object", zap.Error(err))
}
_, err := n.PutSystemObject(ctx, s)
return err
return tags, nil
}
func (n *layer) DeleteBucketTagging(ctx context.Context, bktInfo *data.BucketInfo) error {
return n.DeleteSystemObject(ctx, bktInfo, formBucketTagObjectName(bktInfo.Name))
}
func formTagSet(objInfo *data.ObjectInfo) map[string]string {
var tagSet map[string]string
if objInfo != nil {
tagSet = make(map[string]string, len(objInfo.Headers))
for k, v := range objInfo.Headers {
if strings.HasPrefix(k, tagPrefix) {
if v == tagEmptyMark {
v = ""
}
tagSet[strings.TrimPrefix(k, tagPrefix)] = v
}
}
func (n *layer) PutBucketTagging(ctx context.Context, cnrID *cid.ID, tagSet map[string]string) error {
if err := n.treeService.PutBucketTagging(ctx, cnrID, tagSet); err != nil {
return err
}
if err := n.systemCache.PutTagging(bucketTaggingCacheKey(cnrID), tagSet); err != nil {
n.log.Error("couldn't cache system object", zap.Error(err))
}
return tagSet
return nil
}
func (n *layer) DeleteBucketTagging(ctx context.Context, cnrID *cid.ID) error {
n.systemCache.Delete(bucketTaggingCacheKey(cnrID))
return n.treeService.DeleteBucketTagging(ctx, cnrID)
}
func objectTaggingCacheKey(p *data.ObjectTaggingInfo) string {
return ".tagset." + p.CnrID.EncodeToString() + "." + p.ObjName + "." + p.VersionID
}
func bucketTaggingCacheKey(cnrID *cid.ID) string {
return ".tagset." + cnrID.EncodeToString()
}
func (n *layer) getTaggedObjectVersion(ctx context.Context, p *data.ObjectTaggingInfo) (*data.NodeVersion, error) {
var (
err error

4
api/layer/tree_service.go
View file

@ -34,6 +34,10 @@ type TreeService interface {
PutObjectTagging(ctx context.Context, cnrID *cid.ID, objVersion *data.NodeVersion, tagSet map[string]string) error
DeleteObjectTagging(ctx context.Context, cnrID *cid.ID, objVersion *data.NodeVersion) error
GetBucketTagging(ctx context.Context, cnrID *cid.ID) (map[string]string, error)
PutBucketTagging(ctx context.Context, cnrID *cid.ID, tagSet map[string]string) error
DeleteBucketTagging(ctx context.Context, cnrID *cid.ID) error
GetVersions(ctx context.Context, cnrID *cid.ID, objectName string) ([]*data.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)

8
api/layer/versioning_test.go
View file

@ -557,13 +557,13 @@ func TestDeleteSystemObjectsVersioning(t *testing.T) {
"tag1": "val1",
}
err := tc.layer.PutBucketTagging(tc.ctx, tc.bktInfo, tagSet)
err := tc.layer.PutBucketTagging(tc.ctx, &tc.bktInfo.CID, tagSet)
require.NoError(t, err)
objMeta := tc.getSystemObject(formBucketTagObjectName(tc.bktInfo.CID.EncodeToString()))
tagSet["tag2"] = "val2"
err = tc.layer.PutBucketTagging(tc.ctx, tc.bktInfo, tagSet)
err = tc.layer.PutBucketTagging(tc.ctx, &tc.bktInfo.CID, tagSet)
require.NoError(t, err)
// simulate failed deletion
@ -571,7 +571,7 @@ func TestDeleteSystemObjectsVersioning(t *testing.T) {
objID, _ := objMeta.ID()
tc.testNeoFS.AddObject(newAddress(cnrID, objID).EncodeToString(), objMeta)
tagging, err := tc.layer.GetBucketTagging(tc.ctx, tc.bktInfo)
tagging, err := tc.layer.GetBucketTagging(tc.ctx, &tc.bktInfo.CID)
require.NoError(t, err)
expectedTagSet := map[string]string{
@ -580,7 +580,7 @@ func TestDeleteSystemObjectsVersioning(t *testing.T) {
}
require.Equal(t, expectedTagSet, tagging)
err = tc.layer.DeleteBucketTagging(tc.ctx, tc.bktInfo)
err = tc.layer.DeleteBucketTagging(tc.ctx, &tc.bktInfo.CID)
require.NoError(t, err)
require.Nil(t, tc.getSystemObject(formBucketTagObjectName(tc.bktInfo.Name)))

135
internal/neofs/tree.go
View file

@ -35,6 +35,16 @@ type (
TimeStamp uint64
Meta map[string]string
}
getNodesParams struct {
CnrID *cid.ID
TreeID string
PathAttr string
Path []string
Meta []string
LatestOnly bool
AllAttrs bool
}
)
const (
@ -52,10 +62,11 @@ const (
ownerKV = "Owner"
createdKV = "Created"
settingsFileName = "bucket-settings"
notifConfFileName = "bucket-notifications"
corsFilename = "bucket-cors"
emptyFileName = "<empty>" // to handle trailing slash in name
settingsFileName = "bucket-settings"
notifConfFileName = "bucket-notifications"
corsFilename = "bucket-cors"
emptyFileName = "<empty>" // to handle trailing slash in name
bucketTaggingFilename = "bucket-tagging"
// versionTree -- ID of a tree with object versions.
versionTree = "version"
@ -322,6 +333,53 @@ func (c *TreeClient) DeleteObjectTagging(ctx context.Context, cnrID *cid.ID, obj
return c.removeNode(ctx, cnrID, versionTree, tagNode.ID)
}
func (c *TreeClient) GetBucketTagging(ctx context.Context, cnrID *cid.ID) (map[string]string, error) {
node, err := c.getSystemNodeWithAllAttributes(ctx, cnrID, systemTree, []string{bucketTaggingFilename})
if err != nil {
if strings.Contains(err.Error(), "not found") {
return nil, layer.ErrNodeNotFound
}
return nil, err
}
delete(node.Meta, systemNameKV)
return node.Meta, nil
}
func (c *TreeClient) PutBucketTagging(ctx context.Context, cnrID *cid.ID, tagSet map[string]string) error {
node, err := c.getSystemNode(ctx, cnrID, systemTree, []string{bucketTaggingFilename}, []string{})
isErrNotFound := errors.Is(err, layer.ErrNodeNotFound)
if err != nil && !isErrNotFound {
return fmt.Errorf("couldn't get node: %w", err)
}
tagSet[systemNameKV] = bucketTaggingFilename
if isErrNotFound {
_, err = c.addNode(ctx, cnrID, systemTree, 0, tagSet)
} else {
err = c.moveNode(ctx, cnrID, systemTree, node.ID, 0, tagSet)
}
delete(tagSet, systemNameKV)
return err
}
func (c *TreeClient) DeleteBucketTagging(ctx context.Context, cnrID *cid.ID) error {
node, err := c.getSystemNode(ctx, cnrID, systemTree, []string{bucketTaggingFilename}, nil)
if err != nil && !errors.Is(err, layer.ErrNodeNotFound) {
return err
}
if node != nil {
return c.removeNode(ctx, cnrID, systemTree, node.ID)
}
return nil
}
func (c *TreeClient) getObjectTaggingNode(ctx context.Context, cnrID *cid.ID, objVersion *data.NodeVersion) (*TreeNode, error) {
subtree, err := c.getSubTree(ctx, cnrID, versionTree, objVersion.ID, 1)
if err != nil {
@ -403,9 +461,16 @@ func (c *TreeClient) GetLatestVersionsByPrefix(ctx context.Context, cnrID *cid.I
}
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)
p := &getNodesParams{
CnrID: cnrID,
TreeID: versionTree,
PathAttr: fileNameKV,
Path: prefixPath,
Meta: []string{fileNameKV, oidKV},
LatestOnly: false,
AllAttrs: false,
}
nodes, err := c.getNodes(ctx, p)
if err != nil {
return 0, err
}
@ -541,7 +606,16 @@ func (c *TreeClient) GetSystemVersion(ctx context.Context, cnrID *cid.ID, object
}
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)
p := &getNodesParams{
CnrID: cnrID,
TreeID: treeID,
PathAttr: attrPath,
Path: path,
Meta: meta,
LatestOnly: true,
AllAttrs: false,
}
nodes, err := c.getNodes(ctx, p)
if err != nil {
if strings.Contains(err.Error(), "not found") {
return nil, layer.ErrNodeNotFound
@ -643,7 +717,16 @@ func (c *TreeClient) addVersion(ctx context.Context, cnrID *cid.ID, treeID, attr
func (c *TreeClient) getVersions(ctx context.Context, cnrID *cid.ID, treeID, filepath string, onlyUnversioned bool) ([]*data.NodeVersion, error) {
keysToReturn := []string{oidKV, isUnversionedKV, isDeleteMarkerKV}
path := pathFromName(filepath)
nodes, err := c.getNodes(ctx, cnrID, treeID, fileNameKV, path, keysToReturn, false)
p := &getNodesParams{
CnrID: cnrID,
TreeID: treeID,
PathAttr: fileNameKV,
Path: path,
Meta: keysToReturn,
LatestOnly: false,
AllAttrs: false,
}
nodes, err := c.getNodes(ctx, p)
if err != nil {
if strings.Contains(err.Error(), "not found") {
return nil, nil
@ -733,11 +816,24 @@ func metaFromSettings(settings *data.BucketSettings) map[string]string {
}
func (c *TreeClient) getSystemNode(ctx context.Context, cnrID *cid.ID, treeID string, path, meta []string) (*TreeNode, error) {
return c.getNode(ctx, cnrID, treeID, systemNameKV, path, meta)
return c.getNode(ctx, cnrID, treeID, systemNameKV, path, meta, false)
}
func (c *TreeClient) getNode(ctx context.Context, cnrID *cid.ID, treeID, pathAttr string, path, meta []string) (*TreeNode, error) {
nodes, err := c.getNodes(ctx, cnrID, treeID, pathAttr, path, meta, false)
func (c *TreeClient) getSystemNodeWithAllAttributes(ctx context.Context, cnrID *cid.ID, treeID string, path []string) (*TreeNode, error) {
return c.getNode(ctx, cnrID, treeID, systemNameKV, path, []string{}, true)
}
func (c *TreeClient) getNode(ctx context.Context, cnrID *cid.ID, treeID, pathAttr string, path, meta []string, allAttrs bool) (*TreeNode, error) {
p := &getNodesParams{
CnrID: cnrID,
TreeID: treeID,
PathAttr: pathAttr,
Path: path,
Meta: meta,
LatestOnly: false,
AllAttrs: allAttrs,
}
nodes, err := c.getNodes(ctx, p)
if err != nil {
if strings.Contains(err.Error(), "not found") {
return nil, layer.ErrNodeNotFound
@ -754,15 +850,16 @@ func (c *TreeClient) getNode(ctx context.Context, cnrID *cid.ID, treeID, pathAtt
return newTreeNode(nodes[0])
}
func (c *TreeClient) getNodes(ctx context.Context, cnrID *cid.ID, treeID, pathAttr string, path, meta []string, latestOnly bool) ([]*tree.GetNodeByPathResponse_Info, error) {
func (c *TreeClient) getNodes(ctx context.Context, p *getNodesParams) ([]*tree.GetNodeByPathResponse_Info, error) {
request := &tree.GetNodeByPathRequest{
Body: &tree.GetNodeByPathRequest_Body{
ContainerId: cnrID[:],
TreeId: treeID,
Path: path,
Attributes: meta,
PathAttribute: pathAttr,
LatestOnly: latestOnly,
ContainerId: p.CnrID[:],
TreeId: p.TreeID,
Path: p.Path,
Attributes: p.Meta,
PathAttribute: p.PathAttr,
LatestOnly: p.LatestOnly,
AllAttributes: p.AllAttrs,
BearerToken: getBearer(ctx),
},
}

30
internal/neofstest/tree/tree_mock.go
View file

@ -32,6 +32,21 @@ func (t *TreeServiceMock) DeleteObjectTagging(ctx context.Context, cnrID *cid.ID
panic("implement me")
}
func (t *TreeServiceMock) GetBucketTagging(ctx context.Context, cnrID *cid.ID) (map[string]string, error) {
//TODO implement me
panic("implement me")
}
func (t *TreeServiceMock) PutBucketTagging(ctx context.Context, cnrID *cid.ID, tagSet map[string]string) error {
//TODO implement me
panic("implement me")
}
func (t *TreeServiceMock) DeleteBucketTagging(ctx context.Context, cnrID *cid.ID) error {
//TODO implement me
panic("implement me")
}
var ErrNodeNotFound = errors.New("not found")
func NewTreeService() *TreeServiceMock {
@ -220,18 +235,3 @@ func (t *TreeServiceMock) RemoveSystemVersion(ctx context.Context, cnrID *cid.ID
func (t *TreeServiceMock) GetAllVersionsByPrefix(ctx context.Context, cnrID *cid.ID, prefix string) ([]*data.NodeVersion, error) {
panic("implement me")
}
func (t *TreeServiceMock) GetObjectTagging(ctx context.Context, p *data.ObjectTaggingInfo) (map[string]string, error) {
//TODO implement me
panic("implement me")
}
func (t *TreeServiceMock) PutObjectTagging(ctx context.Context, p *data.ObjectTaggingInfo) error {
//TODO implement me
panic("implement me")
}
func (t *TreeServiceMock) DeleteObjectTagging(ctx context.Context, p *data.ObjectTaggingInfo) error {
//TODO implement me
panic("implement me")
}