Mirror
/
seaweedfs
mirror of https://github.com/seaweedfs/seaweedfs.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Merge pull request #314 from hxiaodon/master 

made a little change for Amazon S3 API's compatibility(v2)
pull/315/head
Chris Lu 9 years ago
parent
0e3140c54c 3a25af223f
commit
e416679cea
1 changed files with 185 additions and 37 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 216
      weed/server/filer_server_handlers_write.go

216
weed/server/filer_server_handlers_write.go
View File

@ -2,10 +2,16 @@ package weed_server
import (
"bytes"
"crypto/md5"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"io"
"io/ioutil"
"mime/multipart"
"net/http"
"net/textproto"
"net/url"
"strings"
@ -16,13 +22,6 @@ import (
"github.com/syndtr/goleveldb/leveldb"
)
type analogueReader struct {
*bytes.Buffer
}
// So that it implements the io.ReadCloser interface
func (m analogueReader) Close() error { return nil }
type FilerPostResult struct {
Name string `json:"name,omitempty"`
Size uint32 `json:"size,omitempty"`
@ -31,31 +30,89 @@ type FilerPostResult struct {
Url string `json:"url,omitempty"`
}
func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request) {
query := r.URL.Query()
replication := query.Get("replication")
if replication == "" {
replication = fs.defaultReplication
var quoteEscaper = strings.NewReplacer("\\", "\\\\", `"`, "\\\"")
func escapeQuotes(s string) string {
return quoteEscaper.Replace(s)
}
collection := query.Get("collection")
if collection == "" {
collection = fs.collection
func createFormFile(writer *multipart.Writer, fieldname, filename, mime string) (io.Writer, error) {
h := make(textproto.MIMEHeader)
h.Set("Content-Disposition",
fmt.Sprintf(`form-data; name="%s"; filename="%s"`,
escapeQuotes(fieldname), escapeQuotes(filename)))
if len(mime) == 0 {
mime = "application/octet-stream"
}
h.Set("Content-Type", mime)
return writer.CreatePart(h)
}
var fileId string
var err error
var urlLocation string
func makeFormData(filename, mimeType string, content io.Reader) (formData io.Reader, contentType string, err error) {
buf := new(bytes.Buffer)
writer := multipart.NewWriter(buf)
defer writer.Close()
part, err := createFormFile(writer, "file", filename, mimeType)
if err != nil {
glog.V(0).Infoln(err)
return
}
_, err = io.Copy(part, content)
if err != nil {
glog.V(0).Infoln(err)
return
}
formData = buf
contentType = writer.FormDataContentType()
return
}
func (fs *FilerServer) queryFileInfoByPath(w http.ResponseWriter, r *http.Request, path string) (fileId, urlLocation string, err error) {
if fileId, err = fs.filer.FindFile(path); err != nil && err != leveldb.ErrNotFound {
glog.V(0).Infoln("failing to find path in filer store", path, err.Error())
writeJsonError(w, r, http.StatusInternalServerError, err)
return
} else if fileId != "" && err == nil {
urlLocation, err = operation.LookupFileId(fs.getMasterNode(), fileId)
if err != nil {
glog.V(1).Infoln("operation LookupFileId %s failed, err is %s", fileId, err.Error())
w.WriteHeader(http.StatusNotFound)
return
}
}
return
}
func (fs *FilerServer) assignNewFileInfo(w http.ResponseWriter, r *http.Request, replication, collection string) (fileId, urlLocation string, err error) {
assignResult, ae := operation.Assign(fs.getMasterNode(), 1, replication, collection, r.URL.Query().Get("ttl"))
if ae != nil {
glog.V(0).Infoln("failing to assign a file id", ae.Error())
writeJsonError(w, r, http.StatusInternalServerError, ae)
err = ae
return
}
fileId = assignResult.Fid
urlLocation = "http://" + assignResult.Url + "/" + assignResult.Fid
return
}
func (fs *FilerServer) multipartUploadAnalyzer(w http.ResponseWriter, r *http.Request, replication, collection string) (fileId, urlLocation string, err error) {
//Default handle way for http multipart
if r.Method == "PUT" {
buf, _ := ioutil.ReadAll(r.Body)
r.Body = analogueReader{bytes.NewBuffer(buf)}
r.Body = ioutil.NopCloser(bytes.NewBuffer(buf))
fileName, _, _, _, _, _, _, pe := storage.ParseUpload(r)
if pe != nil {
glog.V(0).Infoln("failing to parse post body", pe.Error())
writeJsonError(w, r, http.StatusInternalServerError, pe)
err = pe
return
}
//reconstruct http request body for following new request to volume server
r.Body = analogueReader{bytes.NewBuffer(buf)}
r.Body = ioutil.NopCloser(bytes.NewBuffer(buf))
path := r.URL.Path
if strings.HasSuffix(path, "/") {
@ -63,29 +120,120 @@ func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request) {
path += fileName
}
}
fileId, urlLocation, err = fs.queryFileInfoByPath(w, r, path)
} else {
fileId, urlLocation, err = fs.assignNewFileInfo(w, r, replication, collection)
}
return
}
if fileId, err = fs.filer.FindFile(path); err != nil && err != leveldb.ErrNotFound {
glog.V(0).Infoln("failing to find path in filer store", path, err.Error())
func multipartHttpBodyBuilder(w http.ResponseWriter, r *http.Request, fileName string) (err error) {
body, contentType, te := makeFormData(fileName, r.Header.Get("Content-Type"), r.Body)
if te != nil {
glog.V(0).Infoln("S3 protocol to raw seaweed protocol failed", te.Error())
writeJsonError(w, r, http.StatusInternalServerError, te)
err = te
return
}
if body != nil {
switch v := body.(type) {
case *bytes.Buffer:
r.ContentLength = int64(v.Len())
case *bytes.Reader:
r.ContentLength = int64(v.Len())
case *strings.Reader:
r.ContentLength = int64(v.Len())
}
}
r.Header.Set("Content-Type", contentType)
rc, ok := body.(io.ReadCloser)
if !ok && body != nil {
rc = ioutil.NopCloser(body)
}
r.Body = rc
return
}
func checkContentMD5(w http.ResponseWriter, r *http.Request) (err error) {
if contentMD5 := r.Header.Get("Content-MD5"); contentMD5 != "" {
buf, _ := ioutil.ReadAll(r.Body)
//checkMD5
sum := md5.Sum(buf)
fileDataMD5 := base64.StdEncoding.EncodeToString(sum[0:len(sum)])
if strings.ToLower(fileDataMD5) != strings.ToLower(contentMD5) {
glog.V(0).Infof("fileDataMD5 [%s] is not equal to Content-MD5 [%s]", fileDataMD5, contentMD5)
err = fmt.Errorf("MD5 check failed")
writeJsonError(w, r, http.StatusNotAcceptable, err)
return
}
//reconstruct http request body for following new request to volume server
r.Body = ioutil.NopCloser(bytes.NewBuffer(buf))
}
return
}
func (fs *FilerServer) monolithicUploadAnalyzer(w http.ResponseWriter, r *http.Request, replication, collection string) (fileId, urlLocation string, err error) {
/*
Amazon S3 ref link:[http://docs.aws.amazon.com/AmazonS3/latest/API/Welcome.html]
There is a long way to provide a completely compatibility against all Amazon S3 API, I just made
a simple data stream adapter between S3 PUT API and seaweedfs's volume storage Write API
1. The request url format should be http://$host:$port/$bucketName/$objectName
2. bucketName will be mapped to seaweedfs's collection name
3. You could customize and make your enhancement.
*/
lastPos := strings.LastIndex(r.URL.Path, "/")
if lastPos == -1 || lastPos == 0 || lastPos == len(r.URL.Path)-1 {
glog.V(0).Infoln("URL Path [%s] is invalid, could not retrieve file name", r.URL.Path)
err = fmt.Errorf("URL Path is invalid")
writeJsonError(w, r, http.StatusInternalServerError, err)
return
} else if fileId != "" && err == nil {
var le error
urlLocation, le = operation.LookupFileId(fs.getMasterNode(), fileId)
if le != nil {
glog.V(1).Infoln("operation LookupFileId %s failed, err is %s", fileId, le.Error())
w.WriteHeader(http.StatusNotFound)
}
if err = checkContentMD5(w, r); err != nil {
return
}
fileName := r.URL.Path[lastPos+1:]
if err = multipartHttpBodyBuilder(w, r, fileName); err != nil {
return
}
secondPos := strings.Index(r.URL.Path[1:], "/") + 1
collection = r.URL.Path[1:secondPos]
path := r.URL.Path
if fileId, urlLocation, err = fs.queryFileInfoByPath(w, r, path); err == nil && fileId == "" {
fileId, urlLocation, err = fs.assignNewFileInfo(w, r, replication, collection)
}
return
}
func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request) {
query := r.URL.Query()
replication := query.Get("replication")
if replication == "" {
replication = fs.defaultReplication
}
collection := query.Get("collection")
if collection == "" {
collection = fs.collection
}
var fileId, urlLocation string
var err error
if strings.HasPrefix(r.Header.Get("Content-Type"), "multipart/form-data; boundary=") {
fileId, urlLocation, err = fs.multipartUploadAnalyzer(w, r, replication, collection)
if err != nil {
return
}
} else {
assignResult, ae := operation.Assign(fs.getMasterNode(), 1, replication, collection, query.Get("ttl"))
if ae != nil {
glog.V(0).Infoln("failing to assign a file id", ae.Error())
writeJsonError(w, r, http.StatusInternalServerError, ae)
fileId, urlLocation, err = fs.monolithicUploadAnalyzer(w, r, replication, collection)
if err != nil {
return
}
fileId = assignResult.Fid
urlLocation = "http://" + assignResult.Url + "/" + assignResult.Fid
}
u, _ := url.Parse(urlLocation)

Write Preview
Loading…
Cancel
Save