seaweedfs/weed/s3api/filer_multipart.go

367 lines
12 KiB
Go
Raw Normal View History

package s3api
import (
"encoding/hex"
2019-02-27 08:21:37 +00:00
"encoding/xml"
2018-09-09 23:25:43 +00:00
"fmt"
2022-11-10 15:17:29 +00:00
"github.com/google/uuid"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
"golang.org/x/exp/slices"
"math"
2018-09-09 23:25:43 +00:00
"path/filepath"
"sort"
2018-09-09 23:26:11 +00:00
"strconv"
"strings"
"time"
2018-09-09 23:25:43 +00:00
"github.com/aws/aws-sdk-go/aws"
2018-09-09 23:26:11 +00:00
"github.com/aws/aws-sdk-go/service/s3"
2020-02-26 06:23:59 +00:00
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
)
2018-09-12 07:46:12 +00:00
type InitiateMultipartUploadResult struct {
2019-02-27 08:21:37 +00:00
XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ InitiateMultipartUploadResult"`
2018-09-12 07:46:12 +00:00
s3.CreateMultipartUploadOutput
}
2020-09-19 21:09:58 +00:00
func (s3a *S3ApiServer) createMultipartUpload(input *s3.CreateMultipartUploadInput) (output *InitiateMultipartUploadResult, code s3err.ErrorCode) {
2020-09-21 17:51:24 +00:00
glog.V(2).Infof("createMultipartUpload input %v", input)
uploadIdString := s3a.generateUploadID(*input.Key)
uploadIdString = uploadIdString + "_" + strings.ReplaceAll(uuid.New().String(), "-", "")
2022-11-10 15:17:29 +00:00
2020-02-26 06:23:59 +00:00
if err := s3a.mkdir(s3a.genUploadsFolder(*input.Bucket), uploadIdString, func(entry *filer_pb.Entry) {
if entry.Extended == nil {
2018-09-07 20:12:52 +00:00
entry.Extended = make(map[string][]byte)
}
2018-09-07 20:12:52 +00:00
entry.Extended["key"] = []byte(*input.Key)
for k, v := range input.Metadata {
entry.Extended[k] = []byte(*v)
}
if input.ContentType != nil {
entry.Attributes.Mime = *input.ContentType
}
}); err != nil {
glog.Errorf("NewMultipartUpload error: %v", err)
2020-09-19 21:09:58 +00:00
return nil, s3err.ErrInternalError
}
2018-09-12 07:46:12 +00:00
output = &InitiateMultipartUploadResult{
2019-02-27 08:21:37 +00:00
CreateMultipartUploadOutput: s3.CreateMultipartUploadOutput{
2018-09-12 07:46:12 +00:00
Bucket: input.Bucket,
2019-07-08 19:37:20 +00:00
Key: objectKey(input.Key),
2018-09-12 07:46:12 +00:00
UploadId: aws.String(uploadIdString),
},
}
return
}
2018-09-12 07:46:12 +00:00
type CompleteMultipartUploadResult struct {
2019-02-27 08:21:37 +00:00
XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ CompleteMultipartUploadResult"`
2018-09-12 07:46:12 +00:00
s3.CompleteMultipartUploadOutput
}
func (s3a *S3ApiServer) completeMultipartUpload(input *s3.CompleteMultipartUploadInput, parts *CompleteMultipartUpload) (output *CompleteMultipartUploadResult, code s3err.ErrorCode) {
2018-09-09 23:25:43 +00:00
2020-09-21 17:51:24 +00:00
glog.V(2).Infof("completeMultipartUpload input %v", input)
completedParts := parts.Parts
slices.SortFunc(completedParts, func(a, b CompletedPart) bool {
return a.PartNumber < b.PartNumber
})
2018-09-09 23:25:43 +00:00
uploadDirectory := s3a.genUploadsFolder(*input.Bucket) + "/" + *input.UploadId
entries, _, err := s3a.list(uploadDirectory, "", "", false, maxPartsList)
if err != nil || len(entries) == 0 {
glog.Errorf("completeMultipartUpload %s %s error: %v, entries:%d", *input.Bucket, *input.UploadId, err, len(entries))
2020-09-19 21:09:58 +00:00
return nil, s3err.ErrNoSuchUpload
2018-09-09 23:25:43 +00:00
}
2021-07-21 21:38:12 +00:00
pentry, err := s3a.getEntry(s3a.genUploadsFolder(*input.Bucket), *input.UploadId)
if err != nil {
glog.Errorf("completeMultipartUpload %s %s error: %v", *input.Bucket, *input.UploadId, err)
return nil, s3err.ErrNoSuchUpload
}
// check whether completedParts is more than received parts
{
partNumbers := make(map[int]struct{}, len(entries))
for _, entry := range entries {
if strings.HasSuffix(entry.Name, ".part") && !entry.IsDirectory {
partNumberString := entry.Name[:len(entry.Name)-len(".part")]
partNumber, err := strconv.Atoi(partNumberString)
if err == nil {
partNumbers[partNumber] = struct{}{}
}
}
}
for _, part := range completedParts {
if _, found := partNumbers[part.PartNumber]; !found {
return nil, s3err.ErrInvalidPart
}
}
}
mime := pentry.Attributes.Mime
2018-09-09 23:25:43 +00:00
var finalParts []*filer_pb.FileChunk
var offset int64
for _, entry := range entries {
if strings.HasSuffix(entry.Name, ".part") && !entry.IsDirectory {
partETag, found := findByPartNumber(entry.Name, completedParts)
if !found {
continue
2021-10-14 10:03:11 +00:00
}
entryETag := hex.EncodeToString(entry.Attributes.GetMd5())
if partETag != "" && len(partETag) == 32 && entryETag != "" && entryETag != partETag {
glog.Errorf("completeMultipartUpload %s ETag mismatch chunk: %s part: %s", entry.Name, entryETag, partETag)
return nil, s3err.ErrInvalidPart
}
for _, chunk := range entry.GetChunks() {
2018-09-12 07:46:12 +00:00
p := &filer_pb.FileChunk{
FileId: chunk.GetFileIdString(),
Offset: offset,
Size: chunk.Size,
ModifiedTsNs: chunk.ModifiedTsNs,
CipherKey: chunk.CipherKey,
ETag: chunk.ETag,
2018-09-12 07:46:12 +00:00
}
finalParts = append(finalParts, p)
2018-09-09 23:25:43 +00:00
offset += int64(chunk.Size)
}
}
}
2018-09-09 23:25:43 +00:00
entryName := filepath.Base(*input.Key)
dirName := filepath.ToSlash(filepath.Dir(*input.Key))
2018-09-09 23:25:43 +00:00
if dirName == "." {
dirName = ""
}
2018-09-12 07:46:12 +00:00
if strings.HasPrefix(dirName, "/") {
dirName = dirName[1:]
}
2018-09-09 23:25:43 +00:00
dirName = fmt.Sprintf("%s/%s/%s", s3a.option.BucketsPath, *input.Bucket, dirName)
// remove suffix '/'
if strings.HasSuffix(dirName, "/") {
dirName = dirName[:len(dirName)-1]
}
2021-07-21 21:38:12 +00:00
err = s3a.mkFile(dirName, entryName, finalParts, func(entry *filer_pb.Entry) {
if entry.Extended == nil {
entry.Extended = make(map[string][]byte)
}
2021-07-21 21:38:12 +00:00
for k, v := range pentry.Extended {
if k != "key" {
entry.Extended[k] = v
}
}
if pentry.Attributes.Mime != "" {
entry.Attributes.Mime = pentry.Attributes.Mime
2021-10-14 10:03:11 +00:00
} else if mime != "" {
entry.Attributes.Mime = mime
}
2022-09-15 08:27:02 +00:00
entry.Attributes.FileSize = uint64(offset)
})
2018-09-09 23:25:43 +00:00
if err != nil {
glog.Errorf("completeMultipartUpload %s/%s error: %v", dirName, entryName, err)
2020-09-19 21:09:58 +00:00
return nil, s3err.ErrInternalError
2018-09-09 23:25:43 +00:00
}
2018-09-12 07:46:12 +00:00
output = &CompleteMultipartUploadResult{
2019-02-27 08:21:37 +00:00
CompleteMultipartUploadOutput: s3.CompleteMultipartUploadOutput{
Location: aws.String(fmt.Sprintf("http://%s%s/%s", s3a.option.Filer.ToHttpAddress(), urlEscapeObject(dirName), urlPathEscape(entryName))),
2019-07-22 04:51:38 +00:00
Bucket: input.Bucket,
2020-09-01 07:21:19 +00:00
ETag: aws.String("\"" + filer.ETagChunks(finalParts) + "\""),
2019-07-22 04:51:38 +00:00
Key: objectKey(input.Key),
2018-09-12 07:46:12 +00:00
},
2018-09-09 23:25:43 +00:00
}
if err = s3a.rm(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, false, true); err != nil {
2018-09-12 08:00:51 +00:00
glog.V(1).Infof("completeMultipartUpload cleanup %s upload %s: %v", *input.Bucket, *input.UploadId, err)
}
return
}
func findByPartNumber(fileName string, parts []CompletedPart) (etag string, found bool) {
partNumber, formatErr := strconv.Atoi(fileName[:4])
if formatErr != nil {
return
}
2022-03-29 14:53:12 +00:00
x := sort.Search(len(parts), func(i int) bool {
return parts[i].PartNumber >= partNumber
})
if x >= len(parts) {
return
}
2022-03-29 14:53:12 +00:00
if parts[x].PartNumber != partNumber {
return
}
y := 0
for i, part := range parts[x:] {
if part.PartNumber == partNumber {
2022-03-29 14:53:12 +00:00
y = i
} else {
break
}
}
2022-03-29 14:53:12 +00:00
return parts[x+y].ETag, true
}
2020-09-19 21:09:58 +00:00
func (s3a *S3ApiServer) abortMultipartUpload(input *s3.AbortMultipartUploadInput) (output *s3.AbortMultipartUploadOutput, code s3err.ErrorCode) {
2020-09-21 17:51:24 +00:00
glog.V(2).Infof("abortMultipartUpload input %v", input)
2020-02-26 06:23:59 +00:00
exists, err := s3a.exists(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, true)
if err != nil {
2018-09-09 23:25:43 +00:00
glog.V(1).Infof("bucket %s abort upload %s: %v", *input.Bucket, *input.UploadId, err)
2020-09-19 21:09:58 +00:00
return nil, s3err.ErrNoSuchUpload
}
2018-09-09 23:25:43 +00:00
if exists {
err = s3a.rm(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, true, true)
2018-09-09 23:25:43 +00:00
}
if err != nil {
glog.V(1).Infof("bucket %s remove upload %s: %v", *input.Bucket, *input.UploadId, err)
2020-09-19 21:09:58 +00:00
return nil, s3err.ErrInternalError
2018-09-09 23:25:43 +00:00
}
2020-09-19 21:09:58 +00:00
return &s3.AbortMultipartUploadOutput{}, s3err.ErrNone
2018-09-09 23:25:43 +00:00
}
2018-09-12 07:46:12 +00:00
type ListMultipartUploadsResult struct {
2019-02-27 08:21:37 +00:00
XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ ListMultipartUploadsResult"`
// copied from s3.ListMultipartUploadsOutput, the Uploads is not converting to <Upload></Upload>
Bucket *string `type:"string"`
Delimiter *string `type:"string"`
EncodingType *string `type:"string" enum:"EncodingType"`
IsTruncated *bool `type:"boolean"`
KeyMarker *string `type:"string"`
MaxUploads *int64 `type:"integer"`
NextKeyMarker *string `type:"string"`
NextUploadIdMarker *string `type:"string"`
Prefix *string `type:"string"`
UploadIdMarker *string `type:"string"`
Upload []*s3.MultipartUpload `locationName:"Upload" type:"list" flattened:"true"`
2018-09-12 07:46:12 +00:00
}
2020-09-19 21:09:58 +00:00
func (s3a *S3ApiServer) listMultipartUploads(input *s3.ListMultipartUploadsInput) (output *ListMultipartUploadsResult, code s3err.ErrorCode) {
// https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListMultipartUploads.html
2018-09-09 23:25:43 +00:00
2020-09-21 17:51:24 +00:00
glog.V(2).Infof("listMultipartUploads input %v", input)
2018-09-12 07:46:12 +00:00
output = &ListMultipartUploadsResult{
Bucket: input.Bucket,
Delimiter: input.Delimiter,
EncodingType: input.EncodingType,
KeyMarker: input.KeyMarker,
MaxUploads: input.MaxUploads,
Prefix: input.Prefix,
IsTruncated: aws.Bool(false),
}
2018-09-09 23:25:43 +00:00
entries, _, err := s3a.list(s3a.genUploadsFolder(*input.Bucket), "", *input.UploadIdMarker, false, math.MaxInt32)
2018-09-09 23:25:43 +00:00
if err != nil {
glog.Errorf("listMultipartUploads %s error: %v", *input.Bucket, err)
return
}
uploadsCount := int64(0)
for _, entry := range entries {
if entry.Extended != nil {
2020-09-21 17:51:24 +00:00
key := string(entry.Extended["key"])
if *input.KeyMarker != "" && *input.KeyMarker != key {
continue
}
if *input.Prefix != "" && !strings.HasPrefix(key, *input.Prefix) {
continue
}
output.Upload = append(output.Upload, &s3.MultipartUpload{
2020-09-21 17:51:24 +00:00
Key: objectKey(aws.String(key)),
UploadId: aws.String(entry.Name),
})
uploadsCount += 1
}
if uploadsCount >= *input.MaxUploads {
output.IsTruncated = aws.Bool(true)
output.NextUploadIdMarker = aws.String(entry.Name)
break
}
}
2018-09-12 08:00:51 +00:00
return
}
2018-09-12 07:46:12 +00:00
type ListPartsResult struct {
2019-02-27 08:21:37 +00:00
XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ ListPartsResult"`
// copied from s3.ListPartsOutput, the Parts is not converting to <Part></Part>
Bucket *string `type:"string"`
IsTruncated *bool `type:"boolean"`
Key *string `min:"1" type:"string"`
MaxParts *int64 `type:"integer"`
NextPartNumberMarker *int64 `type:"integer"`
PartNumberMarker *int64 `type:"integer"`
Part []*s3.Part `locationName:"Part" type:"list" flattened:"true"`
StorageClass *string `type:"string" enum:"StorageClass"`
UploadId *string `type:"string"`
2018-09-12 07:46:12 +00:00
}
2020-09-19 21:09:58 +00:00
func (s3a *S3ApiServer) listObjectParts(input *s3.ListPartsInput) (output *ListPartsResult, code s3err.ErrorCode) {
2020-09-11 22:07:19 +00:00
// https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListParts.html
2020-09-21 17:51:24 +00:00
glog.V(2).Infof("listObjectParts input %v", input)
2018-09-12 07:46:12 +00:00
output = &ListPartsResult{
Bucket: input.Bucket,
Key: objectKey(input.Key),
UploadId: input.UploadId,
MaxParts: input.MaxParts, // the maximum number of parts to return.
PartNumberMarker: input.PartNumberMarker, // the part number starts after this, exclusive
StorageClass: aws.String("STANDARD"),
2018-09-09 23:25:43 +00:00
}
entries, isLast, err := s3a.list(s3a.genUploadsFolder(*input.Bucket)+"/"+*input.UploadId, "", fmt.Sprintf("%04d.part", *input.PartNumberMarker), false, uint32(*input.MaxParts))
2018-09-09 23:25:43 +00:00
if err != nil {
2018-09-11 20:01:51 +00:00
glog.Errorf("listObjectParts %s %s error: %v", *input.Bucket, *input.UploadId, err)
2020-09-19 21:09:58 +00:00
return nil, s3err.ErrNoSuchUpload
2018-09-09 23:25:43 +00:00
}
2022-02-19 06:14:40 +00:00
// Note: The upload directory is sort of a marker of the existence of an multipart upload request.
// So can not just delete empty upload folders.
output.IsTruncated = aws.Bool(!isLast)
2018-09-09 23:25:43 +00:00
for _, entry := range entries {
if strings.HasSuffix(entry.Name, ".part") && !entry.IsDirectory {
partNumberString := entry.Name[:len(entry.Name)-len(".part")]
partNumber, err := strconv.Atoi(partNumberString)
if err != nil {
2018-09-11 20:01:51 +00:00
glog.Errorf("listObjectParts %s %s parse %s: %v", *input.Bucket, *input.UploadId, entry.Name, err)
2018-09-09 23:25:43 +00:00
continue
}
output.Part = append(output.Part, &s3.Part{
2018-09-09 23:25:43 +00:00
PartNumber: aws.Int64(int64(partNumber)),
LastModified: aws.Time(time.Unix(entry.Attributes.Mtime, 0).UTC()),
2020-09-01 07:21:19 +00:00
Size: aws.Int64(int64(filer.FileSize(entry))),
ETag: aws.String("\"" + filer.ETag(entry) + "\""),
2018-09-09 23:25:43 +00:00
})
if !isLast {
output.NextPartNumberMarker = aws.Int64(int64(partNumber))
}
2018-09-09 23:25:43 +00:00
}
}
return
}