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

[#XX] tmp

Browse source 
Signed-off-by: Denis Kirillov <d.kirillov@yadro.com>
This commit is contained in:
Denis Kirillov 2023-08-15 09:44:46 +03:00
parent 6b109eee92
commit 5add8eef3a
7 changed files with 98 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
api/handler/copy.go
View file

@ -53,6 +53,10 @@ func (h *handler) CopyObjectHandler(w http.ResponseWriter, r *http.Request) {
containsACL = containsACLHeaders(r)
)
if containsACL && r.Header.Get(api.AmzACL) == basicACLPrivate {
containsACL = false
}
src := r.Header.Get(api.AmzCopySource)
// Check https://docs.aws.amazon.com/AmazonS3/latest/dev/ObjectVersioning.html
// Regardless of whether you have enabled versioning, each object in your bucket

2
api/handler/multipart_upload.go
View file

@ -177,7 +177,7 @@ func (h *handler) CreateMultipartUploadHandler(w http.ResponseWriter, r *http.Re
func formACLHeadersForMultipart(header http.Header) map[string]string {
result := make(map[string]string)
if value := header.Get(api.AmzACL); value != "" {
if value := header.Get(api.AmzACL); value != "" && value != "private" {
result[api.AmzACL] = value
}
if value := header.Get(api.AmzGrantRead); value != "" {

9
api/handler/object_list.go
View file

@ -191,9 +191,16 @@ func fillContentsWithOwner(src []*data.ObjectInfo, encode string) []Object {
func fillContents(src []*data.ObjectInfo, encode string, fetchOwner bool) []Object {
var dst []Object
for _, obj := range src {
objSize := obj.Size
if len(obj.Headers[layer.MultipartObjectSize]) > 0 {
if size, err := strconv.ParseUint(obj.Headers[layer.MultipartObjectSize], 10, 64); err == nil {
objSize = size
}
}
res := Object{
Key: s3PathEncode(obj.Name, encode),
Size: obj.Size,
Size: objSize,
LastModified: obj.Created.UTC().Format(time.RFC3339),
ETag: obj.HashSum,
}

4
api/handler/put.go
View file

@ -184,6 +184,10 @@ func (h *handler) PutObjectHandler(w http.ResponseWriter, r *http.Request) {
reqInfo = middleware.GetReqInfo(ctx)
)
if containsACL && r.Header.Get(api.AmzACL) == basicACLPrivate {
containsACL = false
}
if containsACL {
if sessionTokenEACL, err = getSessionTokenSetEACL(r.Context()); err != nil {
h.logAndSendError(w, "could not get eacl session token from a box", reqInfo, err)

12
api/layer/multipart_upload.go
View file

@ -288,10 +288,18 @@ func (n *layer) UploadPartCopy(ctx context.Context, p *UploadCopyParams) (*data.
}
size := p.SrcObjInfo.Size
srcObjectSize := p.SrcObjInfo.Size
if len(p.SrcObjInfo.Headers[MultipartObjectSize]) > 0 {
if multSize, err := strconv.ParseUint(p.SrcObjInfo.Headers[MultipartObjectSize], 10, 64); err == nil {
srcObjectSize = multSize
}
}
if p.Range != nil {
size = p.Range.End - p.Range.Start + 1
if p.Range.End > p.SrcObjInfo.Size {
return nil, fmt.Errorf("%w: %d-%d/%d", s3errors.GetAPIError(s3errors.ErrInvalidCopyPartRangeSource), p.Range.Start, p.Range.End, p.SrcObjInfo.Size)
if p.Range.End > srcObjectSize {
return nil, fmt.Errorf("%w: %d-%d/%d", s3errors.GetAPIError(s3errors.ErrInvalidCopyPartRangeSource), p.Range.Start, p.Range.End, srcObjectSize)
}
}
if size > UploadMaxSize {

3
internal/frostfs/frostfs.go
View file

@ -451,6 +451,9 @@ func handleObjectError(msg string, err error) error {
if err == nil {
return nil
}
if err == io.EOF {
return err
}
if reason, ok := errorsFrost.IsErrObjectAccessDenied(err); ok {
return fmt.Errorf("%s: %w: %s", msg, layer.ErrAccessDenied, reason)

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

@ -220,6 +220,30 @@ func newNodeVersionFromTreeNode(filePath string, treeNode *treeNode) *data.NodeV
return version
}
func newMultipartInfoFromTreeNode(filePath string, treeNode *treeNode) (*data.MultipartInfo, error) {
uploadID, _ := treeNode.Get(uploadIDKV)
if uploadID == "" {
return nil, fmt.Errorf("it's not a multipart node")
}
multipartInfo := &data.MultipartInfo{
ID: treeNode.ID,
Key: filePath,
UploadID: uploadID,
Meta: treeNode.Meta,
}
ownerID, _ := treeNode.Get(ownerKV)
_ = multipartInfo.Owner.DecodeString(ownerID)
created, _ := treeNode.Get(createdKV)
if utcMilli, err := strconv.ParseInt(created, 10, 64); err == nil {
multipartInfo.Created = time.UnixMilli(utcMilli)
}
return multipartInfo, nil
}
func newMultipartInfo(node NodeResponse) (*data.MultipartInfo, error) {
multipartInfo := &data.MultipartInfo{
ID: node.GetNodeID(),
@ -857,14 +881,14 @@ func (c *Tree) CreateMultipartUpload(ctx context.Context, bktInfo *data.BucketIn
}
func (c *Tree) GetMultipartUploadsByPrefix(ctx context.Context, bktInfo *data.BucketInfo, prefix string) ([]*data.MultipartInfo, error) {
subTreeNodes, _, err := c.getSubTreeByPrefix(ctx, bktInfo, systemTree, prefix, false)
subTreeNodes, headPrefix, err := c.getSubTreeByPrefix(ctx, bktInfo, systemTree, prefix, false)
if err != nil {
return nil, err
}
var result []*data.MultipartInfo
for _, node := range subTreeNodes {
multipartUploads, err := c.getSubTreeMultipartUploads(ctx, bktInfo, node.GetNodeID())
multipartUploads, err := c.getSubTreeMultipartUploads(ctx, bktInfo, node.GetNodeID(), headPrefix)
if err != nil {
return nil, err
}
@ -874,19 +898,55 @@ func (c *Tree) GetMultipartUploadsByPrefix(ctx context.Context, bktInfo *data.Bu
return result, nil
}
func (c *Tree) getSubTreeMultipartUploads(ctx context.Context, bktInfo *data.BucketInfo, nodeID uint64) ([]*data.MultipartInfo, error) {
func (c *Tree) getSubTreeMultipartUploads(ctx context.Context, bktInfo *data.BucketInfo, nodeID uint64, parentFilePath string) ([]*data.MultipartInfo, error) {
subTree, err := c.service.GetSubTree(ctx, bktInfo, systemTree, nodeID, maxGetSubTreeDepth)
if err != nil {
return nil, err
}
result := make([]*data.MultipartInfo, 0, len(subTree))
for _, node := range subTree {
multipartInfo, err := newMultipartInfo(node)
if err != nil { // missed uploadID (it's a part node)
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 filepath string
namesMap := make(map[uint64]string, len(subTree))
multiparts := make(map[string][]*data.MultipartInfo, len(subTree))
for i, node := range subTree {
treeNode, fileName, err := parseTreeNode(node)
if err != nil {
continue
}
result = append(result, multipartInfo)
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
}
multipartInfo, err := newMultipartInfoFromTreeNode(filepath, treeNode)
if err != nil {
continue
}
key := formLatestNodeKey(node.GetParentID(), fileName)
multipartInfos, ok := multiparts[key]
if !ok {
multipartInfos = []*data.MultipartInfo{multipartInfo}
} else {
multipartInfos = append(multipartInfos, multipartInfo)
}
multiparts[key] = multipartInfos
}
result := make([]*data.MultipartInfo, 0, len(multiparts))
for _, multipartInfo := range multiparts {
result = append(result, multipartInfo...)
}
return result, nil