gsh-digital-services/seaweedFS
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

preparing to support S3 multipart uploads

Browse Source
This commit is contained in:
Chris Lu
2018-09-04 00:42:44 -07:00
parent 7f760f16b0
commit ce6a2140a2
8 changed files with 543 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
weed/s3api/filer_multipart.go Normal file
View File

@@ -0,0 +1,70 @@
package s3api
import (
"github.com/aws/aws-sdk-go/service/s3"
"github.com/aws/aws-sdk-go/aws"
"github.com/satori/go.uuid"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
)
func (s3a *S3ApiServer) createMultipartUpload(input *s3.CreateMultipartUploadInput) (output *s3.CreateMultipartUploadOutput, code ErrorCode) {
uploadId, _ := uuid.NewV4()
uploadIdString := uploadId.String()
if err := s3a.mkdir(s3a.genUploadsFolder(*input.Bucket), uploadIdString, func(entry *filer_pb.Entry) {
if entry.Extended == nil {
entry.Extended = make(map[string]string)
}
entry.Extended["key"] = *input.Key
}); err != nil {
glog.Errorf("NewMultipartUpload error: %v", err)
return nil, ErrInternalError
}
output = &s3.CreateMultipartUploadOutput{
Bucket: input.Bucket,
Key: input.Key,
UploadId: aws.String(uploadIdString),
}
return
}
func (s3a *S3ApiServer) completeMultipartUpload(input *s3.CompleteMultipartUploadInput) (output *s3.CompleteMultipartUploadOutput, code ErrorCode) {
return
}
func (s3a *S3ApiServer) abortMultipartUpload(input *s3.AbortMultipartUploadInput) (output *s3.AbortMultipartUploadOutput, code ErrorCode) {
return
}
func (s3a *S3ApiServer) listMultipartUploads(input *s3.ListMultipartUploadsInput) (output *s3.ListMultipartUploadsOutput, code ErrorCode) {
entries, err := s3a.list(s3a.genUploadsFolder(*input.Bucket))
if err != nil {
glog.Errorf("listMultipartUploads %s error: %v", *input.Bucket, err)
return nil, ErrNoSuchUpload
}
output = &s3.ListMultipartUploadsOutput{
Bucket: input.Bucket,
Delimiter: input.Delimiter,
EncodingType: input.EncodingType,
KeyMarker: input.KeyMarker,
MaxUploads: input.MaxUploads,
Prefix: input.Prefix,
}
for _, entry := range entries {
if entry.Extended != nil {
key := entry.Extended["key"]
output.Uploads = append(output.Uploads, &s3.MultipartUpload{
Key: aws.String(key),
UploadId: aws.String(entry.Name),
})
}
}
return
}
func (s3a *S3ApiServer) listObjectParts(input *s3.ListPartsInput) (output *s3.ListPartsOutput, code ErrorCode) {
return
}

55
weed/s3api/filer_util.go
View File

@@ -10,22 +10,28 @@ import (
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
)
func (s3a *S3ApiServer) mkdir(parentDirectoryPath string, dirName string) error {
func (s3a *S3ApiServer) mkdir(parentDirectoryPath string, dirName string, fn func(entry *filer_pb.Entry)) error {
return s3a.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
entry := &filer_pb.Entry{
Name: dirName,
IsDirectory: true,
Attributes: &filer_pb.FuseAttributes{
Mtime: time.Now().Unix(),
Crtime: time.Now().Unix(),
FileMode: uint32(0777 | os.ModeDir),
Uid: OS_UID,
Gid: OS_GID,
},
}
if fn != nil {
fn(entry)
}
request := &filer_pb.CreateEntryRequest{
Directory: parentDirectoryPath,
Entry: &filer_pb.Entry{
Name: dirName,
IsDirectory: true,
Attributes: &filer_pb.FuseAttributes{
Mtime: time.Now().Unix(),
Crtime: time.Now().Unix(),
FileMode: uint32(0777 | os.ModeDir),
Uid: OS_UID,
Gid: OS_GID,
},
},
Entry: entry,
}
glog.V(1).Infof("create bucket: %v", request)
@@ -83,3 +89,28 @@ func (s3a *S3ApiServer) rm(parentDirectoryPath string, entryName string, isDirec
})
}
func (s3a *S3ApiServer) exists(parentDirectoryPath string, entryName string, isDirectory bool) (exists bool, err error) {
err = s3a.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
ctx := context.Background()
request := &filer_pb.LookupDirectoryEntryRequest{
Directory: parentDirectoryPath,
Name: entryName,
}
glog.V(1).Infof("delete entry %v/%v: %v", parentDirectoryPath, entryName, request)
resp, err := client.LookupDirectoryEntry(ctx, request)
if err != nil {
return fmt.Errorf("delete entry %s/%s: %v", parentDirectoryPath, entryName, err)
}
exists = resp.Entry.IsDirectory == isDirectory
return nil
})
return
}

42
weed/s3api/s3api_errors.go
View File

@@ -32,10 +32,17 @@ const (
ErrBucketAlreadyExists
ErrBucketAlreadyOwnedByYou
ErrNoSuchBucket
ErrNoSuchUpload
ErrInvalidBucketName
ErrInvalidDigest
ErrInvalidMaxKeys
ErrInvalidMaxUploads
ErrInvalidMaxParts
ErrInvalidPartNumberMarker
ErrInvalidPart
ErrInvalidPartOrder
ErrInternalError
ErrMalformedXML
ErrNotImplemented
)
@@ -72,21 +79,56 @@ var errorCodeResponse = map[ErrorCode]APIError{
Description: "The Content-Md5 you specified is not valid.",
HTTPStatusCode: http.StatusBadRequest,
},
ErrInvalidMaxUploads: {
Code: "InvalidArgument",
Description: "Argument max-uploads must be an integer between 0 and 2147483647",
HTTPStatusCode: http.StatusBadRequest,
},
ErrInvalidMaxKeys: {
Code: "InvalidArgument",
Description: "Argument maxKeys must be an integer between 0 and 2147483647",
HTTPStatusCode: http.StatusBadRequest,
},
ErrInvalidMaxParts: {
Code: "InvalidArgument",
Description: "Argument max-parts must be an integer between 0 and 2147483647",
HTTPStatusCode: http.StatusBadRequest,
},
ErrInvalidPartNumberMarker: {
Code: "InvalidArgument",
Description: "Argument partNumberMarker must be an integer.",
HTTPStatusCode: http.StatusBadRequest,
},
ErrNoSuchBucket: {
Code: "NoSuchBucket",
Description: "The specified bucket does not exist",
HTTPStatusCode: http.StatusNotFound,
},
ErrNoSuchUpload: {
Code: "NoSuchUpload",
Description: "The specified multipart upload does not exist. The upload ID may be invalid, or the upload may have been aborted or completed.",
HTTPStatusCode: http.StatusNotFound,
},
ErrInternalError: {
Code: "InternalError",
Description: "We encountered an internal error, please try again.",
HTTPStatusCode: http.StatusInternalServerError,
},
ErrMalformedXML: {
Code: "MalformedXML",
Description: "The XML you provided was not well-formed or did not validate against our published schema.",
HTTPStatusCode: http.StatusBadRequest,
},
ErrInvalidPart: {
Code: "InvalidPart",
Description: "One or more of the specified parts could not be found. The part may not have been uploaded, or the specified entity tag may not match the part's entity tag.",
HTTPStatusCode: http.StatusBadRequest,
},
ErrInvalidPartOrder: {
Code: "InvalidPartOrder",
Description: "The list of parts was not in ascending order. The parts list must be specified in order by part number.",
HTTPStatusCode: http.StatusBadRequest,
},
ErrNotImplemented: {
Code: "NotImplemented",
Description: "A header you provided implies functionality that is not implemented",

104
weed/s3api/s3api_object_handlers.go
View File

@@ -49,50 +49,15 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request)
uploadUrl := fmt.Sprintf("http://%s%s/%s/%s?collection=%s",
s3a.option.Filer, s3a.option.BucketsPath, bucket, object, bucket)
proxyReq, err := http.NewRequest("PUT", uploadUrl, dataReader)
if err != nil {
glog.Errorf("NewRequest %s: %v", uploadUrl, err)
writeErrorResponse(w, ErrInternalError, r.URL)
etag, errCode := s3a.putToFiler(r, uploadUrl, dataReader)
if errCode != ErrNone {
writeErrorResponse(w, errCode, r.URL)
return
}
proxyReq.Header.Set("Host", s3a.option.Filer)
proxyReq.Header.Set("X-Forwarded-For", r.RemoteAddr)
for header, values := range r.Header {
for _, value := range values {
proxyReq.Header.Add(header, value)
}
}
resp, postErr := client.Do(proxyReq)
if postErr != nil {
glog.Errorf("post to filer: %v", postErr)
writeErrorResponse(w, ErrInternalError, r.URL)
return
}
defer resp.Body.Close()
resp_body, ra_err := ioutil.ReadAll(resp.Body)
if ra_err != nil {
glog.Errorf("upload to filer response read: %v", ra_err)
writeErrorResponse(w, ErrInternalError, r.URL)
return
}
var ret UploadResult
unmarshal_err := json.Unmarshal(resp_body, &ret)
if unmarshal_err != nil {
glog.Errorf("failing to read upload to %s : %v", uploadUrl, string(resp_body))
writeErrorResponse(w, ErrInternalError, r.URL)
return
}
if ret.Error != "" {
glog.Errorf("upload to filer error: %v", ret.Error)
writeErrorResponse(w, ErrInternalError, r.URL)
return
}
setEtag(w, etag)
writeSuccessResponseEmpty(w)
}
@@ -134,6 +99,12 @@ func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Reque
}
// DeleteMultipleObjectsHandler - Delete multiple objects
func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *http.Request) {
// TODO
writeErrorResponse(w, ErrNotImplemented, r.URL)
}
func (s3a *S3ApiServer) proxyToFiler(w http.ResponseWriter, r *http.Request, destUrl string, responseFn func(proxyResonse *http.Response, w http.ResponseWriter)) {
glog.V(2).Infof("s3 proxying %s to %s", r.Method, destUrl)
@@ -173,3 +144,56 @@ func passThroghResponse(proxyResonse *http.Response, w http.ResponseWriter) {
w.WriteHeader(proxyResonse.StatusCode)
io.Copy(w, proxyResonse.Body)
}
func (s3a *S3ApiServer) putToFiler(r *http.Request, uploadUrl string, dataReader io.ReadCloser) (etag string, code ErrorCode) {
proxyReq, err := http.NewRequest("PUT", uploadUrl, dataReader)
if err != nil {
glog.Errorf("NewRequest %s: %v", uploadUrl, err)
return "", ErrInternalError
}
proxyReq.Header.Set("Host", s3a.option.Filer)
proxyReq.Header.Set("X-Forwarded-For", r.RemoteAddr)
for header, values := range r.Header {
for _, value := range values {
proxyReq.Header.Add(header, value)
}
}
resp, postErr := client.Do(proxyReq)
if postErr != nil {
glog.Errorf("post to filer: %v", postErr)
return "", ErrInternalError
}
defer resp.Body.Close()
etag = resp.Header.Get("ETag")
resp_body, ra_err := ioutil.ReadAll(resp.Body)
if ra_err != nil {
glog.Errorf("upload to filer response read: %v", ra_err)
return etag, ErrInternalError
}
var ret UploadResult
unmarshal_err := json.Unmarshal(resp_body, &ret)
if unmarshal_err != nil {
glog.Errorf("failing to read upload to %s : %v", uploadUrl, string(resp_body))
return etag, ErrInternalError
}
if ret.Error != "" {
glog.Errorf("upload to filer error: %v", ret.Error)
return etag, ErrInternalError
}
return etag, ErrNone
}
func setEtag(w http.ResponseWriter, etag string) {
if etag != "" {
w.Header().Set("ETag", "\""+etag+"\"")
}
}

242
weed/s3api/s3api_object_multipart_handlers.go
View File

@@ -2,39 +2,263 @@ package s3api
import (
"net/http"
"github.com/gorilla/mux"
"fmt"
"github.com/aws/aws-sdk-go/service/s3"
"github.com/aws/aws-sdk-go/aws"
"net/url"
"strconv"
"io/ioutil"
"encoding/xml"
"sort"
"strings"
)
const (
maxObjectList = 1000 // Limit number of objects in a listObjectsResponse.
maxUploadsList = 1000 // Limit number of uploads in a listUploadsResponse.
maxPartsList = 1000 // Limit number of parts in a listPartsResponse.
globalMaxPartID = 10000
)
// NewMultipartUploadHandler - New multipart upload.
func (api *S3ApiServer) NewMultipartUploadHandler(w http.ResponseWriter, r *http.Request) {
func (s3a *S3ApiServer) NewMultipartUploadHandler(w http.ResponseWriter, r *http.Request) {
var object, bucket string
vars := mux.Vars(r)
bucket = vars["bucket"]
object = vars["object"]
response, errCode := s3a.createMultipartUpload(&s3.CreateMultipartUploadInput{
Bucket: aws.String(bucket),
Key: aws.String(object),
})
if errCode != ErrNone {
writeErrorResponse(w, errCode, r.URL)
return
}
writeSuccessResponseXML(w, encodeResponse(response))
}
// CompleteMultipartUploadHandler - Completes multipart upload.
func (api *S3ApiServer) CompleteMultipartUploadHandler(w http.ResponseWriter, r *http.Request) {
func (s3a *S3ApiServer) CompleteMultipartUploadHandler(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
bucket := vars["bucket"]
object := vars["object"]
// Get upload id.
uploadID, _, _, _ := getObjectResources(r.URL.Query())
completeMultipartBytes, err := ioutil.ReadAll(r.Body)
if err != nil {
writeErrorResponse(w, ErrInternalError, r.URL)
return
}
completedMultipartUpload := &s3.CompletedMultipartUpload{}
if err = xml.Unmarshal(completeMultipartBytes, completedMultipartUpload); err != nil {
writeErrorResponse(w, ErrMalformedXML, r.URL)
return
}
if len(completedMultipartUpload.Parts) == 0 {
writeErrorResponse(w, ErrMalformedXML, r.URL)
return
}
if !sort.IsSorted(byCompletedPartNumber(completedMultipartUpload.Parts)) {
writeErrorResponse(w, ErrInvalidPartOrder, r.URL)
return
}
response, errCode := s3a.completeMultipartUpload(&s3.CompleteMultipartUploadInput{
Bucket: aws.String(bucket),
Key: aws.String(object),
MultipartUpload: completedMultipartUpload,
UploadId: aws.String(uploadID),
})
if errCode != ErrNone {
writeErrorResponse(w, errCode, r.URL)
return
}
writeSuccessResponseXML(w, encodeResponse(response))
}
// AbortMultipartUploadHandler - Aborts multipart upload.
func (api *S3ApiServer) AbortMultipartUploadHandler(w http.ResponseWriter, r *http.Request) {
func (s3a *S3ApiServer) AbortMultipartUploadHandler(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
bucket := vars["bucket"]
object := vars["object"]
// Get upload id.
uploadID, _, _, _ := getObjectResources(r.URL.Query())
response, errCode := s3a.abortMultipartUpload(&s3.AbortMultipartUploadInput{
Bucket: aws.String(bucket),
Key: aws.String(object),
UploadId: aws.String(uploadID),
})
if errCode != ErrNone {
writeErrorResponse(w, errCode, r.URL)
return
}
writeSuccessResponseXML(w, encodeResponse(response))
}
// ListMultipartUploadsHandler - Lists multipart uploads.
func (api *S3ApiServer) ListMultipartUploadsHandler(w http.ResponseWriter, r *http.Request) {
func (s3a *S3ApiServer) ListMultipartUploadsHandler(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
bucket := vars["bucket"]
prefix, keyMarker, uploadIDMarker, delimiter, maxUploads, encodingType := getBucketMultipartResources(r.URL.Query())
if maxUploads < 0 {
writeErrorResponse(w, ErrInvalidMaxUploads, r.URL)
return
}
if keyMarker != "" {
// Marker not common with prefix is not implemented.
if !strings.HasPrefix(keyMarker, prefix) {
writeErrorResponse(w, ErrNotImplemented, r.URL)
return
}
}
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),
})
if errCode != ErrNone {
writeErrorResponse(w, errCode, r.URL)
return
}
writeSuccessResponseXML(w, encodeResponse(response))
}
// ListObjectPartsHandler - Lists object parts in a multipart upload.
func (api *S3ApiServer) ListObjectPartsHandler(w http.ResponseWriter, r *http.Request) {
func (s3a *S3ApiServer) ListObjectPartsHandler(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
bucket := vars["bucket"]
object := vars["object"]
uploadID, partNumberMarker, maxParts, _ := getObjectResources(r.URL.Query())
if partNumberMarker < 0 {
writeErrorResponse(w, ErrInvalidPartNumberMarker, r.URL)
return
}
if maxParts < 0 {
writeErrorResponse(w, ErrInvalidMaxParts, r.URL)
return
}
response, errCode := s3a.listObjectParts(&s3.ListPartsInput{
Bucket: aws.String(bucket),
Key: aws.String(object),
MaxParts: aws.Int64(int64(maxParts)),
PartNumberMarker: aws.Int64(int64(partNumberMarker)),
UploadId: aws.String(uploadID),
})
if errCode != ErrNone {
writeErrorResponse(w, errCode, r.URL)
return
}
writeSuccessResponseXML(w, encodeResponse(response))
}
// PutObjectPartHandler - Put an object part in a multipart upload.
func (api *S3ApiServer) PutObjectPartHandler(w http.ResponseWriter, r *http.Request) {
func (s3a *S3ApiServer) PutObjectPartHandler(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
bucket := vars["bucket"]
rAuthType := getRequestAuthType(r)
uploadID := r.URL.Query().Get("uploadId")
exists, err := s3a.exists(s3a.genUploadsFolder(bucket), uploadID, true)
if !exists {
writeErrorResponse(w, ErrNoSuchUpload, r.URL)
return
}
partIDString := r.URL.Query().Get("partNumber")
partID, err := strconv.Atoi(partIDString)
if err != nil {
writeErrorResponse(w, ErrInvalidPart, r.URL)
return
}
if partID > globalMaxPartID {
writeErrorResponse(w, ErrInvalidMaxParts, r.URL)
return
}
dataReader := r.Body
if rAuthType == authTypeStreamingSigned {
dataReader = newSignV4ChunkedReader(r)
}
uploadUrl := fmt.Sprintf("http://%s%s/%s/%04d.part",
s3a.option.Filer, s3a.genUploadsFolder(bucket), uploadID, partID-1)
etag, errCode := s3a.putToFiler(r, uploadUrl, dataReader)
if errCode != ErrNone {
writeErrorResponse(w, errCode, r.URL)
return
}
setEtag(w, etag)
writeSuccessResponseEmpty(w)
}
// DeleteMultipleObjectsHandler - Delete an object part in a multipart upload.
func (api *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *http.Request) {
func (s3a *S3ApiServer) genUploadsFolder(bucket string) string {
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 }

6
weed/s3api/s3api_server.go
View File

@@ -37,7 +37,7 @@ func (s3a *S3ApiServer) registerRouter(router *mux.Router) {
apiRouter := router.PathPrefix("/").Subrouter()
var routers []*mux.Router
if s3a.option.DomainName != "" {
routers = append(routers, apiRouter.Host("{bucket:.+}."+s3a.option.DomainName).Subrouter())
routers = append(routers, apiRouter.Host("{bucket:.+}."+ s3a.option.DomainName).Subrouter())
}
routers = append(routers, apiRouter.PathPrefix("/{bucket}").Subrouter())
@@ -59,8 +59,6 @@ func (s3a *S3ApiServer) registerRouter(router *mux.Router) {
bucket.Methods("POST").Path("/{object:.+}").HandlerFunc(s3a.CompleteMultipartUploadHandler).Queries("uploadId", "{uploadId:.*}")
// NewMultipartUpload
bucket.Methods("POST").Path("/{object:.+}").HandlerFunc(s3a.NewMultipartUploadHandler).Queries("uploads", "")
// DeleteMultipleObjects
bucket.Methods("POST").HandlerFunc(s3a.DeleteMultipleObjectsHandler).Queries("delete", "")
// AbortMultipartUpload
bucket.Methods("DELETE").Path("/{object:.+}").HandlerFunc(s3a.AbortMultipartUploadHandler).Queries("uploadId", "{uploadId:.*}")
// ListObjectParts
@@ -80,6 +78,8 @@ func (s3a *S3ApiServer) registerRouter(router *mux.Router) {
// ListObjectsV1 (Legacy)
bucket.Methods("GET").HandlerFunc(s3a.ListObjectsV1Handler)
// DeleteMultipleObjects
bucket.Methods("POST").HandlerFunc(s3a.DeleteMultipleObjectsHandler).Queries("delete", "")
/*
// CopyObject
bucket.Methods("PUT").Path("/{object:.+}").HeadersRegexp("X-Amz-Copy-Source", ".*?(\\/|%2F).*?").HandlerFunc(s3a.CopyObjectHandler)