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

[#186] Fix versioning issues in UploadPart

Browse source 
Signed-off-by: Angira Kekteeva <kira@nspcc.ru>
This commit is contained in:
Angira Kekteeva 2021-11-25 18:08:02 +03:00 committed by Alex Vanin
parent 873622d4d5
commit d6f0ab8ea4
5 changed files with 43 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
api/handler/multipart_upload.go
View file

@ -5,11 +5,13 @@ import (
"encoding/json"
"encoding/xml"
"net/http"
"net/url"
"strconv"
"time"
"github.com/google/uuid"
"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/errors"
"github.com/nspcc-dev/neofs-s3-gw/api/layer"
"go.uber.org/zap"
@ -251,6 +253,8 @@ func (h *handler) UploadPartCopy(w http.ResponseWriter, r *http.Request) {
}
var (
versionID string
queryValues = reqInfo.URL.Query()
uploadID = queryValues.Get(uploadIDHeaderName)
additional = []zap.Field{zap.String("uploadID", uploadID), zap.String("Key", reqInfo.ObjectName)}
@ -263,6 +267,10 @@ func (h *handler) UploadPartCopy(w http.ResponseWriter, r *http.Request) {
}
src := r.Header.Get(api.AmzCopySource)
if u, err := url.Parse(src); err == nil {
versionID = u.Query().Get(api.QueryVersionID)
src = u.Path
}
srcBucket, srcObject := path2BucketObject(src)
srcRange, err := parseRange(r.Header.Get(api.AmzCopySourceRange))
@ -275,11 +283,27 @@ func (h *handler) UploadPartCopy(w http.ResponseWriter, r *http.Request) {
srcInfo, err := h.obj.GetObjectInfo(r.Context(), &layer.HeadObjectParams{
Bucket: srcBucket,
Object: srcObject,
VersionID: versionID,
})
if err != nil {
if errors.IsS3Error(err, errors.ErrNoSuchKey) && versionID != "" {
h.logAndSendError(w, "could not head source object version", reqInfo,
errors.GetAPIError(errors.ErrBadRequest), additional...)
return
}
h.logAndSendError(w, "could not head source object", reqInfo, err, additional...)
return
}
if isDeleted(srcInfo) {
if versionID != "" {
h.logAndSendError(w, "could not head source object version", reqInfo,
errors.GetAPIError(errors.ErrBadRequest), additional...)
return
}
h.logAndSendError(w, "could not head source object", reqInfo,
errors.GetAPIError(errors.ErrNoSuchKey), additional...)
return
}
args, err := parseCopyObjectArgs(r.Header)
if err != nil {
@ -643,3 +667,7 @@ func encodeListPartsToResponse(info *layer.ListPartsInfo, params *layer.ListPart
Parts: info.Parts,
}
}
func isDeleted(objInfo *data.ObjectInfo) bool {
return objInfo.Headers[layer.VersionsDeleteMarkAttr] == layer.DelMarkFullObject
}

6
api/layer/layer.go
View file

@ -571,7 +571,7 @@ func (n *layer) deleteObject(ctx context.Context, bkt *data.BucketInfo, obj *Ver
p := &PutObjectParams{
Object: obj.Name,
Reader: bytes.NewReader(nil),
Header: map[string]string{versionsDeleteMarkAttr: obj.VersionID},
Header: map[string]string{VersionsDeleteMarkAttr: obj.VersionID},
}
if len(obj.VersionID) != 0 {
version, err := n.checkVersionsExist(ctx, bkt, obj)
@ -580,13 +580,13 @@ func (n *layer) deleteObject(ctx context.Context, bkt *data.BucketInfo, obj *Ver
return obj
}
ids = []*object.ID{version.ID}
if version.Headers[versionsDeleteMarkAttr] == delMarkFullObject {
if version.Headers[VersionsDeleteMarkAttr] == DelMarkFullObject {
obj.DeleteMarkVersion = version.Version()
}
p.Header[versionsDelAttr] = obj.VersionID
} else {
p.Header[versionsDeleteMarkAttr] = delMarkFullObject
p.Header[VersionsDeleteMarkAttr] = DelMarkFullObject
}
objInfo, err := n.objectPut(ctx, bkt, p)
if err != nil {

2
api/layer/object.go
View file

@ -171,7 +171,7 @@ func (n *layer) objectPut(ctx context.Context, bkt *data.BucketInfo, p *PutObjec
return nil, err
}
if p.Header[versionsDeleteMarkAttr] == delMarkFullObject {
if p.Header[VersionsDeleteMarkAttr] == DelMarkFullObject {
if last := versions.getLast(); last != nil {
n.objCache.Delete(last.Address())
}

15
api/layer/versioning.go
View file

@ -21,14 +21,15 @@ type objectVersions struct {
}
const (
VersionsDeleteMarkAttr = "S3-Versions-delete-mark"
DelMarkFullObject = "*"
unversionedObjectVersionID = "null"
objectSystemAttributeName = "S3-System-name"
attrVersionsIgnore = "S3-Versions-ignore"
attrSettingsVersioningEnabled = "S3-Settings-Versioning-enabled"
versionsDelAttr = "S3-Versions-del"
versionsAddAttr = "S3-Versions-add"
versionsDeleteMarkAttr = "S3-Versions-delete-mark"
delMarkFullObject = "*"
)
func newObjectVersions(name string) *objectVersions {
@ -169,11 +170,11 @@ func (v *objectVersions) getLast() *data.ObjectInfo {
existedVersions := v.existedVersions()
for i := len(v.objects) - 1; i >= 0; i-- {
if contains(existedVersions, v.objects[i].Version()) {
delMarkHeader := v.objects[i].Headers[versionsDeleteMarkAttr]
delMarkHeader := v.objects[i].Headers[VersionsDeleteMarkAttr]
if delMarkHeader == "" {
return v.objects[i]
}
if delMarkHeader == delMarkFullObject {
if delMarkHeader == DelMarkFullObject {
return nil
}
}
@ -203,8 +204,8 @@ func (v *objectVersions) getFiltered(reverse bool) []*data.ObjectInfo {
res := make([]*data.ObjectInfo, 0, len(v.objects))
for _, version := range v.objects {
delMark := version.Headers[versionsDeleteMarkAttr]
if contains(existedVersions, version.Version()) && (delMark == delMarkFullObject || delMark == "") {
delMark := version.Headers[VersionsDeleteMarkAttr]
if contains(existedVersions, version.Version()) && (delMark == DelMarkFullObject || delMark == "") {
res = append(res, version)
}
}
@ -335,7 +336,7 @@ func triageVersions(objVersions []*ObjectVersionInfo) ([]*ObjectVersionInfo, []*
var resDelMarkVersions []*ObjectVersionInfo
for _, version := range objVersions {
if version.Object.Headers[versionsDeleteMarkAttr] == delMarkFullObject {
if version.Object.Headers[VersionsDeleteMarkAttr] == DelMarkFullObject {
resDelMarkVersions = append(resDelMarkVersions, version)
} else {
resVersion = append(resVersion, version)

2
api/layer/versioning_test.go
View file

@ -672,7 +672,7 @@ func getTestObjectInfo(id byte, addAttr, delAttr, delMarkAttr string) *data.Obje
headers[versionsDelAttr] = delAttr
}
if delMarkAttr != "" {
headers[versionsDeleteMarkAttr] = delMarkAttr
headers[VersionsDeleteMarkAttr] = delMarkAttr
}
return &data.ObjectInfo{