mirrors/versitygw
1
0
Fork 0
mirror of https://github.com/versity/versitygw.git synced 2026-08-17 20:56:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: Bug fixing for azure backend. Added a new integration test case for ListParts

Browse Source
This commit is contained in:
jonaustin09
2024-06-11 16:14:35 -04:00
parent f4cf0132e5
commit 7ea386aec9
6 changed files with 206 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/azure/azure.go
+134 -49
View File
@@ -25,6 +25,7 @@ import (
"io"
"math"
"os"
"slices"
"strconv"
"strings"
"time"
@@ -215,7 +216,17 @@ func (az *Azure) HeadBucket(ctx context.Context, input *s3.HeadBucketInput) (*s3
}
func (az *Azure) DeleteBucket(ctx context.Context, input *s3.DeleteBucketInput) error {
_, err := az.client.DeleteContainer(ctx, *input.Bucket, nil)
pager := az.client.NewListBlobsFlatPager(*input.Bucket, nil)
pg, err := pager.NextPage(ctx)
if err != nil {
return azureErrToS3Err(err)
}
if len(pg.Segment.BlobItems) > 0 {
return s3err.GetAPIError(s3err.ErrBucketNotEmpty)
}
_, err = az.client.DeleteContainer(ctx, *input.Bucket, nil)
return azureErrToS3Err(err)
}
@@ -319,7 +330,7 @@ func (az *Azure) DeleteBucketTagging(ctx context.Context, bucket string) error {
func (az *Azure) GetObject(ctx context.Context, input *s3.GetObjectInput, writer io.Writer) (*s3.GetObjectOutput, error) {
var opts *azblob.DownloadStreamOptions
if *input.Range != "" {
offset, count, err := parseRange(*input.Range)
offset, count, err := backend.ParseRange(0, *input.Range)
if err != nil {
return nil, err
}
@@ -360,6 +371,37 @@ func (az *Azure) GetObject(ctx context.Context, input *s3.GetObjectInput, writer
}
func (az *Azure) HeadObject(ctx context.Context, input *s3.HeadObjectInput) (*s3.HeadObjectOutput, error) {
if input.PartNumber != nil {
client, err := az.getBlockBlobClient(*input.Bucket, *input.Key)
if err != nil {
return nil, err
}
res, err := client.GetBlockList(ctx, blockblob.BlockListTypeUncommitted, nil)
if err != nil {
return nil, azureErrToS3Err(err)
}
partsCount := int32(len(res.UncommittedBlocks))
for _, block := range res.UncommittedBlocks {
partNumber, err := decodeBlockId(*block.Name)
if err != nil {
return nil, err
}
if partNumber == int(*input.PartNumber) {
return &s3.HeadObjectOutput{
ContentLength: block.Size,
ETag: block.Name,
PartsCount: &partsCount,
}, nil
}
}
return nil, s3err.GetAPIError(s3err.ErrNoSuchKey)
}
client, err := az.getBlobClient(*input.Bucket, *input.Key)
if err != nil {
return nil, err
@@ -370,7 +412,7 @@ func (az *Azure) HeadObject(ctx context.Context, input *s3.HeadObjectInput) (*s3
return nil, azureErrToS3Err(err)
}
return &s3.HeadObjectOutput{
result := &s3.HeadObjectOutput{
AcceptRanges: resp.AcceptRanges,
ContentLength: resp.ContentLength,
ContentType: resp.ContentType,
@@ -381,7 +423,27 @@ func (az *Azure) HeadObject(ctx context.Context, input *s3.HeadObjectInput) (*s3
LastModified: resp.LastModified,
Metadata: parseAzMetadata(resp.Metadata),
Expires: resp.ExpiresOn,
}, nil
}
status, ok := resp.Metadata[string(keyObjLegalHold)]
if ok {
if *status == "1" {
result.ObjectLockLegalHoldStatus = types.ObjectLockLegalHoldStatusOn
} else {
result.ObjectLockLegalHoldStatus = types.ObjectLockLegalHoldStatusOff
}
}
retention, ok := resp.Metadata[string(keyObjRetention)]
if ok {
var config types.ObjectLockRetention
if err := json.Unmarshal([]byte(*retention), &config); err == nil {
result.ObjectLockMode = types.ObjectLockMode(config.Mode)
result.ObjectLockRetainUntilDate = config.RetainUntilDate
}
}
return result, nil
}
func (az *Azure) GetObjectAttributes(ctx context.Context, input *s3.GetObjectAttributesInput) (s3response.GetObjectAttributesResult, error) {
@@ -433,7 +495,7 @@ func (az *Azure) GetObjectAttributes(ctx context.Context, input *s3.GetObjectAtt
IsTruncated: resp.IsTruncated,
MaxParts: resp.MaxParts,
PartNumberMarker: resp.PartNumberMarker,
NextPartNumberMarker: resp.PartNumberMarker,
NextPartNumberMarker: resp.NextPartNumberMarker,
Parts: parts,
},
}, nil
@@ -494,8 +556,14 @@ Pager:
}
func (az *Azure) ListObjectsV2(ctx context.Context, input *s3.ListObjectsV2Input) (*s3.ListObjectsV2Output, error) {
marker := ""
if *input.ContinuationToken > *input.StartAfter {
marker = *input.ContinuationToken
} else {
marker = *input.StartAfter
}
pager := az.client.NewListBlobsFlatPager(*input.Bucket, &azblob.ListBlobsFlatOptions{
Marker: input.ContinuationToken,
Marker: &marker,
MaxResults: input.MaxKeys,
Prefix: input.Prefix,
})
@@ -544,6 +612,7 @@ Pager:
NextContinuationToken: nextMarker,
Prefix: input.Prefix,
IsTruncated: &isTruncated,
Delimiter: input.Delimiter,
}, nil
}
@@ -760,20 +829,20 @@ func (az *Azure) ListParts(ctx context.Context, input *s3.ListPartsInput) (s3res
if err != nil {
return s3response.ListPartsResult{}, err
}
if partNumberMarker != 0 && partNumberMarker < partNumber {
if partNumberMarker != 0 && partNumberMarker >= partNumber {
continue
}
if len(parts) >= int(maxParts) {
nextPartNumberMarker = partNumber
isTruncated = true
break
}
parts = append(parts, s3response.Part{
Size: *el.Size,
ETag: *el.Name,
PartNumber: partNumber,
LastModified: time.Now().Format(backend.RFC3339TimeFormat),
})
if len(parts) >= int(maxParts) {
nextPartNumberMarker = partNumber
isTruncated = true
break
}
}
return s3response.ListPartsResult{
Bucket: *input.Bucket,
@@ -861,9 +930,37 @@ func (az *Azure) CompleteMultipartUpload(ctx context.Context, input *s3.Complete
return nil, err
}
blockIds := []string{}
for _, el := range input.MultipartUpload.Parts {
blockIds = append(blockIds, *el.ETag)
blockList, err := client.GetBlockList(ctx, blockblob.BlockListTypeUncommitted, nil)
if err != nil {
return nil, azureErrToS3Err(err)
}
if len(blockList.UncommittedBlocks) != len(input.MultipartUpload.Parts) {
return nil, s3err.GetAPIError(s3err.ErrInvalidPart)
}
slices.SortFunc(blockList.UncommittedBlocks, func(a *blockblob.Block, b *blockblob.Block) int {
ptNumber, _ := decodeBlockId(*a.Name)
nextPtNumber, _ := decodeBlockId(*b.Name)
return ptNumber - nextPtNumber
})
for i, block := range blockList.UncommittedBlocks {
ptNumber, err := decodeBlockId(*block.Name)
if err != nil {
return nil, s3err.GetAPIError(s3err.ErrInvalidPart)
}
if *input.MultipartUpload.Parts[i].ETag != *block.Name {
return nil, s3err.GetAPIError(s3err.ErrInvalidPart)
}
if *input.MultipartUpload.Parts[i].PartNumber != int32(ptNumber) {
return nil, s3err.GetAPIError(s3err.ErrInvalidPart)
}
blockIds = append(blockIds, *block.Name)
}
resp, err := client.CommitBlockList(ctx, blockIds, nil)
if err != nil {
return nil, parseMpError(err)
@@ -954,7 +1051,7 @@ func (az *Azure) GetBucketPolicy(ctx context.Context, bucket string) ([]byte, er
policyPtr, ok := props.Metadata[string(keyPolicy)]
if !ok {
return nil, s3err.GetAPIError(s3err.ErrNoSuchBucket)
return nil, s3err.GetAPIError(s3err.ErrNoSuchBucketPolicy)
}
policy, err := base64.StdEncoding.DecodeString(*policyPtr)
@@ -1064,7 +1161,28 @@ func (az *Azure) PutObjectRetention(ctx context.Context, bucket, object, version
string(keyObjRetention): backend.GetStringPtr(string(retention)),
}
} else {
meta[string(keyObjRetention)] = backend.GetStringPtr(string(retention))
objLockCfg, ok := meta[string(keyObjRetention)]
if !ok {
meta[string(keyObjRetention)] = backend.GetStringPtr(string(retention))
} else {
var lockCfg types.ObjectLockRetention
if err := json.Unmarshal([]byte(*objLockCfg), &lockCfg); err != nil {
return fmt.Errorf("unmarshal object lock config: %w", err)
}
switch lockCfg.Mode {
// Compliance mode can't be overridden
case types.ObjectLockRetentionModeCompliance:
return s3err.GetAPIError(s3err.ErrMethodNotAllowed)
// To override governance mode user should have "s3:BypassGovernanceRetention" permission
case types.ObjectLockRetentionModeGovernance:
if !bypass {
return s3err.GetAPIError(s3err.ErrMethodNotAllowed)
}
}
meta[string(keyObjRetention)] = backend.GetStringPtr(string(retention))
}
}
_, err = blobClient.SetMetadata(ctx, meta, nil)
@@ -1356,39 +1474,6 @@ func decodeBlockId(blockID string) (int, error) {
return int(binary.LittleEndian.Uint32(slice)), nil
}
func parseRange(rg string) (offset, count int64, err error) {
rangeKv := strings.Split(rg, "=")
if len(rangeKv) < 2 {
return 0, 0, s3err.GetAPIError(s3err.ErrInvalidRange)
}
bRange := strings.Split(rangeKv[1], "-")
if len(bRange) < 1 || len(bRange) > 2 {
return 0, 0, s3err.GetAPIError(s3err.ErrInvalidRange)
}
offset, err = strconv.ParseInt(bRange[0], 10, 64)
if err != nil {
return 0, 0, s3err.GetAPIError(s3err.ErrInvalidRange)
}
if len(bRange) == 1 || bRange[1] == "" {
return offset, count, nil
}
count, err = strconv.ParseInt(bRange[1], 10, 64)
if err != nil {
return 0, 0, s3err.GetAPIError(s3err.ErrInvalidRange)
}
if count < offset {
return 0, 0, s3err.GetAPIError(s3err.ErrInvalidRange)
}
return offset, count - offset + 1, nil
}
func getAclFromMetadata(meta map[string]*string, key key) (*auth.ACL, error) {
aclPtr, ok := meta[string(key)]
if !ok {
backend/common.go
+2 -3
View File
@@ -18,7 +18,6 @@ import (
"crypto/md5"
"encoding/hex"
"fmt"
"io/fs"
"strconv"
"strings"
"time"
@@ -61,9 +60,9 @@ var (
// ParseRange parses input range header and returns startoffset, length, and
// error. If no endoffset specified, then length is set to -1.
func ParseRange(fi fs.FileInfo, acceptRange string) (int64, int64, error) {
func ParseRange(size int64, acceptRange string) (int64, int64, error) {
if acceptRange == "" {
return 0, fi.Size(), nil
return 0, size, nil
}
rangeKv := strings.Split(acceptRange, "=")
backend/posix/posix.go
+2 -2
View File
@@ -1199,7 +1199,7 @@ func (p *Posix) UploadPartCopy(ctx context.Context, upi *s3.UploadPartCopyInput)
return s3response.CopyObjectResult{}, fmt.Errorf("stat object: %w", err)
}
startOffset, length, err := backend.ParseRange(fi, *upi.CopySourceRange)
startOffset, length, err := backend.ParseRange(fi.Size(), *upi.CopySourceRange)
if err != nil {
return s3response.CopyObjectResult{}, err
}
@@ -1565,7 +1565,7 @@ func (p *Posix) GetObject(_ context.Context, input *s3.GetObjectInput, writer io
}
acceptRange := *input.Range
startOffset, length, err := backend.ParseRange(fi, acceptRange)
startOffset, length, err := backend.ParseRange(fi.Size(), acceptRange)
if err != nil {
return nil, err
}
backend/scoutfs/scoutfs.go
+1 -1
View File
@@ -611,7 +611,7 @@ func (s *ScoutFS) GetObject(_ context.Context, input *s3.GetObjectInput, writer
return nil, fmt.Errorf("stat object: %w", err)
}
startOffset, length, err := backend.ParseRange(fi, acceptRange)
startOffset, length, err := backend.ParseRange(fi.Size(), acceptRange)
if err != nil {
return nil, err
}
tests/integration/group-tests.go
+2
View File
@@ -236,6 +236,7 @@ func TestUploadPartCopy(s *S3Conf) {
func TestListParts(s *S3Conf) {
ListParts_incorrect_uploadId(s)
ListParts_incorrect_object_key(s)
ListParts_truncated(s)
ListParts_success(s)
}
@@ -605,6 +606,7 @@ func GetIntTests() IntTests {
"UploadPartCopy_by_range_success": UploadPartCopy_by_range_success,
"ListParts_incorrect_uploadId": ListParts_incorrect_uploadId,
"ListParts_incorrect_object_key": ListParts_incorrect_object_key,
"ListParts_truncated": ListParts_truncated,
"ListParts_success": ListParts_success,
"ListMultipartUploads_non_existing_bucket": ListMultipartUploads_non_existing_bucket,
"ListMultipartUploads_empty_result": ListMultipartUploads_empty_result,
tests/integration/tests.go
+65 -1
View File
@@ -5274,6 +5274,70 @@ func ListParts_incorrect_object_key(s *S3Conf) error {
})
}
func ListParts_truncated(s *S3Conf) error {
testName := "ListParts_truncated"
return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
obj := "my-obj"
out, err := createMp(s3client, bucket, obj)
if err != nil {
return err
}
parts, err := uploadParts(s3client, 5*1024*1024, 5, bucket, obj, *out.UploadId)
if err != nil {
return err
}
maxParts := int32(3)
ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
res, err := s3client.ListParts(ctx, &s3.ListPartsInput{
Bucket: &bucket,
Key: &obj,
UploadId: out.UploadId,
MaxParts: &maxParts,
})
cancel()
if err != nil {
return err
}
if !*res.IsTruncated {
return fmt.Errorf("expected the result to be truncated")
}
if *res.MaxParts != maxParts {
return fmt.Errorf("expected max-parts to be %v, instead got %v", maxParts, *res.MaxParts)
}
if *res.NextPartNumberMarker != fmt.Sprint(*parts[2].PartNumber) {
return fmt.Errorf("expected next part number marker to be %v, instead got %v", fmt.Sprint(*parts[2].PartNumber), *res.NextPartNumberMarker)
}
if ok := compareParts(res.Parts, parts[:3]); !ok {
return fmt.Errorf("expected the parts data to be %v, instead got %v", parts[:3], res.Parts)
}
ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
res2, err := s3client.ListParts(ctx, &s3.ListPartsInput{
Bucket: &bucket,
Key: &obj,
UploadId: out.UploadId,
PartNumberMarker: res.NextPartNumberMarker,
})
cancel()
if err != nil {
return err
}
if *res2.PartNumberMarker != *res.NextPartNumberMarker {
return fmt.Errorf("expected part number marker to be %v, instead got %v", *res.NextPartNumberMarker, *res2.PartNumberMarker)
}
if ok := compareParts(parts[3:], res2.Parts); !ok {
return fmt.Errorf("expected the parts data to be %v, instead got %v", parts[3:], res2.Parts)
}
return nil
})
}
func ListParts_success(s *S3Conf) error {
testName := "ListParts_success"
return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
@@ -5471,7 +5535,7 @@ func ListMultipartUploads_ignore_upload_id_marker(s *S3Conf) error {
}
func ListMultipartUploads_success(s *S3Conf) error {
testName := "ListMultipartUploads_max_uploads"
testName := "ListMultipartUploads_success"
return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
obj1, obj2 := "my-obj-1", "my-obj-2"
out1, err := createMp(s3client, bucket, obj1)