diff options
Diffstat (limited to 'weed/s3api/filer_multipart.go')
| -rw-r--r-- | weed/s3api/filer_multipart.go | 218 |
1 files changed, 146 insertions, 72 deletions
diff --git a/weed/s3api/filer_multipart.go b/weed/s3api/filer_multipart.go index 765a5679e..e9cd6a0c4 100644 --- a/weed/s3api/filer_multipart.go +++ b/weed/s3api/filer_multipart.go @@ -1,11 +1,12 @@ package s3api import ( + "cmp" "encoding/hex" "encoding/xml" "fmt" - "github.com/google/uuid" - "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/stats" "golang.org/x/exp/slices" "math" "path/filepath" @@ -16,12 +17,19 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/s3" + "github.com/google/uuid" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" ) +const ( + multipartExt = ".part" + multiPartMinSize = 5 * 1024 * 1024 +) + type InitiateMultipartUploadResult struct { XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ InitiateMultipartUploadResult"` s3.CreateMultipartUploadOutput @@ -70,61 +78,129 @@ type CompleteMultipartUploadResult struct { func (s3a *S3ApiServer) completeMultipartUpload(input *s3.CompleteMultipartUploadInput, parts *CompleteMultipartUpload) (output *CompleteMultipartUploadResult, code s3err.ErrorCode) { glog.V(2).Infof("completeMultipartUpload input %v", input) - - completedParts := parts.Parts - slices.SortFunc(completedParts, func(a, b CompletedPart) int { - return a.PartNumber - b.PartNumber - }) + if len(parts.Parts) == 0 { + stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedNoSuchUpload).Inc() + return nil, s3err.ErrNoSuchUpload + } + completedPartNumbers := []int{} + completedPartMap := make(map[int][]string) + for _, part := range parts.Parts { + if _, ok := completedPartMap[part.PartNumber]; !ok { + completedPartNumbers = append(completedPartNumbers, part.PartNumber) + } + completedPartMap[part.PartNumber] = append(completedPartMap[part.PartNumber], part.ETag) + } + sort.Ints(completedPartNumbers) uploadDirectory := s3a.genUploadsFolder(*input.Bucket) + "/" + *input.UploadId - entries, _, err := s3a.list(uploadDirectory, "", "", false, maxPartsList) - if err != nil || len(entries) == 0 { + if err != nil { glog.Errorf("completeMultipartUpload %s %s error: %v, entries:%d", *input.Bucket, *input.UploadId, err, len(entries)) + stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedNoSuchUpload).Inc() + return nil, s3err.ErrNoSuchUpload + } + + if len(entries) == 0 { + entryName, dirName := s3a.getEntryNameAndDir(input) + if entry, _ := s3a.getEntry(dirName, entryName); entry != nil && entry.Extended != nil { + if uploadId, ok := entry.Extended[s3_constants.X_SeaweedFS_Header_Upload_Id]; ok && *input.UploadId == string(uploadId) { + return &CompleteMultipartUploadResult{ + CompleteMultipartUploadOutput: s3.CompleteMultipartUploadOutput{ + Location: aws.String(fmt.Sprintf("http://%s%s/%s", s3a.option.Filer.ToHttpAddress(), urlEscapeObject(dirName), urlPathEscape(entryName))), + Bucket: input.Bucket, + ETag: aws.String("\"" + filer.ETagChunks(entry.GetChunks()) + "\""), + Key: objectKey(input.Key), + }, + }, s3err.ErrNone + } + } + stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedNoSuchUpload).Inc() return nil, s3err.ErrNoSuchUpload } 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) + stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedNoSuchUpload).Inc() 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{}{} + deleteEntries := []*filer_pb.Entry{} + partEntries := make(map[int][]*filer_pb.Entry, len(entries)) + entityTooSmall := false + for _, entry := range entries { + foundEntry := false + glog.V(4).Infof("completeMultipartUpload part entries %s", entry.Name) + if entry.IsDirectory || !strings.HasSuffix(entry.Name, multipartExt) { + continue + } + partNumber, err := parsePartNumber(entry.Name) + if err != nil { + stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedPartNumber).Inc() + glog.Errorf("completeMultipartUpload failed to pasre partNumber %s:%s", entry.Name, err) + continue + } + completedPartsByNumber, ok := completedPartMap[partNumber] + if !ok { + continue + } + for _, partETag := range completedPartsByNumber { + partETag = strings.Trim(partETag, `"`) + entryETag := hex.EncodeToString(entry.Attributes.GetMd5()) + if partETag != "" && len(partETag) == 32 && entryETag != "" { + if entryETag != partETag { + glog.Errorf("completeMultipartUpload %s ETag mismatch chunk: %s part: %s", entry.Name, entryETag, partETag) + stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedEtagMismatch).Inc() + continue } + } else { + glog.Warningf("invalid complete etag %s, partEtag %s", partETag, entryETag) + stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedEtagInvalid).Inc() + } + if len(entry.Chunks) == 0 { + glog.Warningf("completeMultipartUpload %s empty chunks", entry.Name) + stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedPartEmpty).Inc() + continue } + //there maybe multi same part, because of client retry + partEntries[partNumber] = append(partEntries[partNumber], entry) + foundEntry = true } - for _, part := range completedParts { - if _, found := partNumbers[part.PartNumber]; !found { - return nil, s3err.ErrInvalidPart + if foundEntry { + if len(completedPartNumbers) > 1 && partNumber != completedPartNumbers[len(completedPartNumbers)-1] && + entry.Attributes.FileSize < multiPartMinSize { + glog.Warningf("completeMultipartUpload %s part file size less 5mb", entry.Name) + entityTooSmall = true } + } else { + deleteEntries = append(deleteEntries, entry) } } - + if entityTooSmall { + stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompleteEntityTooSmall).Inc() + return nil, s3err.ErrEntityTooSmall + } mime := pentry.Attributes.Mime - 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 { + for _, partNumber := range completedPartNumbers { + partEntriesByNumber, ok := partEntries[partNumber] + if !ok { + glog.Errorf("part %d has no entry", partNumber) + stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedPartNotFound).Inc() + return nil, s3err.ErrInvalidPart + } + found := false + if len(partEntriesByNumber) > 1 { + slices.SortFunc(partEntriesByNumber, func(a, b *filer_pb.Entry) int { + return cmp.Compare(b.Chunks[0].ModifiedTsNs, a.Chunks[0].ModifiedTsNs) + }) + } + for _, entry := range partEntriesByNumber { + if found { + deleteEntries = append(deleteEntries, entry) + stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedPartEntryMismatch).Inc() continue } - 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() { p := &filer_pb.FileChunk{ FileId: chunk.GetFileIdString(), @@ -137,28 +213,16 @@ func (s3a *S3ApiServer) completeMultipartUpload(input *s3.CompleteMultipartUploa finalParts = append(finalParts, p) offset += int64(chunk.Size) } + found = true } } - entryName := filepath.Base(*input.Key) - dirName := filepath.ToSlash(filepath.Dir(*input.Key)) - if dirName == "." { - dirName = "" - } - if strings.HasPrefix(dirName, "/") { - dirName = dirName[1:] - } - dirName = fmt.Sprintf("%s/%s/%s", s3a.option.BucketsPath, *input.Bucket, dirName) - - // remove suffix '/' - if strings.HasSuffix(dirName, "/") { - dirName = dirName[:len(dirName)-1] - } - + entryName, dirName := s3a.getEntryNameAndDir(input) err = s3a.mkFile(dirName, entryName, finalParts, func(entry *filer_pb.Entry) { if entry.Extended == nil { entry.Extended = make(map[string][]byte) } + entry.Extended[s3_constants.X_SeaweedFS_Header_Upload_Id] = []byte(*input.UploadId) for k, v := range pentry.Extended { if k != "key" { entry.Extended[k] = v @@ -186,6 +250,13 @@ func (s3a *S3ApiServer) completeMultipartUpload(input *s3.CompleteMultipartUploa }, } + for _, deleteEntry := range deleteEntries { + //delete unused part data + glog.Infof("completeMultipartUpload cleanup %s upload %s unused %s", *input.Bucket, *input.UploadId, deleteEntry.Name) + if err = s3a.rm(uploadDirectory, deleteEntry.Name, true, true); err != nil { + glog.Warningf("completeMultipartUpload cleanup %s upload %s unused %s : %v", *input.Bucket, *input.UploadId, deleteEntry.Name, err) + } + } if err = s3a.rm(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, false, true); err != nil { glog.V(1).Infof("completeMultipartUpload cleanup %s upload %s: %v", *input.Bucket, *input.UploadId, err) } @@ -193,29 +264,33 @@ func (s3a *S3ApiServer) completeMultipartUpload(input *s3.CompleteMultipartUploa return } -func findByPartNumber(fileName string, parts []CompletedPart) (etag string, found bool) { - partNumber, formatErr := strconv.Atoi(fileName[:4]) - if formatErr != nil { - return +func (s3a *S3ApiServer) getEntryNameAndDir(input *s3.CompleteMultipartUploadInput) (string, string) { + entryName := filepath.Base(*input.Key) + dirName := filepath.ToSlash(filepath.Dir(*input.Key)) + if dirName == "." { + dirName = "" } - x := sort.Search(len(parts), func(i int) bool { - return parts[i].PartNumber >= partNumber - }) - if x >= len(parts) { - return + if strings.HasPrefix(dirName, "/") { + dirName = dirName[1:] } - if parts[x].PartNumber != partNumber { - return + dirName = fmt.Sprintf("%s/%s/%s", s3a.option.BucketsPath, *input.Bucket, dirName) + + // remove suffix '/' + if strings.HasSuffix(dirName, "/") { + dirName = dirName[:len(dirName)-1] } - y := 0 - for i, part := range parts[x:] { - if part.PartNumber == partNumber { - y = i - } else { - break - } + return entryName, dirName +} + +func parsePartNumber(fileName string) (int, error) { + var partNumberString string + index := strings.Index(fileName, "_") + if index != -1 { + partNumberString = fileName[:index] + } else { + partNumberString = fileName[:len(fileName)-len(multipartExt)] } - return parts[x+y].ETag, true + return strconv.Atoi(partNumberString) } func (s3a *S3ApiServer) abortMultipartUpload(input *s3.AbortMultipartUploadInput) (output *s3.AbortMultipartUploadOutput, code s3err.ErrorCode) { @@ -331,7 +406,7 @@ func (s3a *S3ApiServer) listObjectParts(input *s3.ListPartsInput) (output *ListP StorageClass: aws.String("STANDARD"), } - entries, isLast, err := s3a.list(s3a.genUploadsFolder(*input.Bucket)+"/"+*input.UploadId, "", fmt.Sprintf("%04d.part", *input.PartNumberMarker), false, uint32(*input.MaxParts)) + entries, isLast, err := s3a.list(s3a.genUploadsFolder(*input.Bucket)+"/"+*input.UploadId, "", fmt.Sprintf("%04d%s", *input.PartNumberMarker, multipartExt), false, uint32(*input.MaxParts)) if err != nil { glog.Errorf("listObjectParts %s %s error: %v", *input.Bucket, *input.UploadId, err) return nil, s3err.ErrNoSuchUpload @@ -343,9 +418,8 @@ func (s3a *S3ApiServer) listObjectParts(input *s3.ListPartsInput) (output *ListP output.IsTruncated = aws.Bool(!isLast) 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 strings.HasSuffix(entry.Name, multipartExt) && !entry.IsDirectory { + partNumber, err := parsePartNumber(entry.Name) if err != nil { glog.Errorf("listObjectParts %s %s parse %s: %v", *input.Bucket, *input.UploadId, entry.Name, err) continue |