frostfs-s3-gw/api/layer/tagging.go
Denis Kirillov 18878b66d3 [#175] Use gate owner as object owner
This is required because node check session token owner
TrueCloudLab/frostfs-node#528
For client cut TrueCloudLab/frostfs-sdk-go#114
such owner will be gate owner

Signed-off-by: Denis Kirillov <d.kirillov@yadro.com>
2023-08-08 12:22:33 +03:00

210 lines
6.3 KiB
Go

package layer
import (
"context"
"errors"
"fmt"
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/data"
s3errors "git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/errors"
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/user"
"go.uber.org/zap"
)
type GetObjectTaggingParams struct {
ObjectVersion *ObjectVersion
// NodeVersion can be nil. If not nil we save one request to tree service.
NodeVersion *data.NodeVersion // optional
}
type PutObjectTaggingParams struct {
ObjectVersion *ObjectVersion
TagSet map[string]string
// NodeVersion can be nil. If not nil we save one request to tree service.
NodeVersion *data.NodeVersion // optional
}
func (n *layer) GetObjectTagging(ctx context.Context, p *GetObjectTaggingParams) (string, map[string]string, error) {
var err error
owner := n.BearerOwner(ctx)
if len(p.ObjectVersion.VersionID) != 0 && p.ObjectVersion.VersionID != data.UnversionedObjectVersionID {
if tags := n.cache.GetTagging(owner, objectTaggingCacheKey(p.ObjectVersion)); tags != nil {
return p.ObjectVersion.VersionID, tags, nil
}
}
nodeVersion := p.NodeVersion
if nodeVersion == nil {
nodeVersion, err = n.getNodeVersionFromCacheOrFrostfs(ctx, p.ObjectVersion)
if err != nil {
return "", nil, err
}
}
p.ObjectVersion.VersionID = nodeVersion.OID.EncodeToString()
if tags := n.cache.GetTagging(owner, objectTaggingCacheKey(p.ObjectVersion)); tags != nil {
return p.ObjectVersion.VersionID, tags, nil
}
tags, err := n.treeService.GetObjectTagging(ctx, p.ObjectVersion.BktInfo, nodeVersion)
if err != nil {
if errors.Is(err, ErrNodeNotFound) {
return "", nil, fmt.Errorf("%w: %s", s3errors.GetAPIError(s3errors.ErrNoSuchKey), err.Error())
}
return "", nil, err
}
n.cache.PutTagging(owner, objectTaggingCacheKey(p.ObjectVersion), tags)
return p.ObjectVersion.VersionID, tags, nil
}
func (n *layer) PutObjectTagging(ctx context.Context, p *PutObjectTaggingParams) (nodeVersion *data.NodeVersion, err error) {
nodeVersion = p.NodeVersion
if nodeVersion == nil {
nodeVersion, err = n.getNodeVersionFromCacheOrFrostfs(ctx, p.ObjectVersion)
if err != nil {
return nil, err
}
}
p.ObjectVersion.VersionID = nodeVersion.OID.EncodeToString()
err = n.treeService.PutObjectTagging(ctx, p.ObjectVersion.BktInfo, nodeVersion, p.TagSet)
if err != nil {
if errors.Is(err, ErrNodeNotFound) {
return nil, fmt.Errorf("%w: %s", s3errors.GetAPIError(s3errors.ErrNoSuchKey), err.Error())
}
return nil, err
}
n.cache.PutTagging(n.BearerOwner(ctx), objectTaggingCacheKey(p.ObjectVersion), p.TagSet)
return nodeVersion, nil
}
func (n *layer) DeleteObjectTagging(ctx context.Context, p *ObjectVersion) (*data.NodeVersion, error) {
version, err := n.getNodeVersion(ctx, p)
if err != nil {
return nil, err
}
err = n.treeService.DeleteObjectTagging(ctx, p.BktInfo, version)
if err != nil {
if errors.Is(err, ErrNodeNotFound) {
return nil, fmt.Errorf("%w: %s", s3errors.GetAPIError(s3errors.ErrNoSuchKey), err.Error())
}
return nil, err
}
p.VersionID = version.OID.EncodeToString()
n.cache.DeleteTagging(objectTaggingCacheKey(p))
return version, nil
}
func (n *layer) GetBucketTagging(ctx context.Context, bktInfo *data.BucketInfo) (map[string]string, error) {
owner := n.BearerOwner(ctx)
if tags := n.cache.GetTagging(owner, bucketTaggingCacheKey(bktInfo.CID)); tags != nil {
return tags, nil
}
tags, err := n.treeService.GetBucketTagging(ctx, bktInfo)
if err != nil && !errors.Is(err, ErrNodeNotFound) {
return nil, err
}
n.cache.PutTagging(owner, bucketTaggingCacheKey(bktInfo.CID), tags)
return tags, nil
}
func (n *layer) PutBucketTagging(ctx context.Context, bktInfo *data.BucketInfo, tagSet map[string]string) error {
if err := n.treeService.PutBucketTagging(ctx, bktInfo, tagSet); err != nil {
return err
}
n.cache.PutTagging(n.BearerOwner(ctx), bucketTaggingCacheKey(bktInfo.CID), tagSet)
return nil
}
func (n *layer) DeleteBucketTagging(ctx context.Context, bktInfo *data.BucketInfo) error {
n.cache.DeleteTagging(bucketTaggingCacheKey(bktInfo.CID))
return n.treeService.DeleteBucketTagging(ctx, bktInfo)
}
func objectTaggingCacheKey(p *ObjectVersion) string {
return ".tagset." + p.BktInfo.CID.EncodeToString() + "." + p.ObjectName + "." + p.VersionID
}
func bucketTaggingCacheKey(cnrID cid.ID) string {
return ".tagset." + cnrID.EncodeToString()
}
func (n *layer) getNodeVersion(ctx context.Context, objVersion *ObjectVersion) (*data.NodeVersion, error) {
var err error
var version *data.NodeVersion
if objVersion.VersionID == data.UnversionedObjectVersionID {
version, err = n.treeService.GetUnversioned(ctx, objVersion.BktInfo, objVersion.ObjectName)
} else if len(objVersion.VersionID) == 0 {
version, err = n.treeService.GetLatestVersion(ctx, objVersion.BktInfo, objVersion.ObjectName)
} else {
versions, err2 := n.treeService.GetVersions(ctx, objVersion.BktInfo, objVersion.ObjectName)
if err2 != nil {
return nil, err2
}
for _, v := range versions {
if v.OID.EncodeToString() == objVersion.VersionID {
version = v
break
}
}
if version == nil {
err = fmt.Errorf("%w: there isn't tree node with requested version id", s3errors.GetAPIError(s3errors.ErrNoSuchVersion))
}
}
if err == nil && version.IsDeleteMarker() && !objVersion.NoErrorOnDeleteMarker {
return nil, fmt.Errorf("%w: found version is delete marker", s3errors.GetAPIError(s3errors.ErrNoSuchKey))
} else if errors.Is(err, ErrNodeNotFound) {
return nil, fmt.Errorf("%w: %s", s3errors.GetAPIError(s3errors.ErrNoSuchKey), err.Error())
}
if err == nil && version != nil && !version.IsDeleteMarker() {
n.reqLogger(ctx).Debug("get tree node",
zap.Stringer("cid", objVersion.BktInfo.CID), zap.Stringer("oid", version.OID))
}
return version, err
}
func (n *layer) getNodeVersionFromCache(owner user.ID, o *ObjectVersion) *data.NodeVersion {
if len(o.VersionID) == 0 || o.VersionID == data.UnversionedObjectVersionID {
return nil
}
var objID oid.ID
if objID.DecodeString(o.VersionID) != nil {
return nil
}
var addr oid.Address
addr.SetContainer(o.BktInfo.CID)
addr.SetObject(objID)
extObjectInfo := n.cache.GetObject(owner, addr)
if extObjectInfo == nil {
return nil
}
return extObjectInfo.NodeVersion
}