seaweedfs/weed/s3api/s3api_object_multipart_handlers.go

266 lines
7.8 KiB
Go
Raw Normal View History

package s3api
import (
"fmt"
2020-11-11 10:01:24 +00:00
"github.com/chrislusf/seaweedfs/weed/glog"
2020-09-19 21:09:58 +00:00
"github.com/chrislusf/seaweedfs/weed/s3api/s3err"
weed_server "github.com/chrislusf/seaweedfs/weed/server"
2018-09-09 23:26:11 +00:00
"net/http"
"net/url"
"strconv"
"strings"
2020-02-10 01:42:17 +00:00
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/s3"
)
const (
maxObjectListSizeLimit = 10000 // Limit number of objects in a listObjectsResponse.
maxUploadsList = 10000 // Limit number of uploads in a listUploadsResponse.
maxPartsList = 10000 // Limit number of parts in a listPartsResponse.
globalMaxPartID = 100000
)
// NewMultipartUploadHandler - New multipart upload.
func (s3a *S3ApiServer) NewMultipartUploadHandler(w http.ResponseWriter, r *http.Request) {
2020-07-26 19:58:58 +00:00
bucket, object := getBucketAndObject(r)
createMultipartUploadInput := &s3.CreateMultipartUploadInput{
Bucket: aws.String(bucket),
Key: objectKey(aws.String(object)),
Metadata: make(map[string]*string),
}
metadata := weed_server.SaveAmzMetaData(r, nil, false)
for k, v := range metadata {
createMultipartUploadInput.Metadata[k] = aws.String(string(v))
}
response, errCode := s3a.createMultipartUpload(createMultipartUploadInput)
2021-07-15 19:17:48 +00:00
glog.V(2).Info("NewMultipartUploadHandler", string(s3err.EncodeXMLResponse(response)), errCode)
2020-11-11 10:01:24 +00:00
2020-09-19 21:09:58 +00:00
if errCode != s3err.ErrNone {
2021-06-11 04:50:21 +00:00
s3err.WriteErrorResponse(w, errCode, r)
return
}
2021-06-11 04:50:21 +00:00
writeSuccessResponseXML(w, response)
}
// CompleteMultipartUploadHandler - Completes multipart upload.
func (s3a *S3ApiServer) CompleteMultipartUploadHandler(w http.ResponseWriter, r *http.Request) {
2020-07-26 19:58:58 +00:00
bucket, object := getBucketAndObject(r)
// Get upload id.
uploadID, _, _, _ := getObjectResources(r.URL.Query())
2020-02-26 06:23:59 +00:00
response, errCode := s3a.completeMultipartUpload(&s3.CompleteMultipartUploadInput{
2018-09-12 07:46:12 +00:00
Bucket: aws.String(bucket),
2019-07-08 19:37:20 +00:00
Key: objectKey(aws.String(object)),
2018-09-12 07:46:12 +00:00
UploadId: aws.String(uploadID),
})
2021-07-15 19:17:48 +00:00
glog.V(2).Info("CompleteMultipartUploadHandler", string(s3err.EncodeXMLResponse(response)), errCode)
2018-09-12 07:46:12 +00:00
2020-09-19 21:09:58 +00:00
if errCode != s3err.ErrNone {
2021-06-11 04:50:21 +00:00
s3err.WriteErrorResponse(w, errCode, r)
return
}
2021-06-11 04:50:21 +00:00
writeSuccessResponseXML(w, response)
}
// AbortMultipartUploadHandler - Aborts multipart upload.
func (s3a *S3ApiServer) AbortMultipartUploadHandler(w http.ResponseWriter, r *http.Request) {
2020-07-26 19:58:58 +00:00
bucket, object := getBucketAndObject(r)
// Get upload id.
uploadID, _, _, _ := getObjectResources(r.URL.Query())
2020-02-26 06:23:59 +00:00
response, errCode := s3a.abortMultipartUpload(&s3.AbortMultipartUploadInput{
Bucket: aws.String(bucket),
2019-07-08 19:37:20 +00:00
Key: objectKey(aws.String(object)),
UploadId: aws.String(uploadID),
})
2020-09-19 21:09:58 +00:00
if errCode != s3err.ErrNone {
2021-06-11 04:50:21 +00:00
s3err.WriteErrorResponse(w, errCode, r)
return
}
2021-07-15 19:17:48 +00:00
glog.V(2).Info("AbortMultipartUploadHandler", string(s3err.EncodeXMLResponse(response)))
2018-09-12 07:46:12 +00:00
2021-06-11 04:50:21 +00:00
writeSuccessResponseXML(w, response)
}
// ListMultipartUploadsHandler - Lists multipart uploads.
func (s3a *S3ApiServer) ListMultipartUploadsHandler(w http.ResponseWriter, r *http.Request) {
2020-07-26 19:58:58 +00:00
bucket, _ := getBucketAndObject(r)
prefix, keyMarker, uploadIDMarker, delimiter, maxUploads, encodingType := getBucketMultipartResources(r.URL.Query())
if maxUploads < 0 {
2021-06-11 04:50:21 +00:00
s3err.WriteErrorResponse(w, s3err.ErrInvalidMaxUploads, r)
return
}
if keyMarker != "" {
// Marker not common with prefix is not implemented.
if !strings.HasPrefix(keyMarker, prefix) {
2021-06-11 04:50:21 +00:00
s3err.WriteErrorResponse(w, s3err.ErrNotImplemented, r)
return
}
}
2020-02-26 06:23:59 +00:00
response, errCode := s3a.listMultipartUploads(&s3.ListMultipartUploadsInput{
Bucket: aws.String(bucket),
Delimiter: aws.String(delimiter),
EncodingType: aws.String(encodingType),
KeyMarker: aws.String(keyMarker),
MaxUploads: aws.Int64(int64(maxUploads)),
Prefix: aws.String(prefix),
UploadIdMarker: aws.String(uploadIDMarker),
})
2021-07-15 19:17:48 +00:00
glog.V(2).Info("ListMultipartUploadsHandler", string(s3err.EncodeXMLResponse(response)), errCode)
2020-11-11 10:01:24 +00:00
2020-09-19 21:09:58 +00:00
if errCode != s3err.ErrNone {
2021-06-11 04:50:21 +00:00
s3err.WriteErrorResponse(w, errCode, r)
return
}
2018-09-09 23:25:43 +00:00
// TODO handle encodingType
2021-06-11 04:50:21 +00:00
writeSuccessResponseXML(w, response)
}
// ListObjectPartsHandler - Lists object parts in a multipart upload.
func (s3a *S3ApiServer) ListObjectPartsHandler(w http.ResponseWriter, r *http.Request) {
2020-07-26 19:58:58 +00:00
bucket, object := getBucketAndObject(r)
uploadID, partNumberMarker, maxParts, _ := getObjectResources(r.URL.Query())
if partNumberMarker < 0 {
2021-06-11 04:50:21 +00:00
s3err.WriteErrorResponse(w, s3err.ErrInvalidPartNumberMarker, r)
return
}
if maxParts < 0 {
2021-06-11 04:50:21 +00:00
s3err.WriteErrorResponse(w, s3err.ErrInvalidMaxParts, r)
return
}
2020-02-26 06:23:59 +00:00
response, errCode := s3a.listObjectParts(&s3.ListPartsInput{
Bucket: aws.String(bucket),
2019-07-08 19:37:20 +00:00
Key: objectKey(aws.String(object)),
MaxParts: aws.Int64(int64(maxParts)),
PartNumberMarker: aws.Int64(int64(partNumberMarker)),
UploadId: aws.String(uploadID),
})
2021-07-15 19:17:48 +00:00
glog.V(2).Info("ListObjectPartsHandler", string(s3err.EncodeXMLResponse(response)), errCode)
2020-11-11 10:01:24 +00:00
2020-09-19 21:09:58 +00:00
if errCode != s3err.ErrNone {
2021-06-11 04:50:21 +00:00
s3err.WriteErrorResponse(w, errCode, r)
return
}
2021-06-11 04:50:21 +00:00
writeSuccessResponseXML(w, response)
}
// PutObjectPartHandler - Put an object part in a multipart upload.
func (s3a *S3ApiServer) PutObjectPartHandler(w http.ResponseWriter, r *http.Request) {
2020-07-26 19:58:58 +00:00
bucket, _ := getBucketAndObject(r)
uploadID := r.URL.Query().Get("uploadId")
2020-02-26 06:23:59 +00:00
exists, err := s3a.exists(s3a.genUploadsFolder(bucket), uploadID, true)
if !exists {
2021-06-11 04:50:21 +00:00
s3err.WriteErrorResponse(w, s3err.ErrNoSuchUpload, r)
return
}
partIDString := r.URL.Query().Get("partNumber")
partID, err := strconv.Atoi(partIDString)
if err != nil {
2021-06-11 04:50:21 +00:00
s3err.WriteErrorResponse(w, s3err.ErrInvalidPart, r)
return
}
if partID > globalMaxPartID {
2021-06-11 04:50:21 +00:00
s3err.WriteErrorResponse(w, s3err.ErrInvalidMaxParts, r)
return
}
dataReader := r.Body
if s3a.iam.isEnabled() {
rAuthType := getRequestAuthType(r)
2020-09-19 21:09:58 +00:00
var s3ErrCode s3err.ErrorCode
switch rAuthType {
case authTypeStreamingSigned:
dataReader, s3ErrCode = s3a.iam.newSignV4ChunkedReader(r)
case authTypeSignedV2, authTypePresignedV2:
_, s3ErrCode = s3a.iam.isReqAuthenticatedV2(r)
case authTypePresigned, authTypeSigned:
_, s3ErrCode = s3a.iam.reqSignatureV4Verify(r)
}
2020-09-19 21:09:58 +00:00
if s3ErrCode != s3err.ErrNone {
2021-06-11 04:50:21 +00:00
s3err.WriteErrorResponse(w, s3ErrCode, r)
return
}
}
2020-02-14 17:09:15 +00:00
defer dataReader.Close()
uploadUrl := fmt.Sprintf("http://%s%s/%s/%04d.part?collection=%s",
2020-09-21 17:51:24 +00:00
s3a.option.Filer, s3a.genUploadsFolder(bucket), uploadID, partID, bucket)
etag, errCode := s3a.putToFiler(r, uploadUrl, dataReader)
2020-09-19 21:09:58 +00:00
if errCode != s3err.ErrNone {
2021-06-11 04:50:21 +00:00
s3err.WriteErrorResponse(w, errCode, r)
return
}
setEtag(w, etag)
writeSuccessResponseEmpty(w)
}
func (s3a *S3ApiServer) genUploadsFolder(bucket string) string {
2018-09-20 05:03:16 +00:00
return fmt.Sprintf("%s/%s/.uploads", s3a.option.BucketsPath, bucket)
}
// Parse bucket url queries for ?uploads
func getBucketMultipartResources(values url.Values) (prefix, keyMarker, uploadIDMarker, delimiter string, maxUploads int, encodingType string) {
prefix = values.Get("prefix")
keyMarker = values.Get("key-marker")
uploadIDMarker = values.Get("upload-id-marker")
delimiter = values.Get("delimiter")
if values.Get("max-uploads") != "" {
maxUploads, _ = strconv.Atoi(values.Get("max-uploads"))
} else {
maxUploads = maxUploadsList
}
encodingType = values.Get("encoding-type")
return
}
// Parse object url queries
func getObjectResources(values url.Values) (uploadID string, partNumberMarker, maxParts int, encodingType string) {
uploadID = values.Get("uploadId")
partNumberMarker, _ = strconv.Atoi(values.Get("part-number-marker"))
if values.Get("max-parts") != "" {
maxParts, _ = strconv.Atoi(values.Get("max-parts"))
} else {
maxParts = maxPartsList
}
encodingType = values.Get("encoding-type")
return
}
type byCompletedPartNumber []*s3.CompletedPart
func (a byCompletedPartNumber) Len() int { return len(a) }
func (a byCompletedPartNumber) Swap(i, j int) { a[i], a[j] = a[j], a[i] }
func (a byCompletedPartNumber) Less(i, j int) bool { return *a[i].PartNumber < *a[j].PartNumber }