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

[#637] Add header to override CopiesNumber

Browse source 
Signed-off-by: Denis Kirillov <denis@nspcc.ru>
This commit is contained in:
Denis Kirillov 2022-08-17 14:18:36 +03:00 committed by Kirillov Denis
parent d2c68589b5
commit c3ad6d2faf
7 changed files with 127 additions and 69 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
api/data/tree.go
View file

@ -67,12 +67,13 @@ type ObjectTaggingInfo struct {
type MultipartInfo struct {
// ID is node id in tree service.
// It's ignored when creating a new multipart upload.
ID uint64
Key string
UploadID string
Owner user.ID
Created time.Time
Meta map[string]string
ID uint64
Key string
UploadID string
Owner user.ID
Created time.Time
Meta map[string]string
CopiesNumber uint32
}
// PartInfo is upload information about part.

21
api/handler/copy.go
View file

@ -121,14 +121,21 @@ func (h *handler) CopyObjectHandler(w http.ResponseWriter, r *http.Request) {
metadata[api.ContentType] = contentType
}
copiesNumber, err := getCopiesNumberOrDefault(metadata, h.cfg.CopiesNumber)
if err != nil {
h.logAndSendError(w, "invalid copies number", reqInfo, err)
return
}
params := &layer.CopyObjectParams{
SrcObject: objInfo,
ScrBktInfo: p.BktInfo,
DstBktInfo: dstBktInfo,
DstObject: reqInfo.ObjectName,
SrcSize: objInfo.Size,
Header: metadata,
Encryption: encryptionParams,
SrcObject: objInfo,
ScrBktInfo: p.BktInfo,
DstBktInfo: dstBktInfo,
DstObject: reqInfo.ObjectName,
SrcSize: objInfo.Size,
Header: metadata,
Encryption: encryptionParams,
CopiesNuber: copiesNumber,
}
settings, err := h.obj.GetBucketSettings(r.Context(), dstBktInfo)

22
api/handler/multipart_upload.go
View file

@ -148,6 +148,12 @@ func (h *handler) CreateMultipartUploadHandler(w http.ResponseWriter, r *http.Re
p.Header[api.ContentType] = contentType
}
p.CopiesNumber, err = getCopiesNumberOrDefault(p.Header, h.cfg.CopiesNumber)
if err != nil {
h.logAndSendError(w, "invalid copies number", reqInfo, err)
return
}
if err = h.obj.CreateMultipartUpload(r.Context(), p); err != nil {
h.logAndSendError(w, "could create multipart upload", reqInfo, err, additional...)
return
@ -215,10 +221,9 @@ func (h *handler) UploadPartHandler(w http.ResponseWriter, r *http.Request) {
Bkt: bktInfo,
Key: reqInfo.ObjectName,
},
PartNumber: partNumber,
Size: r.ContentLength,
Reader: r.Body,
CopiesNumber: h.cfg.CopiesNumber,
PartNumber: partNumber,
Size: r.ContentLength,
Reader: r.Body,
}
p.Info.Encryption, err = h.formEncryptionParams(r.Header)
@ -316,11 +321,10 @@ func (h *handler) UploadPartCopy(w http.ResponseWriter, r *http.Request) {
Bkt: bktInfo,
Key: reqInfo.ObjectName,
},
SrcObjInfo: srcInfo,
SrcBktInfo: srcBktInfo,
PartNumber: partNumber,
Range: srcRange,
CopiesNumber: h.cfg.CopiesNumber,
SrcObjInfo: srcInfo,
SrcBktInfo: srcBktInfo,
PartNumber: partNumber,
Range: srcRange,
}
p.Info.Encryption, err = h.formEncryptionParams(r.Header)

22
api/handler/put.go
View file

@ -212,6 +212,12 @@ func (h *handler) PutObjectHandler(w http.ResponseWriter, r *http.Request) {
metadata[api.Expires] = expires
}
copiesNumber, err := getCopiesNumberOrDefault(metadata, h.cfg.CopiesNumber)
if err != nil {
h.logAndSendError(w, "invalid copies number", reqInfo, err)
return
}
encryption, err := h.formEncryptionParams(r.Header)
if err != nil {
h.logAndSendError(w, "invalid sse headers", reqInfo, err)
@ -225,7 +231,7 @@ func (h *handler) PutObjectHandler(w http.ResponseWriter, r *http.Request) {
Size: r.ContentLength,
Header: metadata,
Encryption: encryption,
CopiesNumber: h.cfg.CopiesNumber,
CopiesNumber: copiesNumber,
}
settings, err := h.obj.GetBucketSettings(r.Context(), bktInfo)
@ -299,6 +305,20 @@ func (h *handler) PutObjectHandler(w http.ResponseWriter, r *http.Request) {
api.WriteSuccessResponseHeadersOnly(w)
}
func getCopiesNumberOrDefault(metadata map[string]string, defaultCopiesNumber uint32) (uint32, error) {
copiesNumberStr, ok := metadata[layer.AttributeNeofsCopiesNumber]
if !ok {
return defaultCopiesNumber, nil
}
copiesNumber, err := strconv.ParseUint(copiesNumberStr, 10, 32)
if err != nil {
return 0, fmt.Errorf("pasrse copies number: %w", err)
}
return uint32(copiesNumber), nil
}
func (h handler) formEncryptionParams(header http.Header) (enc encryption.Params, err error) {
sseCustomerAlgorithm := header.Get(api.AmzServerSideEncryptionCustomerAlgorithm)
sseCustomerKey := header.Get(api.AmzServerSideEncryptionCustomerKey)

22
api/handler/put_test.go
View file

@ -4,10 +4,12 @@ import (
"encoding/json"
"mime/multipart"
"net/http"
"strings"
"testing"
"time"
"github.com/nspcc-dev/neofs-s3-gw/api"
"github.com/nspcc-dev/neofs-s3-gw/api/layer"
"github.com/stretchr/testify/require"
)
@ -104,3 +106,23 @@ func TestEmptyPostPolicy(t *testing.T) {
_, err := checkPostPolicy(r, reqInfo, metadata)
require.NoError(t, err)
}
func TestPutObjectOverrideCopiesNumber(t *testing.T) {
tc := prepareHandlerContext(t)
bktName, objName := "bucket-for-copies-number", "object-for-copies-number"
bktInfo := createTestBucket(tc.Context(), t, tc, bktName)
w, r := prepareTestRequest(t, bktName, objName, nil)
r.Header.Set(api.MetadataPrefix+strings.ToUpper(layer.AttributeNeofsCopiesNumber), "1")
tc.Handler().PutObjectHandler(w, r)
p := &layer.HeadObjectParams{
BktInfo: bktInfo,
Object: objName,
}
objInfo, err := tc.Layer().GetObjectInfo(tc.Context(), p)
require.NoError(t, err)
require.Equal(t, "1", objInfo.Headers[layer.AttributeNeofsCopiesNumber])
}

34
api/layer/layer.go
View file

@ -141,15 +141,16 @@ type (
// CopyObjectParams stores object copy request parameters.
CopyObjectParams struct {
SrcObject *data.ObjectInfo
ScrBktInfo *data.BucketInfo
DstBktInfo *data.BucketInfo
DstObject string
SrcSize int64
Header map[string]string
Range *RangeParams
Lock *data.ObjectLock
Encryption encryption.Params
SrcObject *data.ObjectInfo
ScrBktInfo *data.BucketInfo
DstBktInfo *data.BucketInfo
DstObject string
SrcSize int64
Header map[string]string
Range *RangeParams
Lock *data.ObjectLock
Encryption encryption.Params
CopiesNuber uint32
}
// CreateBucketParams stores bucket create request parameters.
CreateBucketParams struct {
@ -264,6 +265,8 @@ const (
AttributeDecryptedSize = api.NeoFSSystemMetadataPrefix + "Decrypted-Size"
AttributeHMACSalt = api.NeoFSSystemMetadataPrefix + "HMAC-Salt"
AttributeHMACKey = api.NeoFSSystemMetadataPrefix + "HMAC-Key"
AttributeNeofsCopiesNumber = "neofs-copies-number" // such formate to match X-Amz-Meta-Neofs-Copies-Number header
)
func (t *VersionedObject) String() string {
@ -519,12 +522,13 @@ func (n *layer) CopyObject(ctx context.Context, p *CopyObjectParams) (*data.Obje
}()
return n.PutObject(ctx, &PutObjectParams{
BktInfo: p.DstBktInfo,
Object: p.DstObject,
Size: p.SrcSize,
Reader: pr,
Header: p.Header,
Encryption: p.Encryption,
BktInfo: p.DstBktInfo,
Object: p.DstObject,
Size: p.SrcSize,
Reader: pr,
Header: p.Header,
Encryption: p.Encryption,
CopiesNumber: p.CopiesNuber,
})
}

62
api/layer/multipart_upload.go
View file

@ -45,9 +45,10 @@ type (
}
CreateMultipartParams struct {
Info *UploadInfoParams
Header map[string]string
Data *UploadData
Info *UploadInfoParams
Header map[string]string
Data *UploadData
CopiesNumber uint32
}
UploadData struct {
@ -56,20 +57,18 @@ type (
}
UploadPartParams struct {
Info *UploadInfoParams
PartNumber int
Size int64
Reader io.Reader
CopiesNumber uint32
Info *UploadInfoParams
PartNumber int
Size int64
Reader io.Reader
}
UploadCopyParams struct {
Info *UploadInfoParams
SrcObjInfo *data.ObjectInfo
SrcBktInfo *data.BucketInfo
PartNumber int
Range *RangeParams
CopiesNumber uint32
Info *UploadInfoParams
SrcObjInfo *data.ObjectInfo
SrcBktInfo *data.BucketInfo
PartNumber int
Range *RangeParams
}
CompleteMultipartParams struct {
@ -141,11 +140,12 @@ func (n *layer) CreateMultipartUpload(ctx context.Context, p *CreateMultipartPar
}
info := &data.MultipartInfo{
Key: p.Info.Key,
UploadID: p.Info.UploadID,
Owner: n.Owner(ctx),
Created: time.Now(),
Meta: make(map[string]string, metaSize),
Key: p.Info.Key,
UploadID: p.Info.UploadID,
Owner: n.Owner(ctx),
Created: time.Now(),
Meta: make(map[string]string, metaSize),
CopiesNumber: p.CopiesNumber,
}
for key, val := range p.Header {
@ -205,7 +205,7 @@ func (n *layer) uploadPart(ctx context.Context, multipartInfo *data.MultipartInf
Creator: bktInfo.Owner,
Attributes: make([][2]string, 2),
Payload: p.Reader,
CopiesNumber: p.CopiesNumber,
CopiesNumber: multipartInfo.CopiesNumber,
}
decSize := p.Size
@ -301,11 +301,10 @@ func (n *layer) UploadPartCopy(ctx context.Context, p *UploadCopyParams) (*data.
}()
params := &UploadPartParams{
Info: p.Info,
PartNumber: p.PartNumber,
Size: size,
Reader: pr,
CopiesNumber: p.CopiesNumber,
Info: p.Info,
PartNumber: p.PartNumber,
Size: size,
Reader: pr,
}
return n.uploadPart(ctx, multipartInfo, params)
@ -435,12 +434,13 @@ func (n *layer) CompleteMultipartUpload(ctx context.Context, p *CompleteMultipar
r.prm.bktInfo = p.Info.Bkt
obj, err := n.PutObject(ctx, &PutObjectParams{
BktInfo: p.Info.Bkt,
Object: p.Info.Key,
Reader: r,
Header: initMetadata,
Size: multipartObjetSize,
Encryption: p.Info.Encryption,
BktInfo: p.Info.Bkt,
Object: p.Info.Key,
Reader: r,
Header: initMetadata,
Size: multipartObjetSize,
Encryption: p.Info.Encryption,
CopiesNumber: multipartInfo.CopiesNumber,
})
if err != nil {
n.log.Error("could not put a completed object (multipart upload)",