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

[#476] Save full parts info to multipart object

Browse source 
Signed-off-by: Denis Kirillov <denis@nspcc.ru>
This commit is contained in:
Denis Kirillov 2022-06-02 18:26:26 +03:00 committed by Alex Vanin
parent 8731dcfd3c
commit 9dfc7e043f
6 changed files with 112 additions and 23 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
api/data/tree.go
View file

@ -1,6 +1,7 @@
package data
import (
"strconv"
"time"
cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
@ -67,6 +68,11 @@ type PartInfo struct {
Created time.Time
}
// ToHeaderString form short part representation to use in S3-Completed-Parts header.
func (p *PartInfo) ToHeaderString() string {
return strconv.Itoa(p.Number) + "-" + strconv.FormatInt(p.Size, 10) + "-" + p.ETag
}
// LockInfo is lock information to create appropriate tree node.
type LockInfo struct {
ID uint64

105
api/handler/attributes.go
View file

@ -15,31 +15,37 @@ import (
type (
GetObjectAttributesResponse struct {
ETag string `xml:"ETag,omitempty"`
Checksum *Checksum `xml:"Checksum,omitempty"`
ObjectSize int64 `xml:"ObjectSize,omitempty"`
StorageClass string `xml:"StorageClass,omitempty"`
ObjectParts *ObjectParts `xml:"ObjectParts,omitempty"`
}
Checksum struct {
ChecksumSHA256 string `xml:"ChecksumSHA256,omitempty"`
}
ObjectParts struct {
IsTruncated bool `xml:"IsTruncated,omitempty"`
MaxParts int `xml:"MaxParts,omitempty"`
NextPartNumberMarker int `xml:"NextPartNumberMarker,omitempty"`
PartNumberMarker int `xml:"PartNumberMarker,omitempty"`
Parts []Part `xml:"Part,omitempty"`
// Only this field is used.
PartsCount int `xml:"PartsCount,omitempty"`
PartsCount int `xml:"PartsCount,omitempty"`
}
Part struct {
PartNumber int `xml:"PartNumber,omitempty"`
Size int `xml:"Size,omitempty"`
ChecksumSHA256 string `xml:"ChecksumSHA256,omitempty"`
PartNumber int `xml:"PartNumber,omitempty"`
Size int `xml:"Size,omitempty"`
}
GetObjectAttributesArgs struct {
Attributes []string
VersionID string
Conditional *conditionalArgs
MaxParts int
PartNumberMarker int
Attributes []string
VersionID string
Conditional *conditionalArgs
}
)
@ -138,7 +144,23 @@ func parseGetObjectAttributeArgs(r *http.Request) (*GetObjectAttributesArgs, err
res.Attributes = append(res.Attributes, a)
}
return res, nil
var err error
maxPartsVal := r.Header.Get(api.AmzMaxParts)
if maxPartsVal == "" {
res.MaxParts = layer.MaxSizePartsList
} else if res.MaxParts, err = strconv.Atoi(maxPartsVal); err != nil || res.MaxParts < 0 {
return nil, errors.GetAPIError(errors.ErrInvalidMaxKeys)
}
markerVal := r.Header.Get(api.AmzPartNumberMarker)
if markerVal != "" {
if res.PartNumberMarker, err = strconv.Atoi(markerVal); err != nil || res.PartNumberMarker < 0 {
return nil, errors.GetAPIError(errors.ErrInvalidPartNumberMarker)
}
}
res.Conditional, err = parseConditionalHeaders(r.Header)
return res, err
}
func encodeToObjectAttributesResponse(info *data.ObjectInfo, p *GetObjectAttributesArgs) (*GetObjectAttributesResponse, error) {
@ -152,8 +174,10 @@ func encodeToObjectAttributesResponse(info *data.ObjectInfo, p *GetObjectAttribu
resp.StorageClass = "STANDARD"
case objectSize:
resp.ObjectSize = info.Size
case checksum:
resp.Checksum = &Checksum{ChecksumSHA256: info.HashSum}
case objectParts:
parts, err := formUploadAttributes(info)
parts, err := formUploadAttributes(info, p.MaxParts, p.PartNumberMarker)
if err != nil {
return nil, fmt.Errorf("form upload attributes: %w", err)
}
@ -166,19 +190,62 @@ func encodeToObjectAttributesResponse(info *data.ObjectInfo, p *GetObjectAttribu
return resp, nil
}
func formUploadAttributes(info *data.ObjectInfo) (*ObjectParts, error) {
var err error
res := ObjectParts{}
partsCountStr, ok := info.Headers[layer.UploadCompletedPartsCount]
func formUploadAttributes(info *data.ObjectInfo, maxParts, marker int) (*ObjectParts, error) {
completedParts, ok := info.Headers[layer.UploadCompletedParts]
if !ok {
return nil, nil
}
res.PartsCount, err = strconv.Atoi(partsCountStr)
if err != nil {
return nil, fmt.Errorf("invalid parts count header '%s': %w", partsCountStr, err)
partInfos := strings.Split(completedParts, ",")
parts := make([]Part, len(partInfos))
for i, p := range partInfos {
// partInfo[0] -- part number, partInfo[1] -- part size, partInfo[2] -- checksum
partInfo := strings.Split(p, "-")
if len(partInfo) != 3 {
return nil, fmt.Errorf("invalid completed parts header")
}
num, err := strconv.Atoi(partInfo[0])
if err != nil {
return nil, err
}
size, err := strconv.Atoi(partInfo[1])
if err != nil {
return nil, err
}
parts[i] = Part{
PartNumber: num,
Size: size,
ChecksumSHA256: partInfo[2],
}
}
return &res, nil
res := &ObjectParts{
PartsCount: len(parts),
}
if marker != 0 {
res.PartNumberMarker = marker
var found bool
for i, n := range parts {
if n.PartNumber == marker {
parts = parts[i:]
found = true
break
}
}
if !found {
return nil, errors.GetAPIError(errors.ErrInvalidPartNumberMarker)
}
}
res.MaxParts = maxParts
if len(parts) > maxParts {
res.IsTruncated = true
res.NextPartNumberMarker = parts[maxParts].PartNumber
parts = parts[:maxParts]
}
res.Parts = parts
return res, nil
}

3
api/handler/atttributes_test.go
View file

@ -67,5 +67,8 @@ func TestGetObjectPartsAttributes(t *testing.T) {
result = &GetObjectAttributesResponse{}
parseTestResponse(t, w, result)
require.NotNil(t, result.ObjectParts)
require.Len(t, result.ObjectParts.Parts, 1)
require.Equal(t, etag, result.ObjectParts.Parts[0].ChecksumSHA256)
require.Equal(t, 8, result.ObjectParts.Parts[0].Size)
require.Equal(t, 1, result.ObjectParts.PartsCount)
}

2
api/headers.go
View file

@ -53,6 +53,8 @@ const (
AmzObjectLockRetainUntilDate = "X-Amz-Object-Lock-Retain-Until-Date"
AmzBypassGovernanceRetention = "X-Amz-Bypass-Governance-Retention"
AmzObjectAttributes = "X-Amz-Object-Attributes"
AmzMaxParts = "X-Amz-Max-Parts"
AmzPartNumberMarker = "X-Amz-Part-Number-Marker"
ContainerID = "X-Container-Id"

14
api/layer/multipart_upload.go
View file

@ -22,7 +22,7 @@ import (
const (
UploadIDAttributeName = "S3-Upload-Id"
UploadPartNumberAttributeName = "S3-Upload-Part-Number"
UploadCompletedPartsCount = "S3-Completed-Parts-Count"
UploadCompletedParts = "S3-Completed-Parts"
metaPrefix = "meta-"
aclPrefix = "acl-"
@ -337,6 +337,7 @@ func (n *layer) CompleteMultipartUpload(ctx context.Context, p *CompleteMultipar
parts := make([]*data.PartInfo, 0, len(p.Parts))
var completedPartsHeader strings.Builder
for i, part := range p.Parts {
partInfo := partsInfo[part.PartNumber]
if part.ETag != partInfo.ETag {
@ -347,10 +348,19 @@ func (n *layer) CompleteMultipartUpload(ctx context.Context, p *CompleteMultipar
return nil, nil, errors.GetAPIError(errors.ErrEntityTooSmall)
}
parts = append(parts, partInfo)
partInfoStr := partInfo.ToHeaderString()
if i != len(p.Parts)-1 {
partInfoStr += ","
}
if _, err = completedPartsHeader.WriteString(partInfoStr); err != nil {
return nil, nil, err
}
}
initMetadata := make(map[string]string, len(multipartInfo.Meta)+1)
initMetadata[UploadCompletedPartsCount] = strconv.Itoa(len(p.Parts))
initMetadata[UploadCompletedParts] = completedPartsHeader.String()
uploadData := &UploadData{
TagSet: make(map[string]string),
ACLHeaders: make(map[string]string),

5
api/layer/tree_mock.go
View file

@ -21,8 +21,9 @@ type TreeServiceMock struct {
}
func (t *TreeServiceMock) GetObjectTaggingAndLock(ctx context.Context, cnrID *cid.ID, objVersion *data.NodeVersion) (map[string]string, *data.LockInfo, error) {
// TODO implement me
panic("implement me")
// TODO implement object tagging
lock, err := t.GetLock(ctx, cnrID, objVersion.ID)
return nil, lock, err
}
func (t *TreeServiceMock) GetObjectTagging(ctx context.Context, cnrID *cid.ID, objVersion *data.NodeVersion) (map[string]string, error) {