From f240c57f16e89a03be9ca583ab25703421e82e1e Mon Sep 17 00:00:00 2001 From: tnextday Date: Mon, 14 Dec 2015 22:01:30 +0800 Subject: [PATCH 1/7] Rename ChunkManifest.GetData to ChunkManifest.Marshal --- go/operation/chunked_file.go | 2 +- go/operation/submit.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/go/operation/chunked_file.go b/go/operation/chunked_file.go index f8266087b..cb9ac80df 100644 --- a/go/operation/chunked_file.go +++ b/go/operation/chunked_file.go @@ -65,7 +65,7 @@ func LoadChunkManifest(buffer []byte, isGzipped bool) (*ChunkManifest, error) { return &cm, nil } -func (cm *ChunkManifest) GetData() ([]byte, error) { +func (cm *ChunkManifest) Marshal() ([]byte, error) { return json.Marshal(cm) } diff --git a/go/operation/submit.go b/go/operation/submit.go index ac5a3b55f..d996d63f0 100644 --- a/go/operation/submit.go +++ b/go/operation/submit.go @@ -180,7 +180,7 @@ func upload_one_chunk(filename string, reader io.Reader, master, } func upload_chunked_file_manifest(fileUrl string, manifest *ChunkManifest, jwt security.EncodedJwt) error { - buf, e := manifest.GetData() + buf, e := manifest.Marshal() if e != nil { return e } From aa44028b468674e252316a88a3b62f895b97e898 Mon Sep 17 00:00:00 2001 From: tnextday Date: Mon, 14 Dec 2015 22:14:57 +0800 Subject: [PATCH 2/7] update --- go/operation/chunked_file.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/go/operation/chunked_file.go b/go/operation/chunked_file.go index cb9ac80df..5d7f99176 100644 --- a/go/operation/chunked_file.go +++ b/go/operation/chunked_file.go @@ -74,7 +74,7 @@ func (cm *ChunkManifest) DeleteChunks(master string) error { for _, ci := range cm.Chunks { if e := DeleteFile(master, ci.Fid, ""); e != nil { deleteError++ - glog.V(0).Infoln("delete error:", e, ci.Fid) + glog.V(0).Infof("Delete %s error: %s, master: %s", ci.Fid, e.Error(), master) } } if deleteError > 0 { From b177afc3263f5f33f2182f58c6156b4d77115a0b Mon Sep 17 00:00:00 2001 From: tnextday Date: Tue, 15 Dec 2015 00:14:02 +0800 Subject: [PATCH 3/7] `weed download` command use stream download the large file. --- go/operation/chunked_file.go | 4 -- go/util/http_util.go | 5 +- go/weed/download.go | 94 +++++++++++++++++++++++------------- 3 files changed, 63 insertions(+), 40 deletions(-) diff --git a/go/operation/chunked_file.go b/go/operation/chunked_file.go index 5d7f99176..70564cbd2 100644 --- a/go/operation/chunked_file.go +++ b/go/operation/chunked_file.go @@ -83,10 +83,6 @@ func (cm *ChunkManifest) DeleteChunks(master string) error { return nil } -//func (cm *ChunkManifest) StoredHelper() error { -// return nil -//} - func readChunkNeedle(fileUrl string, w io.Writer, offset int64) (written int64, e error) { req, err := http.NewRequest("GET", fileUrl, nil) if err != nil { diff --git a/go/util/http_util.go b/go/util/http_util.go index d56aaa39a..f80ab0c24 100644 --- a/go/util/http_util.go +++ b/go/util/http_util.go @@ -136,12 +136,11 @@ func GetUrlStream(url string, values url.Values, readFn func(io.Reader) error) e return readFn(r.Body) } -func DownloadUrl(fileUrl string) (filename string, content []byte, e error) { +func DownloadUrl(fileUrl string) (filename string, rc io.ReadCloser, e error) { response, err := client.Get(fileUrl) if err != nil { return "", nil, err } - defer response.Body.Close() contentDisposition := response.Header["Content-Disposition"] if len(contentDisposition) > 0 { if strings.HasPrefix(contentDisposition[0], "filename=") { @@ -149,7 +148,7 @@ func DownloadUrl(fileUrl string) (filename string, content []byte, e error) { filename = strings.Trim(filename, "\"") } } - content, e = ioutil.ReadAll(response.Body) + rc = response.Body return } diff --git a/go/weed/download.go b/go/weed/download.go index 3c55b3a34..392b65edb 100644 --- a/go/weed/download.go +++ b/go/weed/download.go @@ -3,9 +3,11 @@ package main import ( "fmt" "io" - "io/ioutil" "os" "path" + + "io/ioutil" + "strings" "github.com/chrislusf/seaweedfs/go/operation" @@ -43,50 +45,76 @@ var cmdDownload = &Command{ func runDownload(cmd *Command, args []string) bool { for _, fid := range args { - filename, content, e := fetchFileId(*d.server, fid) - if e != nil { - fmt.Println("Fetch Error:", e) - continue + if e := downloadToFile(*d.server, fid, *d.dir); e != nil { + fmt.Println("Download Error:", e) } - if filename == "" { - filename = fid + } + return true +} + +func downloadToFile(server, fileId, saveDir string) error { + fileUrl, lookupError := operation.LookupFileId(server, fileId) + if lookupError != nil { + return lookupError + } + filename, rc, err := util.DownloadUrl(fileUrl) + if err != nil { + return err + } + defer rc.Close() + if filename == "" { + filename = fileId + } + isFileList := false + if strings.HasSuffix(filename, "-list") { + // old command compatible + isFileList = true + filename = filename[0 : len(filename)-len("-list")] + } + f, err := os.OpenFile(path.Join(saveDir, filename), os.O_WRONLY|os.O_CREATE|os.O_TRUNC, os.ModePerm) + if err != nil { + return err + } + defer f.Close() + if isFileList { + content, err := ioutil.ReadAll(rc) + if err != nil { + return err } - if strings.HasSuffix(filename, "-list") { - filename = filename[0 : len(filename)-len("-list")] - fids := strings.Split(string(content), "\n") - f, err := os.OpenFile(path.Join(*d.dir, filename), os.O_WRONLY|os.O_CREATE|os.O_TRUNC, os.ModePerm) - if err != nil { - fmt.Println("File Creation Error:", e) - continue + fids := strings.Split(string(content), "\n") + for _, partId := range fids { + var n int + _, part, err := fetchContent(*d.server, partId) + if err == nil { + n, err = f.Write(part) } - defer f.Close() - for _, partId := range fids { - var n int - _, part, err := fetchFileId(*d.server, partId) - if err == nil { - n, err = f.Write(part) - } - if err == nil && n < len(part) { - err = io.ErrShortWrite - } - if err != nil { - fmt.Println("File Write Error:", err) - break - } + if err == nil && n < len(part) { + err = io.ErrShortWrite } - } else { - ioutil.WriteFile(path.Join(*d.dir, filename), content, os.ModePerm) + if err != nil { + return err + } + } + } else { + if _, err = io.Copy(f, rc); err != nil { + return err } + } - return true + return nil } -func fetchFileId(server string, fileId string) (filename string, content []byte, e error) { +func fetchContent(server string, fileId string) (filename string, content []byte, e error) { fileUrl, lookupError := operation.LookupFileId(server, fileId) if lookupError != nil { return "", nil, lookupError } - filename, content, e = util.DownloadUrl(fileUrl) + var rc io.ReadCloser + if filename, rc, e = util.DownloadUrl(fileUrl); e != nil { + return "", nil, e + } + content, e = ioutil.ReadAll(rc) + rc.Close() return } From 90d410d627aff51bc825ff78861cb119617a07ba Mon Sep 17 00:00:00 2001 From: tnextday Date: Tue, 15 Dec 2015 00:30:18 +0800 Subject: [PATCH 4/7] verbose log --- go/weed/download.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/go/weed/download.go b/go/weed/download.go index 392b65edb..dfe4f88b4 100644 --- a/go/weed/download.go +++ b/go/weed/download.go @@ -46,7 +46,7 @@ var cmdDownload = &Command{ func runDownload(cmd *Command, args []string) bool { for _, fid := range args { if e := downloadToFile(*d.server, fid, *d.dir); e != nil { - fmt.Println("Download Error:", e) + fmt.Println("Download Error: ", fid, e) } } return true From f925374db6049beddd40830315623e7419b28e5a Mon Sep 17 00:00:00 2001 From: tnextday Date: Tue, 15 Dec 2015 11:26:16 +0800 Subject: [PATCH 5/7] Move write response content into a function --- .../volume_server_handlers_read.go | 217 ++++++------------ 1 file changed, 66 insertions(+), 151 deletions(-) diff --git a/go/weed/weed_server/volume_server_handlers_read.go b/go/weed/weed_server/volume_server_handlers_read.go index ffbd717bb..0b3fc1572 100644 --- a/go/weed/weed_server/volume_server_handlers_read.go +++ b/go/weed/weed_server/volume_server_handlers_read.go @@ -9,6 +9,10 @@ import ( "strings" "time" + "path" + + "bytes" + "github.com/chrislusf/seaweedfs/go/glog" "github.com/chrislusf/seaweedfs/go/images" "github.com/chrislusf/seaweedfs/go/operation" @@ -88,34 +92,25 @@ func (vs *VolumeServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) if n.NameSize > 0 && filename == "" { filename = string(n.Name) - dotIndex := strings.LastIndex(filename, ".") - if dotIndex > 0 { - ext = filename[dotIndex:] + if ext == "" { + ext = path.Ext(filename) } } mtype := "" - if ext != "" { - mtype = mime.TypeByExtension(ext) - } if n.MimeSize > 0 { mt := string(n.Mime) if !strings.HasPrefix(mt, "application/octet-stream") { mtype = mt } } - if mtype != "" { - w.Header().Set("Content-Type", mtype) - } - if filename != "" { - w.Header().Set("Content-Disposition", "filename=\""+fileNameEscaper.Replace(filename)+"\"") - } + if ext != ".gz" { if n.IsGzipped() { if strings.Contains(r.Header.Get("Accept-Encoding"), "gzip") { w.Header().Set("Content-Encoding", "gzip") } else { if n.Data, err = operation.UnGzipData(n.Data); err != nil { - glog.V(0).Infoln("lookup error:", err, r.URL.Path) + glog.V(0).Infoln("ungzip error:", err, r.URL.Path) } } } @@ -131,94 +126,9 @@ func (vs *VolumeServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) n.Data, _, _ = images.Resized(ext, n.Data, width, height) } - w.Header().Set("Accept-Ranges", "bytes") - if r.Method == "HEAD" { - w.Header().Set("Content-Length", strconv.Itoa(len(n.Data))) - return - } - rangeReq := r.Header.Get("Range") - if rangeReq == "" { - w.Header().Set("Content-Length", strconv.Itoa(len(n.Data))) - if _, e = w.Write(n.Data); e != nil { - glog.V(4).Infoln("response write error:", e) - } - return - } - - //the rest is dealing with partial content request - //mostly copy from src/pkg/net/http/fs.go - size := int64(len(n.Data)) - ranges, err := parseRange(rangeReq, size) - if err != nil { - http.Error(w, err.Error(), http.StatusRequestedRangeNotSatisfiable) - return - } - if sumRangesSize(ranges) > size { - // The total number of bytes in all the ranges - // is larger than the size of the file by - // itself, so this is probably an attack, or a - // dumb client. Ignore the range request. - ranges = nil - return - } - if len(ranges) == 0 { - return - } - if len(ranges) == 1 { - // RFC 2616, Section 14.16: - // "When an HTTP message includes the content of a single - // range (for example, a response to a request for a - // single range, or to a request for a set of ranges - // that overlap without any holes), this content is - // transmitted with a Content-Range header, and a - // Content-Length header showing the number of bytes - // actually transferred. - // ... - // A response to a request for a single range MUST NOT - // be sent using the multipart/byteranges media type." - ra := ranges[0] - w.Header().Set("Content-Length", strconv.FormatInt(ra.length, 10)) - w.Header().Set("Content-Range", ra.contentRange(size)) - w.WriteHeader(http.StatusPartialContent) - if _, e = w.Write(n.Data[ra.start : ra.start+ra.length]); e != nil { - glog.V(2).Infoln("response write error:", e) - } - return - } - // process multiple ranges - for _, ra := range ranges { - if ra.start > size { - http.Error(w, "Out of Range", http.StatusRequestedRangeNotSatisfiable) - return - } + if e := writeResponseContent(filename, mtype, bytes.NewReader(n.Data), w, r); e != nil { + glog.V(2).Infoln("response write error:", e) } - sendSize := rangesMIMESize(ranges, mtype, size) - pr, pw := io.Pipe() - mw := multipart.NewWriter(pw) - w.Header().Set("Content-Type", "multipart/byteranges; boundary="+mw.Boundary()) - sendContent := pr - defer pr.Close() // cause writing goroutine to fail and exit if CopyN doesn't finish. - go func() { - for _, ra := range ranges { - part, err := mw.CreatePart(ra.mimeHeader(mtype, size)) - if err != nil { - pw.CloseWithError(err) - return - } - if _, err = part.Write(n.Data[ra.start : ra.start+ra.length]); err != nil { - pw.CloseWithError(err) - return - } - } - mw.Close() - pw.Close() - }() - if w.Header().Get("Content-Encoding") == "" { - w.Header().Set("Content-Length", strconv.FormatInt(sendSize, 10)) - } - w.WriteHeader(http.StatusPartialContent) - io.CopyN(w, sendContent, sendSize) - } func (vs *VolumeServer) tryHandleChunkedFile(n *storage.Needle, fileName string, w http.ResponseWriter, r *http.Request) (processed bool) { @@ -236,69 +146,74 @@ func (vs *VolumeServer) tryHandleChunkedFile(n *storage.Needle, fileName string, glog.V(0).Infoln("load chunked manifest error:", e) return false } - ext := "" if fileName == "" && chunkManifest.Name != "" { fileName = chunkManifest.Name - dotIndex := strings.LastIndex(fileName, ".") - if dotIndex > 0 { - ext = fileName[dotIndex:] - } - } - mtype := "" - if ext != "" { - mtype = mime.TypeByExtension(ext) } + mType := "" if chunkManifest.Mime != "" { mt := chunkManifest.Mime if !strings.HasPrefix(mt, "application/octet-stream") { - mtype = mt + mType = mt } } - if mtype != "" { - w.Header().Set("Content-Type", mtype) - } - if fileName != "" { - w.Header().Set("Content-Disposition", `filename="`+fileNameEscaper.Replace(fileName)+`"`) - } + w.Header().Set("X-File-Store", "chunked") - w.Header().Set("Accept-Ranges", "bytes") - if r.Method == "HEAD" { - w.Header().Set("Content-Length", strconv.FormatInt(chunkManifest.Size, 10)) - return true - } chunkedFileReader := &operation.ChunkedFileReader{ Manifest: chunkManifest, Master: vs.GetMasterNode(), } defer chunkedFileReader.Close() + if e := writeResponseContent(fileName, mType, chunkedFileReader, w, r); e != nil { + glog.V(2).Infoln("response write error:", e) + } + return +} + +func writeResponseContent(filename, mimeType string, rs io.ReadSeeker, w http.ResponseWriter, r *http.Request) error { + totalSize, e := rs.Seek(0, 2) + if mimeType == "" { + if ext := path.Ext(filename); ext != "" { + mimeType = mime.TypeByExtension(ext) + } + } + if mimeType != "" { + w.Header().Set("Content-Type", mimeType) + } + if filename != "" { + w.Header().Set("Content-Disposition", `filename="`+fileNameEscaper.Replace(filename)+`"`) + } + w.Header().Set("Accept-Ranges", "bytes") + if r.Method == "HEAD" { + w.Header().Set("Content-Length", strconv.FormatInt(totalSize, 10)) + return nil + } rangeReq := r.Header.Get("Range") if rangeReq == "" { - w.Header().Set("Content-Length", strconv.FormatInt(chunkManifest.Size, 10)) - if _, e = io.Copy(w, chunkedFileReader); e != nil { - glog.V(2).Infoln("response write error:", e) + w.Header().Set("Content-Length", strconv.FormatInt(totalSize, 10)) + if _, e = rs.Seek(0, 0); e != nil { + return e } - return true + _, e = io.Copy(w, rs) + return e } //the rest is dealing with partial content request //mostly copy from src/pkg/net/http/fs.go - size := chunkManifest.Size - ranges, err := parseRange(rangeReq, size) + ranges, err := parseRange(rangeReq, totalSize) if err != nil { http.Error(w, err.Error(), http.StatusRequestedRangeNotSatisfiable) - return + return nil } - if sumRangesSize(ranges) > size { + if sumRangesSize(ranges) > totalSize { // The total number of bytes in all the ranges // is larger than the size of the file by // itself, so this is probably an attack, or a // dumb client. Ignore the range request. - ranges = nil - return + return nil } if len(ranges) == 0 { - return + return nil } if len(ranges) == 1 { // RFC 2616, Section 14.16: @@ -314,24 +229,23 @@ func (vs *VolumeServer) tryHandleChunkedFile(n *storage.Needle, fileName string, // be sent using the multipart/byteranges media type." ra := ranges[0] w.Header().Set("Content-Length", strconv.FormatInt(ra.length, 10)) - w.Header().Set("Content-Range", ra.contentRange(size)) + w.Header().Set("Content-Range", ra.contentRange(totalSize)) w.WriteHeader(http.StatusPartialContent) - if _, e = chunkedFileReader.Seek(ra.start, 0); e != nil { - glog.V(2).Infoln("chunkedFileReader Seek error:", e) + if _, e = rs.Seek(ra.start, 0); e != nil { + return e } - if _, e = io.CopyN(w, chunkedFileReader, ra.length); e != nil { - glog.V(2).Infoln("response write error:", e) - } - return + + _, e = io.CopyN(w, rs, ra.length) + return e } // process multiple ranges for _, ra := range ranges { - if ra.start > size { + if ra.start > totalSize { http.Error(w, "Out of Range", http.StatusRequestedRangeNotSatisfiable) - return + return nil } } - sendSize := rangesMIMESize(ranges, mtype, size) + sendSize := rangesMIMESize(ranges, mimeType, totalSize) pr, pw := io.Pipe() mw := multipart.NewWriter(pw) w.Header().Set("Content-Type", "multipart/byteranges; boundary="+mw.Boundary()) @@ -339,16 +253,17 @@ func (vs *VolumeServer) tryHandleChunkedFile(n *storage.Needle, fileName string, defer pr.Close() // cause writing goroutine to fail and exit if CopyN doesn't finish. go func() { for _, ra := range ranges { - part, err := mw.CreatePart(ra.mimeHeader(mtype, size)) - if err != nil { - pw.CloseWithError(err) + part, e := mw.CreatePart(ra.mimeHeader(mimeType, totalSize)) + if e != nil { + pw.CloseWithError(e) return } - if _, e = chunkedFileReader.Seek(ra.start, 0); e != nil { - glog.V(2).Infoln("response write error:", e) + if _, e = rs.Seek(ra.start, 0); e != nil { + pw.CloseWithError(e) + return } - if _, err = io.CopyN(part, chunkedFileReader, ra.length); err != nil { - pw.CloseWithError(err) + if _, e = io.CopyN(part, rs, ra.length); e != nil { + pw.CloseWithError(e) return } } @@ -359,6 +274,6 @@ func (vs *VolumeServer) tryHandleChunkedFile(n *storage.Needle, fileName string, w.Header().Set("Content-Length", strconv.FormatInt(sendSize, 10)) } w.WriteHeader(http.StatusPartialContent) - io.CopyN(w, sendContent, sendSize) - return + _, e = io.CopyN(w, sendContent, sendSize) + return e } From 521be17a9e28e8f84546f817085e69e5453fe51a Mon Sep 17 00:00:00 2001 From: tnextday Date: Tue, 15 Dec 2015 13:05:59 +0800 Subject: [PATCH 6/7] verbose log --- go/weed/weed_server/volume_server_handlers_read.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/go/weed/weed_server/volume_server_handlers_read.go b/go/weed/weed_server/volume_server_handlers_read.go index 0b3fc1572..c10f87c03 100644 --- a/go/weed/weed_server/volume_server_handlers_read.go +++ b/go/weed/weed_server/volume_server_handlers_read.go @@ -143,7 +143,7 @@ func (vs *VolumeServer) tryHandleChunkedFile(n *storage.Needle, fileName string, chunkManifest, e := operation.LoadChunkManifest(n.Data, n.IsGzipped()) if e != nil { - glog.V(0).Infoln("load chunked manifest error:", e) + glog.V(0).Infof("Load chunked manifest (%s) error: %s", r.URL.Path, e.Error()) return false } if fileName == "" && chunkManifest.Name != "" { From 031d26527f0ebe39bb26c8e8b4503168a849265a Mon Sep 17 00:00:00 2001 From: tnextday Date: Tue, 15 Dec 2015 13:08:09 +0800 Subject: [PATCH 7/7] update --- go/weed/weed_server/volume_server_handlers_read.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/go/weed/weed_server/volume_server_handlers_read.go b/go/weed/weed_server/volume_server_handlers_read.go index c10f87c03..2aa0fc656 100644 --- a/go/weed/weed_server/volume_server_handlers_read.go +++ b/go/weed/weed_server/volume_server_handlers_read.go @@ -143,7 +143,7 @@ func (vs *VolumeServer) tryHandleChunkedFile(n *storage.Needle, fileName string, chunkManifest, e := operation.LoadChunkManifest(n.Data, n.IsGzipped()) if e != nil { - glog.V(0).Infof("Load chunked manifest (%s) error: %s", r.URL.Path, e.Error()) + glog.V(0).Infof("load chunked manifest (%s) error: %s", r.URL.Path, e.Error()) return false } if fileName == "" && chunkManifest.Name != "" {