From a4f64c011620e2ebd5dc41415e6f19713be9897a Mon Sep 17 00:00:00 2001 From: tnextday Date: Thu, 26 Nov 2015 23:30:08 +0800 Subject: [PATCH 01/21] edit git ignore --- .gitignore | 73 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 73 insertions(+) diff --git a/.gitignore b/.gitignore index d121337a5..360cbb2cb 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,76 @@ go/weed/.goxc* tags *.swp +### OSX template +.DS_Store +.AppleDouble +.LSOverride + +# Icon must end with two \r +Icon + +# Thumbnails +._* + +# Files that might appear in the root of a volume +.DocumentRevisions-V100 +.fseventsd +.Spotlight-V100 +.TemporaryItems +.Trashes +.VolumeIcon.icns + +# Directories potentially created on remote AFP share +.AppleDB +.AppleDesktop +Network Trash Folder +Temporary Items +.apdisk +### JetBrains template +# Covers JetBrains IDEs: IntelliJ, RubyMine, PhpStorm, AppCode, PyCharm, CLion, Android Studio + +*.iml + +## Directory-based project format: +.idea/ +# if you remove the above rule, at least ignore the following: + +# User-specific stuff: +# .idea/workspace.xml +# .idea/tasks.xml +# .idea/dictionaries + +# Sensitive or high-churn files: +# .idea/dataSources.ids +# .idea/dataSources.xml +# .idea/sqlDataSources.xml +# .idea/dynamic.xml +# .idea/uiDesigner.xml + +# Gradle: +# .idea/gradle.xml +# .idea/libraries + +# Mongo Explorer plugin: +# .idea/mongoSettings.xml + +## File-based project format: +*.ipr +*.iws + +## Plugin-specific files: + +# IntelliJ +/out/ + +# mpeltonen/sbt-idea plugin +.idea_modules/ + +# JIRA plugin +atlassian-ide-plugin.xml + +# Crashlytics plugin (for Android Studio and IntelliJ) +com_crashlytics_export_strings.xml +crashlytics.properties +crashlytics-build.properties + From 1817864a4589e05fa17a27ed1bc954b0cfcf12b4 Mon Sep 17 00:00:00 2001 From: tnextday Date: Sun, 29 Nov 2015 23:49:41 +0800 Subject: [PATCH 02/21] add chunk file helper to support large file --- go/operation/chunked_file.go | 126 ++++++++++++++++++ go/storage/needle_read_write.go | 9 ++ go/topology/volume_location_list.go | 1 + .../volume_server_handlers_read.go | 2 +- 4 files changed, 137 insertions(+), 1 deletion(-) create mode 100644 go/operation/chunked_file.go diff --git a/go/operation/chunked_file.go b/go/operation/chunked_file.go new file mode 100644 index 000000000..a581af574 --- /dev/null +++ b/go/operation/chunked_file.go @@ -0,0 +1,126 @@ +package operation + +import ( + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "sort" + + "github.com/chrislusf/seaweedfs/go/util" +) + +var ErrOutOfRange = errors.New("Out of Range") + +type ChunkInfo struct { + Fid string `json:"fid,omitempty"` + Offset uint64 `json:"offset,omitempty"` + Size uint32 `json:"size,omitempty"` +} + +type ChunkList []*ChunkInfo + +type ChunkedFile struct { + Name string `json:"name,omitempty"` + Mime string `json:"mime,omitempty"` + Size uint64 `json:"size,omitempty"` + Chunks ChunkList `json:"chunks,omitempty"` + + master string `json:"-"` +} + +func (s ChunkList) Len() int { return len(s) } +func (s ChunkList) Less(i, j int) bool { return s[i].Offset < s[j].Offset } +func (s ChunkList) Swap(i, j int) { s[i], s[j] = s[j], s[i] } + +func NewChunkedNeedle(buffer []byte, master string) (*ChunkedFile, error) { + c := ChunkedFile{} + + if e := json.Unmarshal(buffer, c); e != nil { + return nil, e + } + sort.Sort(c.Chunks) + c.master = master + return &c, nil +} + +func (c *ChunkedFile) Marshal() ([]byte, error) { + return json.Marshal(c) +} + +func copyChunk(fileUrl string, w io.Writer, startOffset, size int64) (written int64, e error) { + req, err := http.NewRequest("GET", fileUrl, nil) + if err != nil { + return written, err + } + if startOffset > 0 { + req.Header.Set("Range", fmt.Sprintf("bytes=%d-", startOffset)) + } + + resp, err := util.Do(req) + if err != nil { + return written, err + } + defer resp.Close() + if startOffset > 0 && resp.StatusCode != 206 { + return written, fmt.Errorf("Cannot Read Needle Position: %d [%s]", startOffset, fileUrl) + } + + if size > 0 { + return io.CopyN(w, resp, size) + } else { + return io.Copy(w, resp) + } +} + +func (c *ChunkedFile) WriteBuffer(w io.Writer, offset, size int64) (written int64, e error) { + if offset >= c.Size || offset+size > c.Size { + return written, ErrOutOfRange + } + chunkIndex := -1 + chunkStartOffset := 0 + for i, ci := range c.Chunks { + if offset >= ci.Offset && offset < ci.Offset+ci.Size { + chunkIndex = i + chunkStartOffset = offset - ci.Offset + break + } + } + if chunkIndex < 0 { + return written, ErrOutOfRange + } + for ; chunkIndex < c.Chunks.Len(); chunkIndex++ { + ci := c.Chunks[chunkIndex] + fileUrl, lookupError := LookupFileId(c.master, ci.Fid) + if lookupError != nil { + return written, lookupError + } + rsize := 0 + if size > 0 { + rsize = size - written + } + if n, e := copyChunk(fileUrl, w, chunkStartOffset, rsize); e != nil { + return written, e + } else { + written += n + } + + if size > 0 && written >= size { + break + } + chunkStartOffset = 0 + } + + return written, nil +} + +func (c *ChunkedFile) DeleteHelper() error { + //TODO Delete all chunks + return nil +} + +func (c *ChunkedFile) StoredHelper() error { + //TODO + return nil +} diff --git a/go/storage/needle_read_write.go b/go/storage/needle_read_write.go index eb2d8d459..2a5bd89ea 100644 --- a/go/storage/needle_read_write.go +++ b/go/storage/needle_read_write.go @@ -16,6 +16,7 @@ const ( FlagHasMime = 0x04 FlagHasLastModifiedDate = 0x08 FlagHasTtl = 0x10 + FlagChunkList = 0x80 LastModifiedBytesLength = 5 TtlBytesLength = 2 ) @@ -280,3 +281,11 @@ func (n *Needle) HasTtl() bool { func (n *Needle) SetHasTtl() { n.Flags = n.Flags | FlagHasTtl } + +func (n *Needle) IsChunkList() bool { + return n.Flags&FlagChunkList > 0 +} + +func (n *Needle) SetChunkList() { + n.Flags = n.Flags | FlagChunkList +} diff --git a/go/topology/volume_location_list.go b/go/topology/volume_location_list.go index 0f892c010..d5eaf5e92 100644 --- a/go/topology/volume_location_list.go +++ b/go/topology/volume_location_list.go @@ -17,6 +17,7 @@ func (dnll *VolumeLocationList) String() string { } func (dnll *VolumeLocationList) Head() *DataNode { + //mark first node as master volume return dnll.list[0] } diff --git a/go/weed/weed_server/volume_server_handlers_read.go b/go/weed/weed_server/volume_server_handlers_read.go index d569f5510..9e252d205 100644 --- a/go/weed/weed_server/volume_server_handlers_read.go +++ b/go/weed/weed_server/volume_server_handlers_read.go @@ -180,7 +180,7 @@ func (vs *VolumeServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) } return } - // process mulitple ranges + // process multiple ranges for _, ra := range ranges { if ra.start > size { http.Error(w, "Out of Range", http.StatusRequestedRangeNotSatisfiable) From f825d237890c7df70c3d752668ddbc2deab3890e Mon Sep 17 00:00:00 2001 From: tnextday Date: Mon, 30 Nov 2015 00:21:42 +0800 Subject: [PATCH 03/21] fix build error --- go/operation/chunked_file.go | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/go/operation/chunked_file.go b/go/operation/chunked_file.go index a581af574..953765149 100644 --- a/go/operation/chunked_file.go +++ b/go/operation/chunked_file.go @@ -15,8 +15,8 @@ var ErrOutOfRange = errors.New("Out of Range") type ChunkInfo struct { Fid string `json:"fid,omitempty"` - Offset uint64 `json:"offset,omitempty"` - Size uint32 `json:"size,omitempty"` + Offset int64 `json:"offset,omitempty"` + Size int64 `json:"size,omitempty"` } type ChunkList []*ChunkInfo @@ -24,7 +24,7 @@ type ChunkList []*ChunkInfo type ChunkedFile struct { Name string `json:"name,omitempty"` Mime string `json:"mime,omitempty"` - Size uint64 `json:"size,omitempty"` + Size int64 `json:"size,omitempty"` Chunks ChunkList `json:"chunks,omitempty"` master string `json:"-"` @@ -62,15 +62,15 @@ func copyChunk(fileUrl string, w io.Writer, startOffset, size int64) (written in if err != nil { return written, err } - defer resp.Close() + defer resp.Body.Close() if startOffset > 0 && resp.StatusCode != 206 { return written, fmt.Errorf("Cannot Read Needle Position: %d [%s]", startOffset, fileUrl) } if size > 0 { - return io.CopyN(w, resp, size) + return io.CopyN(w, resp.Body, size) } else { - return io.Copy(w, resp) + return io.Copy(w, resp.Body) } } @@ -79,7 +79,7 @@ func (c *ChunkedFile) WriteBuffer(w io.Writer, offset, size int64) (written int6 return written, ErrOutOfRange } chunkIndex := -1 - chunkStartOffset := 0 + chunkStartOffset := int64(0) for i, ci := range c.Chunks { if offset >= ci.Offset && offset < ci.Offset+ci.Size { chunkIndex = i @@ -90,13 +90,14 @@ func (c *ChunkedFile) WriteBuffer(w io.Writer, offset, size int64) (written int6 if chunkIndex < 0 { return written, ErrOutOfRange } + //preload next chunk? for ; chunkIndex < c.Chunks.Len(); chunkIndex++ { ci := c.Chunks[chunkIndex] fileUrl, lookupError := LookupFileId(c.master, ci.Fid) if lookupError != nil { return written, lookupError } - rsize := 0 + rsize := int64(0) if size > 0 { rsize = size - written } From 6b0894d80635047479548bb19029300a596b4d55 Mon Sep 17 00:00:00 2001 From: tnextday Date: Tue, 1 Dec 2015 20:23:50 +0800 Subject: [PATCH 04/21] update ChunkedFile to seekable reader, so we can use io.* to read data --- go/operation/chunked_file.go | 205 +++++++++++++----- go/storage/needle.go | 11 +- go/storage/needle_read_write.go | 10 +- go/weed/weed_server/common.go | 2 +- .../volume_server_handlers_read.go | 150 +++++++++++++ 5 files changed, 313 insertions(+), 65 deletions(-) diff --git a/go/operation/chunked_file.go b/go/operation/chunked_file.go index 953765149..ca97566a5 100644 --- a/go/operation/chunked_file.go +++ b/go/operation/chunked_file.go @@ -8,10 +8,18 @@ import ( "net/http" "sort" + "sync" + + "github.com/chrislusf/seaweedfs/go/glog" "github.com/chrislusf/seaweedfs/go/util" ) -var ErrOutOfRange = errors.New("Out of Range") +var ( + // when the remote server does not allow range requests (Accept-Ranges was not set) + ErrRangeRequestsNotSupported = errors.New("Range requests are not supported by the remote server") + // ErrInvalidRange is returned by Read when trying to read past the end of the file + ErrInvalidRange = errors.New("Invalid range") +) type ChunkInfo struct { Fid string `json:"fid,omitempty"` @@ -21,41 +29,74 @@ type ChunkInfo struct { type ChunkList []*ChunkInfo -type ChunkedFile struct { - Name string `json:"name,omitempty"` - Mime string `json:"mime,omitempty"` - Size int64 `json:"size,omitempty"` - Chunks ChunkList `json:"chunks,omitempty"` +type ChunkManifest struct { + Name string `json:"name,omitempty"` + Mime string `json:"mime,omitempty"` + Size int64 `json:"size,omitempty"` + Chunks ChunkList `json:"chunks,omitempty"` +} - master string `json:"-"` +// seekable chunked file reader +type ChunkedFileReader struct { + Manifest *ChunkManifest + Master string + pos int64 + pr *io.PipeReader + pw *io.PipeWriter + mutex sync.Mutex } func (s ChunkList) Len() int { return len(s) } func (s ChunkList) Less(i, j int) bool { return s[i].Offset < s[j].Offset } func (s ChunkList) Swap(i, j int) { s[i], s[j] = s[j], s[i] } -func NewChunkedNeedle(buffer []byte, master string) (*ChunkedFile, error) { - c := ChunkedFile{} - - if e := json.Unmarshal(buffer, c); e != nil { +func LoadChunkedManifest(buffer []byte) (*ChunkManifest, error) { + cm := ChunkManifest{} + if e := json.Unmarshal(buffer, cm); e != nil { return nil, e } - sort.Sort(c.Chunks) - c.master = master - return &c, nil + sort.Sort(cm.Chunks) + return &cm, nil } -func (c *ChunkedFile) Marshal() ([]byte, error) { - return json.Marshal(c) +func (cm *ChunkManifest) GetData() ([]byte, error) { + return json.Marshal(cm) } -func copyChunk(fileUrl string, w io.Writer, startOffset, size int64) (written int64, e error) { +func (cm *ChunkManifest) DeleteChunks(master string) error { + fileIds := make([]string, 0, len(cm.Chunks)) + for _, ci := range cm.Chunks { + fileIds = append(fileIds, ci.Fid) + } + results, e := DeleteFiles(master, fileIds) + if e != nil { + return e + } + deleteError := 0 + for _, ret := range results.Results { + if ret.Error != "" { + deleteError++ + glog.V(0).Infoln("delete error:", ret.Error, ret.Fid) + } + } + if deleteError > 0 { + return errors.New("Not all chunks deleted.") + } + return nil +} + +func (cm *ChunkManifest) StoredHelper() error { + //TODO + return nil +} + +func httpRangeDownload(fileUrl string, w io.Writer, offset int64) (written int64, e error) { req, err := http.NewRequest("GET", fileUrl, nil) if err != nil { return written, err } - if startOffset > 0 { - req.Header.Set("Range", fmt.Sprintf("bytes=%d-", startOffset)) + if offset > 0 { + req.Header.Set("Range", fmt.Sprintf("bytes=%d-", offset)) } resp, err := util.Do(req) @@ -63,65 +104,117 @@ func copyChunk(fileUrl string, w io.Writer, startOffset, size int64) (written in return written, err } defer resp.Body.Close() - if startOffset > 0 && resp.StatusCode != 206 { - return written, fmt.Errorf("Cannot Read Needle Position: %d [%s]", startOffset, fileUrl) - } - if size > 0 { - return io.CopyN(w, resp.Body, size) - } else { - return io.Copy(w, resp.Body) + switch resp.StatusCode { + case http.StatusRequestedRangeNotSatisfiable: + return written, ErrInvalidRange + case http.StatusOK: + if offset > 0 { + return written, ErrRangeRequestsNotSupported + } + case http.StatusPartialContent: + break + default: + return written, fmt.Errorf("Read Needle http error: [%d] %s", resp.StatusCode, fileUrl) + } + return io.Copy(w, resp.Body) } -func (c *ChunkedFile) WriteBuffer(w io.Writer, offset, size int64) (written int64, e error) { - if offset >= c.Size || offset+size > c.Size { - return written, ErrOutOfRange +func (cf *ChunkedFileReader) Seek(offset int64, whence int) (int64, error) { + var err error + switch whence { + case 0: + case 1: + offset += cf.pos + case 2: + offset = cf.Manifest.Size - offset } + if offset > cf.Manifest.Size { + err = ErrInvalidRange + } + if cf.pos != offset { + cf.Close() + } + cf.pos = offset + return cf.pos, err +} + +func (cf *ChunkedFileReader) WriteTo(w io.Writer) (n int64, err error) { + cm := cf.Manifest chunkIndex := -1 chunkStartOffset := int64(0) - for i, ci := range c.Chunks { - if offset >= ci.Offset && offset < ci.Offset+ci.Size { + for i, ci := range cm.Chunks { + if cf.pos >= ci.Offset && cf.pos < ci.Offset+ci.Size { chunkIndex = i - chunkStartOffset = offset - ci.Offset + chunkStartOffset = cf.pos - ci.Offset break } } if chunkIndex < 0 { - return written, ErrOutOfRange + return n, ErrInvalidRange } - //preload next chunk? - for ; chunkIndex < c.Chunks.Len(); chunkIndex++ { - ci := c.Chunks[chunkIndex] - fileUrl, lookupError := LookupFileId(c.master, ci.Fid) + for ; chunkIndex < cm.Chunks.Len(); chunkIndex++ { + ci := cm.Chunks[chunkIndex] + // if we need read date from local volume server first? + fileUrl, lookupError := LookupFileId(cf.Master, ci.Fid) if lookupError != nil { - return written, lookupError + return n, lookupError } - rsize := int64(0) - if size > 0 { - rsize = size - written - } - if n, e := copyChunk(fileUrl, w, chunkStartOffset, rsize); e != nil { - return written, e + if wn, e := httpRangeDownload(fileUrl, w, chunkStartOffset); e != nil { + return n, e } else { - written += n + n += wn + cf.pos += wn } - if size > 0 && written >= size { - break - } chunkStartOffset = 0 } - - return written, nil + return n, nil } -func (c *ChunkedFile) DeleteHelper() error { - //TODO Delete all chunks - return nil +func (cf *ChunkedFileReader) ReadAt(p []byte, off int64) (n int, err error) { + cf.Seek(off, 0) + return cf.Read(p) } -func (c *ChunkedFile) StoredHelper() error { - //TODO - return nil +func (cf *ChunkedFileReader) Read(p []byte) (int, error) { + return cf.getPipeReader().Read(p) +} + +func (cf *ChunkedFileReader) Close() (e error) { + cf.mutex.Lock() + defer cf.mutex.Unlock() + return cf.closePipe() +} + +func (cf *ChunkedFileReader) closePipe() (e error) { + if cf.pr != nil { + if err := cf.pr.Close(); err != nil { + e = err + } + } + cf.pr = nil + if cf.pw != nil { + if err := cf.pw.Close(); err != nil { + e = err + } + } + cf.pw = nil + return e +} + +func (cf *ChunkedFileReader) getPipeReader() io.Reader { + cf.mutex.Lock() + defer cf.mutex.Unlock() + if cf.pr != nil && cf.pw != nil { + return cf.pr + } + cf.closePipe() + cf.pr, cf.pw = io.Pipe() + go func(pw *io.PipeWriter) { + _, e := cf.WriteTo(pw) + pw.CloseWithError(e) + }(cf.pw) + return cf.pr } diff --git a/go/storage/needle.go b/go/storage/needle.go index 04a9dc78d..c09ad061a 100644 --- a/go/storage/needle.go +++ b/go/storage/needle.go @@ -52,7 +52,7 @@ func (n *Needle) String() (str string) { return } -func ParseUpload(r *http.Request) (fileName string, data []byte, mimeType string, isGzipped bool, modifiedTime uint64, ttl *TTL, e error) { +func ParseUpload(r *http.Request) (fileName string, data []byte, mimeType string, isGzipped bool, modifiedTime uint64, ttl *TTL, isChunkedFile bool, e error) { form, fe := r.MultipartReader() if fe != nil { glog.V(0).Infoln("MultipartReader [ERROR]", fe) @@ -132,12 +132,13 @@ func ParseUpload(r *http.Request) (fileName string, data []byte, mimeType string } modifiedTime, _ = strconv.ParseUint(r.FormValue("ts"), 10, 64) ttl, _ = ReadTTL(r.FormValue("ttl")) + isChunkedFile, _ = strconv.ParseBool(r.FormValue("cf")) return } func NewNeedle(r *http.Request, fixJpgOrientation bool) (n *Needle, e error) { - fname, mimeType, isGzipped := "", "", false + fname, mimeType, isGzipped, isChunkedFile := "", "", false, false n = new(Needle) - fname, n.Data, mimeType, isGzipped, n.LastModified, n.Ttl, e = ParseUpload(r) + fname, n.Data, mimeType, isGzipped, n.LastModified, n.Ttl, isChunkedFile, e = ParseUpload(r) if e != nil { return } @@ -160,6 +161,10 @@ func NewNeedle(r *http.Request, fixJpgOrientation bool) (n *Needle, e error) { n.SetHasTtl() } + if isChunkedFile { + n.SetChunkedFile() + } + if fixJpgOrientation { loweredName := strings.ToLower(fname) if mimeType == "image/jpeg" || strings.HasSuffix(loweredName, ".jpg") || strings.HasSuffix(loweredName, ".jpeg") { diff --git a/go/storage/needle_read_write.go b/go/storage/needle_read_write.go index 2a5bd89ea..28b0e8fde 100644 --- a/go/storage/needle_read_write.go +++ b/go/storage/needle_read_write.go @@ -16,7 +16,7 @@ const ( FlagHasMime = 0x04 FlagHasLastModifiedDate = 0x08 FlagHasTtl = 0x10 - FlagChunkList = 0x80 + FlagChunkedFile = 0x80 LastModifiedBytesLength = 5 TtlBytesLength = 2 ) @@ -282,10 +282,10 @@ func (n *Needle) SetHasTtl() { n.Flags = n.Flags | FlagHasTtl } -func (n *Needle) IsChunkList() bool { - return n.Flags&FlagChunkList > 0 +func (n *Needle) IsChunkedFile() bool { + return n.Flags&FlagChunkedFile > 0 } -func (n *Needle) SetChunkList() { - n.Flags = n.Flags | FlagChunkList +func (n *Needle) SetChunkedFile() { + n.Flags = n.Flags | FlagChunkedFile } diff --git a/go/weed/weed_server/common.go b/go/weed/weed_server/common.go index 4ad9824b1..a7fa2de53 100644 --- a/go/weed/weed_server/common.go +++ b/go/weed/weed_server/common.go @@ -86,7 +86,7 @@ func submitForClientHandler(w http.ResponseWriter, r *http.Request, masterUrl st } debug("parsing upload file...") - fname, data, mimeType, isGzipped, lastModified, _, pe := storage.ParseUpload(r) + fname, data, mimeType, isGzipped, lastModified, _, _, pe := storage.ParseUpload(r) if pe != nil { writeJsonError(w, r, http.StatusBadRequest, pe) return diff --git a/go/weed/weed_server/volume_server_handlers_read.go b/go/weed/weed_server/volume_server_handlers_read.go index 9e252d205..febed354b 100644 --- a/go/weed/weed_server/volume_server_handlers_read.go +++ b/go/weed/weed_server/volume_server_handlers_read.go @@ -81,6 +81,11 @@ func (vs *VolumeServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) return } w.Header().Set("Etag", etag) + + if vs.tryHandleChunkedFile(n, filename, w, r) { + return + } + if n.NameSize > 0 && filename == "" { filename = string(n.Name) dotIndex := strings.LastIndex(filename, ".") @@ -215,3 +220,148 @@ func (vs *VolumeServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) io.CopyN(w, sendContent, sendSize) } + +func (vs *VolumeServer) tryHandleChunkedFile(n *storage.Needle, fileName string, w http.ResponseWriter, r *http.Request) (processed bool) { + if !n.IsChunkedFile() { + return false + } + processed = true + raw, _ := strconv.ParseBool(r.FormValue("raw")) + if raw { + w.Header().Set("Content-Type", "application/json") + w.Header().Set("Content-Length", strconv.Itoa(len(n.Data))) + if _, e := w.Write(n.Data); e != nil { + glog.V(0).Infoln("response write error:", e) + } + return true + } + chunkManifest, e := operation.LoadChunkedManifest(n.Data) + if e != nil { + 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) + } + if chunkManifest.Mime != "" { + mt := chunkManifest.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)+`"`) + } + 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() + 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(0).Infoln("response write error:", e) + } + return true + } + + //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) + 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 = chunkedFileReader.Seek(ra.start, 0); e != nil { + glog.V(0).Infoln("response write error:", e) + } + if _, e = io.CopyN(w, chunkedFileReader, ra.length); e != nil { + glog.V(0).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 + } + } + 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 _, e = chunkedFileReader.Seek(ra.start, 0); e != nil { + glog.V(0).Infoln("response write error:", e) + } + if _, err = io.CopyN(part, chunkedFileReader, 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) + return +} From de5e07ce3ec9e8b7d56fcb10b287884a8f4dbcb5 Mon Sep 17 00:00:00 2001 From: tnextday Date: Tue, 1 Dec 2015 20:53:44 +0800 Subject: [PATCH 05/21] rename FlagChunkedFile to FlagChunkManifest --- go/storage/needle.go | 2 +- go/storage/needle_read_write.go | 10 +++++----- go/weed/weed_server/volume_server_handlers_read.go | 2 +- 3 files changed, 7 insertions(+), 7 deletions(-) diff --git a/go/storage/needle.go b/go/storage/needle.go index c09ad061a..c45951feb 100644 --- a/go/storage/needle.go +++ b/go/storage/needle.go @@ -162,7 +162,7 @@ func NewNeedle(r *http.Request, fixJpgOrientation bool) (n *Needle, e error) { } if isChunkedFile { - n.SetChunkedFile() + n.SetChunkManifest() } if fixJpgOrientation { diff --git a/go/storage/needle_read_write.go b/go/storage/needle_read_write.go index 28b0e8fde..9d7af600a 100644 --- a/go/storage/needle_read_write.go +++ b/go/storage/needle_read_write.go @@ -16,7 +16,7 @@ const ( FlagHasMime = 0x04 FlagHasLastModifiedDate = 0x08 FlagHasTtl = 0x10 - FlagChunkedFile = 0x80 + FlagChunkManifest = 0x80 LastModifiedBytesLength = 5 TtlBytesLength = 2 ) @@ -282,10 +282,10 @@ func (n *Needle) SetHasTtl() { n.Flags = n.Flags | FlagHasTtl } -func (n *Needle) IsChunkedFile() bool { - return n.Flags&FlagChunkedFile > 0 +func (n *Needle) IsChunkedManifest() bool { + return n.Flags&FlagChunkManifest > 0 } -func (n *Needle) SetChunkedFile() { - n.Flags = n.Flags | FlagChunkedFile +func (n *Needle) SetChunkManifest() { + n.Flags = n.Flags | FlagChunkManifest } diff --git a/go/weed/weed_server/volume_server_handlers_read.go b/go/weed/weed_server/volume_server_handlers_read.go index febed354b..36c655bd5 100644 --- a/go/weed/weed_server/volume_server_handlers_read.go +++ b/go/weed/weed_server/volume_server_handlers_read.go @@ -222,7 +222,7 @@ func (vs *VolumeServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) } func (vs *VolumeServer) tryHandleChunkedFile(n *storage.Needle, fileName string, w http.ResponseWriter, r *http.Request) (processed bool) { - if !n.IsChunkedFile() { + if !n.IsChunkedManifest() { return false } processed = true From 2c0a7fe75e1e3c1a0f04ab22414d3054a3095078 Mon Sep 17 00:00:00 2001 From: tnextday Date: Wed, 2 Dec 2015 15:00:46 +0800 Subject: [PATCH 06/21] complete submit chunked file --- go/operation/chunked_file.go | 7 ++- go/operation/submit.go | 43 +++++++++++++++---- go/storage/needle.go | 2 +- .../volume_server_handlers_read.go | 2 +- 4 files changed, 40 insertions(+), 14 deletions(-) diff --git a/go/operation/chunked_file.go b/go/operation/chunked_file.go index ca97566a5..0492d39a7 100644 --- a/go/operation/chunked_file.go +++ b/go/operation/chunked_file.go @@ -85,10 +85,9 @@ func (cm *ChunkManifest) DeleteChunks(master string) error { return nil } -func (cm *ChunkManifest) StoredHelper() error { - //TODO - return nil -} +//func (cm *ChunkManifest) StoredHelper() error { +// return nil +//} func httpRangeDownload(fileUrl string, w io.Writer, offset int64) (written int64, e error) { req, err := http.NewRequest("GET", fileUrl, nil) diff --git a/go/operation/submit.go b/go/operation/submit.go index fec5d3801..8af36c722 100644 --- a/go/operation/submit.go +++ b/go/operation/submit.go @@ -9,6 +9,8 @@ import ( "strconv" "strings" + "net/url" + "github.com/chrislusf/seaweedfs/go/glog" "github.com/chrislusf/seaweedfs/go/security" ) @@ -117,7 +119,13 @@ func (fi FilePart) Upload(maxMB int, master string, secret security.Secret) (ret if maxMB > 0 && fi.FileSize > int64(maxMB*1024*1024) { chunkSize := int64(maxMB * 1024 * 1024) chunks := fi.FileSize/chunkSize + 1 - var fids []string + cm := ChunkManifest{ + Name: fi.FileName, + Size: fi.FileSize, + Mime: fi.MimeType, + Chunks: make([]*ChunkInfo, 0, chunks), + } + for i := int64(0); i < chunks; i++ { id, count, e := upload_one_chunk( fi.FileName+"-"+strconv.FormatInt(i+1, 10), @@ -125,12 +133,24 @@ func (fi FilePart) Upload(maxMB int, master string, secret security.Secret) (ret master, fi.Replication, fi.Collection, fi.Ttl, jwt) if e != nil { + // delete all uploaded chunks + cm.DeleteChunks(master) return 0, e } - fids = append(fids, id) + cm.Chunks = append(cm.Chunks, + &ChunkInfo{ + Offset: i * chunkSize, + Size: int64(count), + Fid: id, + }, + ) retSize += count } - err = upload_file_id_list(fileUrl, fi.FileName+"-list", fids, jwt) + err = upload_chunked_file_manifest(fileUrl, &cm, jwt) + if err != nil { + // delete all uploaded chunks + cm.DeleteChunks(master) + } } else { ret, e := Upload(fileUrl, fi.FileName, fi.Reader, fi.IsGzipped, fi.MimeType, jwt) if e != nil { @@ -158,10 +178,17 @@ func upload_one_chunk(filename string, reader io.Reader, master, return fid, uploadResult.Size, nil } -func upload_file_id_list(fileUrl, filename string, fids []string, jwt security.EncodedJwt) error { - var buf bytes.Buffer - buf.WriteString(strings.Join(fids, "\n")) - glog.V(4).Info("Uploading final list ", filename, " to ", fileUrl, "...") - _, e := Upload(fileUrl, filename, &buf, false, "text/plain", jwt) +func upload_chunked_file_manifest(fileUrl string, manifest *ChunkManifest, jwt security.EncodedJwt) error { + buf, e := manifest.GetData() + if e != nil { + return e + } + bufReader := bytes.NewReader(buf) + glog.V(4).Info("Uploading chunks manifest ", manifest.Name, " to ", fileUrl, "...") + u, _ := url.Parse(fileUrl) + q := u.Query() + q.Set("cm", "1") + u.RawQuery = q.Encode() + _, e = Upload(u.String(), manifest.Name, bufReader, false, "text/plain", jwt) return e } diff --git a/go/storage/needle.go b/go/storage/needle.go index c45951feb..e1902e461 100644 --- a/go/storage/needle.go +++ b/go/storage/needle.go @@ -132,7 +132,7 @@ func ParseUpload(r *http.Request) (fileName string, data []byte, mimeType string } modifiedTime, _ = strconv.ParseUint(r.FormValue("ts"), 10, 64) ttl, _ = ReadTTL(r.FormValue("ttl")) - isChunkedFile, _ = strconv.ParseBool(r.FormValue("cf")) + isChunkedFile, _ = strconv.ParseBool(r.FormValue("cm")) return } func NewNeedle(r *http.Request, fixJpgOrientation bool) (n *Needle, e error) { diff --git a/go/weed/weed_server/volume_server_handlers_read.go b/go/weed/weed_server/volume_server_handlers_read.go index 36c655bd5..d2375099f 100644 --- a/go/weed/weed_server/volume_server_handlers_read.go +++ b/go/weed/weed_server/volume_server_handlers_read.go @@ -270,7 +270,7 @@ func (vs *VolumeServer) tryHandleChunkedFile(n *storage.Needle, fileName string, return true } - chunkedFileReader := operation.ChunkedFileReader{ + chunkedFileReader := &operation.ChunkedFileReader{ Manifest: chunkManifest, Master: vs.GetMasterNode(), } From d0e2475ecea97ac96ddbdc76f98a5de32e4a68ad Mon Sep 17 00:00:00 2001 From: tnextday Date: Wed, 2 Dec 2015 16:35:16 +0800 Subject: [PATCH 07/21] `FilePart.Upload` use base name instead of full path fix chunked file download error --- go/operation/chunked_file.go | 8 ++++---- go/operation/submit.go | 7 ++++--- .../weed_server/volume_server_handlers_read.go | 18 +++++++++++------- 3 files changed, 19 insertions(+), 14 deletions(-) diff --git a/go/operation/chunked_file.go b/go/operation/chunked_file.go index 0492d39a7..0e455e93a 100644 --- a/go/operation/chunked_file.go +++ b/go/operation/chunked_file.go @@ -22,9 +22,9 @@ var ( ) type ChunkInfo struct { - Fid string `json:"fid,omitempty"` - Offset int64 `json:"offset,omitempty"` - Size int64 `json:"size,omitempty"` + Fid string `json:"fid"` + Offset int64 `json:"offset"` + Size int64 `json:"size"` } type ChunkList []*ChunkInfo @@ -52,7 +52,7 @@ func (s ChunkList) Swap(i, j int) { s[i], s[j] = s[j], s[i] } func LoadChunkedManifest(buffer []byte) (*ChunkManifest, error) { cm := ChunkManifest{} - if e := json.Unmarshal(buffer, cm); e != nil { + if e := json.Unmarshal(buffer, &cm); e != nil { return nil, e } sort.Sort(cm.Chunks) diff --git a/go/operation/submit.go b/go/operation/submit.go index 8af36c722..65f20cf55 100644 --- a/go/operation/submit.go +++ b/go/operation/submit.go @@ -116,11 +116,12 @@ func (fi FilePart) Upload(maxMB int, master string, secret security.Secret) (ret if closer, ok := fi.Reader.(io.Closer); ok { defer closer.Close() } + baseName := path.Base(fi.FileName) if maxMB > 0 && fi.FileSize > int64(maxMB*1024*1024) { chunkSize := int64(maxMB * 1024 * 1024) chunks := fi.FileSize/chunkSize + 1 cm := ChunkManifest{ - Name: fi.FileName, + Name: baseName, Size: fi.FileSize, Mime: fi.MimeType, Chunks: make([]*ChunkInfo, 0, chunks), @@ -128,7 +129,7 @@ func (fi FilePart) Upload(maxMB int, master string, secret security.Secret) (ret for i := int64(0); i < chunks; i++ { id, count, e := upload_one_chunk( - fi.FileName+"-"+strconv.FormatInt(i+1, 10), + baseName+"-"+strconv.FormatInt(i+1, 10), io.LimitReader(fi.Reader, chunkSize), master, fi.Replication, fi.Collection, fi.Ttl, jwt) @@ -152,7 +153,7 @@ func (fi FilePart) Upload(maxMB int, master string, secret security.Secret) (ret cm.DeleteChunks(master) } } else { - ret, e := Upload(fileUrl, fi.FileName, fi.Reader, fi.IsGzipped, fi.MimeType, jwt) + ret, e := Upload(fileUrl, baseName, fi.Reader, fi.IsGzipped, fi.MimeType, jwt) if e != nil { return 0, e } diff --git a/go/weed/weed_server/volume_server_handlers_read.go b/go/weed/weed_server/volume_server_handlers_read.go index d2375099f..6a7944d25 100644 --- a/go/weed/weed_server/volume_server_handlers_read.go +++ b/go/weed/weed_server/volume_server_handlers_read.go @@ -225,18 +225,22 @@ func (vs *VolumeServer) tryHandleChunkedFile(n *storage.Needle, fileName string, if !n.IsChunkedManifest() { return false } - processed = true raw, _ := strconv.ParseBool(r.FormValue("raw")) if raw { - w.Header().Set("Content-Type", "application/json") - w.Header().Set("Content-Length", strconv.Itoa(len(n.Data))) - if _, e := w.Write(n.Data); e != nil { - glog.V(0).Infoln("response write error:", e) - } - return true + return false } + processed = true + if n.IsGzipped(){ + var err error + if n.Data, err = storage.UnGzipData(n.Data); err != nil { + glog.V(0).Infoln("ungzip data error:", err, r.URL.Path) + return false + } + } + chunkManifest, e := operation.LoadChunkedManifest(n.Data) if e != nil { + glog.V(0).Infoln("load chunked manifest error:", e) return false } ext := "" From 520875d45504f0b659de7f4ff656634489100416 Mon Sep 17 00:00:00 2001 From: tnextday Date: Wed, 2 Dec 2015 18:25:18 +0800 Subject: [PATCH 08/21] update --- go/operation/submit.go | 2 +- go/weed/weed_server/volume_server_handlers_read.go | 10 +++++----- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/go/operation/submit.go b/go/operation/submit.go index 65f20cf55..ac5a3b55f 100644 --- a/go/operation/submit.go +++ b/go/operation/submit.go @@ -190,6 +190,6 @@ func upload_chunked_file_manifest(fileUrl string, manifest *ChunkManifest, jwt s q := u.Query() q.Set("cm", "1") u.RawQuery = q.Encode() - _, e = Upload(u.String(), manifest.Name, bufReader, false, "text/plain", jwt) + _, e = Upload(u.String(), manifest.Name, bufReader, false, "application/json", jwt) return e } diff --git a/go/weed/weed_server/volume_server_handlers_read.go b/go/weed/weed_server/volume_server_handlers_read.go index 6a7944d25..34b0720b9 100644 --- a/go/weed/weed_server/volume_server_handlers_read.go +++ b/go/weed/weed_server/volume_server_handlers_read.go @@ -140,7 +140,7 @@ func (vs *VolumeServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) if rangeReq == "" { w.Header().Set("Content-Length", strconv.Itoa(len(n.Data))) if _, e = w.Write(n.Data); e != nil { - glog.V(0).Infoln("response write error:", e) + glog.V(4).Infoln("response write error:", e) } return } @@ -181,7 +181,7 @@ func (vs *VolumeServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) 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(0).Infoln("response write error:", e) + glog.V(4).Infoln("response write error:", e) } return } @@ -324,10 +324,10 @@ func (vs *VolumeServer) tryHandleChunkedFile(n *storage.Needle, fileName string, w.Header().Set("Content-Range", ra.contentRange(size)) w.WriteHeader(http.StatusPartialContent) if _, e = chunkedFileReader.Seek(ra.start, 0); e != nil { - glog.V(0).Infoln("response write error:", e) + glog.V(0).Infoln("chunkedFileReader Seek error:", e) } if _, e = io.CopyN(w, chunkedFileReader, ra.length); e != nil { - glog.V(0).Infoln("response write error:", e) + glog.V(4).Infoln("response write error:", e) } return } @@ -352,7 +352,7 @@ func (vs *VolumeServer) tryHandleChunkedFile(n *storage.Needle, fileName string, return } if _, e = chunkedFileReader.Seek(ra.start, 0); e != nil { - glog.V(0).Infoln("response write error:", e) + glog.V(4).Infoln("response write error:", e) } if _, err = io.CopyN(part, chunkedFileReader, ra.length); err != nil { pw.CloseWithError(err) From 662915e6915d18a9e45b73577f93ef3be6562f4e Mon Sep 17 00:00:00 2001 From: tnextday Date: Wed, 2 Dec 2015 21:27:29 +0800 Subject: [PATCH 09/21] Delete all chunks when delete a ChunkManifest LoadChunkManifest can uncompress buffer move compress.go from storage to operation because of import cycle MakeFile add cross complete command --- .gitignore | 1 + Makefile | 25 +++++++++++++------ go/operation/chunked_file.go | 16 ++++++++---- go/{storage => operation}/compress.go | 2 +- go/storage/needle.go | 5 ++-- .../volume_server_handlers_read.go | 11 ++------ .../volume_server_handlers_write.go | 11 ++++++++ 7 files changed, 47 insertions(+), 24 deletions(-) rename go/{storage => operation}/compress.go (98%) diff --git a/.gitignore b/.gitignore index 360cbb2cb..57fd53063 100644 --- a/.gitignore +++ b/.gitignore @@ -74,3 +74,4 @@ com_crashlytics_export_strings.xml crashlytics.properties crashlytics-build.properties +test_data diff --git a/Makefile b/Makefile index 25a6a39cd..6719a7bdd 100644 --- a/Makefile +++ b/Makefile @@ -1,11 +1,22 @@ +BINARY = weed -.clean: - go clean -i -v ./go/weed/ +GO_FLAGS = #-v +SOURCE_DIR = ./go/weed/ -.deps: - go get -d ./go/weed/ +all: build -.build: .deps - go build -v ./go/weed/ +.PHONY : clean deps build linux -all: .build +clean: + go clean -i $(GO_FLAGS) $(SOURCE_DIR) + rm -f $(BINARY) + +deps: + go get $(GO_FLAGS) -d $(SOURCE_DIR) + +build: deps + go build $(GO_FLAGS) -o $(BINARY) $(SOURCE_DIR) + +linux: deps + mkdir -p linux + GOOS=linux GOARCH=amd64 go build $(GO_FLAGS) -o linux/$(BINARY) $(SOURCE_DIR) diff --git a/go/operation/chunked_file.go b/go/operation/chunked_file.go index 0e455e93a..33cb25703 100644 --- a/go/operation/chunked_file.go +++ b/go/operation/chunked_file.go @@ -30,10 +30,10 @@ type ChunkInfo struct { type ChunkList []*ChunkInfo type ChunkManifest struct { - Name string `json:"name,omitempty"` - Mime string `json:"mime,omitempty"` - Size int64 `json:"size,omitempty"` - Chunks ChunkList `json:"chunks,omitempty"` + Name string `json:"name,omitempty"` + Mime string `json:"mime,omitempty"` + Size int64 `json:"size,omitempty"` + Chunks ChunkList `json:"chunks,omitempty"` } // seekable chunked file reader @@ -50,7 +50,13 @@ func (s ChunkList) Len() int { return len(s) } func (s ChunkList) Less(i, j int) bool { return s[i].Offset < s[j].Offset } func (s ChunkList) Swap(i, j int) { s[i], s[j] = s[j], s[i] } -func LoadChunkedManifest(buffer []byte) (*ChunkManifest, error) { +func LoadChunkManifest(buffer []byte, isGzipped bool) (*ChunkManifest, error) { + if isGzipped { + var err error + if buffer, err = UnGzipData(buffer); err != nil { + return nil, err + } + } cm := ChunkManifest{} if e := json.Unmarshal(buffer, &cm); e != nil { return nil, e diff --git a/go/storage/compress.go b/go/operation/compress.go similarity index 98% rename from go/storage/compress.go rename to go/operation/compress.go index 4047c1723..b1105ba4b 100644 --- a/go/storage/compress.go +++ b/go/operation/compress.go @@ -1,4 +1,4 @@ -package storage +package operation import ( "bytes" diff --git a/go/storage/needle.go b/go/storage/needle.go index e1902e461..32ebdae7d 100644 --- a/go/storage/needle.go +++ b/go/storage/needle.go @@ -15,6 +15,7 @@ import ( "github.com/chrislusf/seaweedfs/go/glog" "github.com/chrislusf/seaweedfs/go/images" "github.com/chrislusf/seaweedfs/go/util" + "github.com/chrislusf/seaweedfs/go/operation" ) const ( @@ -117,8 +118,8 @@ func ParseUpload(r *http.Request) (fileName string, data []byte, mimeType string } if part.Header.Get("Content-Encoding") == "gzip" { isGzipped = true - } else if IsGzippable(ext, mtype) { - if data, e = GzipData(data); e != nil { + } else if operation.IsGzippable(ext, mtype) { + if data, e = operation.GzipData(data); e != nil { return } isGzipped = true diff --git a/go/weed/weed_server/volume_server_handlers_read.go b/go/weed/weed_server/volume_server_handlers_read.go index 34b0720b9..6b40194ff 100644 --- a/go/weed/weed_server/volume_server_handlers_read.go +++ b/go/weed/weed_server/volume_server_handlers_read.go @@ -114,7 +114,7 @@ func (vs *VolumeServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) if strings.Contains(r.Header.Get("Accept-Encoding"), "gzip") { w.Header().Set("Content-Encoding", "gzip") } else { - if n.Data, err = storage.UnGzipData(n.Data); err != nil { + if n.Data, err = operation.UnGzipData(n.Data); err != nil { glog.V(0).Infoln("lookup error:", err, r.URL.Path) } } @@ -230,15 +230,8 @@ func (vs *VolumeServer) tryHandleChunkedFile(n *storage.Needle, fileName string, return false } processed = true - if n.IsGzipped(){ - var err error - if n.Data, err = storage.UnGzipData(n.Data); err != nil { - glog.V(0).Infoln("ungzip data error:", err, r.URL.Path) - return false - } - } - chunkManifest, e := operation.LoadChunkedManifest(n.Data) + chunkManifest, e := operation.LoadChunkManifest(n.Data, n.IsGzipped()) if e != nil { glog.V(0).Infoln("load chunked manifest error:", e) return false diff --git a/go/weed/weed_server/volume_server_handlers_write.go b/go/weed/weed_server/volume_server_handlers_write.go index 2f7e79ce9..911ed96fa 100644 --- a/go/weed/weed_server/volume_server_handlers_write.go +++ b/go/weed/weed_server/volume_server_handlers_write.go @@ -66,6 +66,17 @@ func (vs *VolumeServer) DeleteHandler(w http.ResponseWriter, r *http.Request) { glog.V(0).Infoln("delete", r.URL.Path, "with unmaching cookie from ", r.RemoteAddr, "agent", r.UserAgent()) return } + if n.IsChunkedManifest(){ + chunkManifest, e := operation.LoadChunkManifest(n.Data, n.IsGzipped()) + if e != nil { + writeJsonError(w, r, http.StatusInternalServerError, errors.New("Load chunks manifest error: " + e.Error())) + return + } + if e := chunkManifest.DeleteChunks(vs.GetMasterNode()); e != nil { + writeJsonError(w, r, http.StatusInternalServerError, errors.New("Delete chunks error: " + e.Error())) + return + } + } ret := topology.ReplicatedDelete(vs.GetMasterNode(), vs.store, volumeId, n, r) From 34963de45876787ac41bde6c99e163256bb494b0 Mon Sep 17 00:00:00 2001 From: tnextday Date: Wed, 2 Dec 2015 21:35:50 +0800 Subject: [PATCH 10/21] returen chunks file size when delete a chunkmanifest --- go/weed/weed_server/volume_server_handlers_write.go | 1 + 1 file changed, 1 insertion(+) diff --git a/go/weed/weed_server/volume_server_handlers_write.go b/go/weed/weed_server/volume_server_handlers_write.go index 911ed96fa..9aa998c3c 100644 --- a/go/weed/weed_server/volume_server_handlers_write.go +++ b/go/weed/weed_server/volume_server_handlers_write.go @@ -76,6 +76,7 @@ func (vs *VolumeServer) DeleteHandler(w http.ResponseWriter, r *http.Request) { writeJsonError(w, r, http.StatusInternalServerError, errors.New("Delete chunks error: " + e.Error())) return } + count = chunkManifest.Size } ret := topology.ReplicatedDelete(vs.GetMasterNode(), vs.store, volumeId, n, r) From a9a336fdff076211e7c94a4243b430219e5d891c Mon Sep 17 00:00:00 2001 From: tnextday Date: Wed, 2 Dec 2015 21:45:26 +0800 Subject: [PATCH 11/21] delete file return needle size --- go/weed/weed_server/volume_server_handlers_write.go | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/go/weed/weed_server/volume_server_handlers_write.go b/go/weed/weed_server/volume_server_handlers_write.go index 9aa998c3c..c891b5af7 100644 --- a/go/weed/weed_server/volume_server_handlers_write.go +++ b/go/weed/weed_server/volume_server_handlers_write.go @@ -53,9 +53,8 @@ func (vs *VolumeServer) DeleteHandler(w http.ResponseWriter, r *http.Request) { glog.V(2).Infoln("deleting", n) cookie := n.Cookie - count, ok := vs.store.ReadVolumeNeedle(volumeId, n) - if ok != nil { + if _, ok := vs.store.ReadVolumeNeedle(volumeId, n); ok != nil { m := make(map[string]uint32) m["size"] = 0 writeJsonQuiet(w, r, http.StatusNotFound, m) @@ -66,6 +65,9 @@ func (vs *VolumeServer) DeleteHandler(w http.ResponseWriter, r *http.Request) { glog.V(0).Infoln("delete", r.URL.Path, "with unmaching cookie from ", r.RemoteAddr, "agent", r.UserAgent()) return } + + count := int64(n.Size) + if n.IsChunkedManifest(){ chunkManifest, e := operation.LoadChunkManifest(n.Data, n.IsGzipped()) if e != nil { @@ -82,8 +84,8 @@ func (vs *VolumeServer) DeleteHandler(w http.ResponseWriter, r *http.Request) { ret := topology.ReplicatedDelete(vs.GetMasterNode(), vs.store, volumeId, n, r) if ret != 0 { - m := make(map[string]uint32) - m["size"] = uint32(count) + m := make(map[string]int64) + m["size"] = count writeJsonQuiet(w, r, http.StatusAccepted, m) } else { writeJsonError(w, r, http.StatusInternalServerError, errors.New("Deletion Failed.")) From daac5de1bae20de28082367db44b7b7afa9e4aaf Mon Sep 17 00:00:00 2001 From: tnextday Date: Thu, 3 Dec 2015 16:27:02 +0800 Subject: [PATCH 12/21] more check in `http_util.Delete` add status code in `DeleteResult` struct operation.DeleteFiles maybe unsafe, so `ChunkManifest.DeleteChunks` manually delete each chunks --- go/operation/chunked_file.go | 14 ++---- go/operation/delete_content.go | 16 +++++-- go/util/http_util.go | 18 ++++++- .../volume_server_handlers_write.go | 47 +++++++++++++++---- 4 files changed, 70 insertions(+), 25 deletions(-) diff --git a/go/operation/chunked_file.go b/go/operation/chunked_file.go index 33cb25703..dbc450fd8 100644 --- a/go/operation/chunked_file.go +++ b/go/operation/chunked_file.go @@ -70,19 +70,11 @@ func (cm *ChunkManifest) GetData() ([]byte, error) { } func (cm *ChunkManifest) DeleteChunks(master string) error { - fileIds := make([]string, 0, len(cm.Chunks)) - for _, ci := range cm.Chunks { - fileIds = append(fileIds, ci.Fid) - } - results, e := DeleteFiles(master, fileIds) - if e != nil { - return e - } deleteError := 0 - for _, ret := range results.Results { - if ret.Error != "" { + for _, ci := range cm.Chunks { + if e := DeleteFile(master, ci.Fid, ""); e != nil { deleteError++ - glog.V(0).Infoln("delete error:", ret.Error, ret.Fid) + glog.V(0).Infoln("delete error:", e, ci.Fid) } } if deleteError > 0 { diff --git a/go/operation/delete_content.go b/go/operation/delete_content.go index ac3dfa6b2..32ad69b17 100644 --- a/go/operation/delete_content.go +++ b/go/operation/delete_content.go @@ -7,14 +7,17 @@ import ( "strings" "sync" + "net/http" + "github.com/chrislusf/seaweedfs/go/security" "github.com/chrislusf/seaweedfs/go/util" ) type DeleteResult struct { - Fid string `json:"fid"` - Size int `json:"size"` - Error string `json:"error,omitempty"` + Fid string `json:"fid"` + Size int `json:"size"` + Status int `json:"status"` + Error string `json:"error,omitempty"` } func DeleteFile(master string, fileId string, jwt security.EncodedJwt) error { @@ -45,7 +48,11 @@ func DeleteFiles(master string, fileIds []string) (*DeleteFilesResult, error) { for _, fileId := range fileIds { vid, _, err := ParseFileId(fileId) if err != nil { - ret.Results = append(ret.Results, DeleteResult{Fid: vid, Error: err.Error()}) + ret.Results = append(ret.Results, DeleteResult{ + Fid: vid, + Status: http.StatusBadRequest, + Error: err.Error()}, + ) continue } if _, ok := vid_to_fileIds[vid]; !ok { @@ -76,6 +83,7 @@ func DeleteFiles(master string, fileIds []string) (*DeleteFilesResult, error) { } var wg sync.WaitGroup + for server, fidList := range server_to_fileIds { wg.Add(1) go func(server string, fidList []string) { diff --git a/go/util/http_util.go b/go/util/http_util.go index 7854302ab..d56aaa39a 100644 --- a/go/util/http_util.go +++ b/go/util/http_util.go @@ -9,7 +9,10 @@ import ( "net/url" "strings" + "encoding/json" + "github.com/chrislusf/seaweedfs/go/security" + "github.com/syndtr/goleveldb/leveldb/errors" ) var ( @@ -79,10 +82,21 @@ func Delete(url string, jwt security.EncodedJwt) error { return e } defer resp.Body.Close() - if _, err := ioutil.ReadAll(resp.Body); err != nil { + body, err := ioutil.ReadAll(resp.Body) + if err != nil { return err } - return nil + switch resp.StatusCode { + case http.StatusNotFound, http.StatusAccepted, http.StatusOK: + return nil + } + m := make(map[string]interface{}) + if e := json.Unmarshal(body, m); e == nil { + if s, ok := m["error"].(string); ok { + return errors.New(s) + } + } + return errors.New(string(body)) } func GetBufferStream(url string, values url.Values, allocatedBytes []byte, eachBuffer func([]byte)) error { diff --git a/go/weed/weed_server/volume_server_handlers_write.go b/go/weed/weed_server/volume_server_handlers_write.go index c891b5af7..cd4a4c673 100644 --- a/go/weed/weed_server/volume_server_handlers_write.go +++ b/go/weed/weed_server/volume_server_handlers_write.go @@ -63,19 +63,20 @@ func (vs *VolumeServer) DeleteHandler(w http.ResponseWriter, r *http.Request) { if n.Cookie != cookie { glog.V(0).Infoln("delete", r.URL.Path, "with unmaching cookie from ", r.RemoteAddr, "agent", r.UserAgent()) + writeJsonError(w, r, http.StatusBadRequest, errors.New("File Random Cookie does not match.")) return } count := int64(n.Size) - if n.IsChunkedManifest(){ + if n.IsChunkedManifest() { chunkManifest, e := operation.LoadChunkManifest(n.Data, n.IsGzipped()) if e != nil { - writeJsonError(w, r, http.StatusInternalServerError, errors.New("Load chunks manifest error: " + e.Error())) + writeJsonError(w, r, http.StatusInternalServerError, errors.New("Load chunks manifest error: "+e.Error())) return } if e := chunkManifest.DeleteChunks(vs.GetMasterNode()); e != nil { - writeJsonError(w, r, http.StatusInternalServerError, errors.New("Delete chunks error: " + e.Error())) + writeJsonError(w, r, http.StatusInternalServerError, errors.New("Delete chunks error: "+e.Error())) return } count = chunkManifest.Size @@ -100,7 +101,10 @@ func (vs *VolumeServer) batchDeleteHandler(w http.ResponseWriter, r *http.Reques for _, fid := range r.Form["fid"] { vid, id_cookie, err := operation.ParseFileId(fid) if err != nil { - ret = append(ret, operation.DeleteResult{Fid: fid, Error: err.Error()}) + ret = append(ret, operation.DeleteResult{ + Fid: fid, + Status: http.StatusBadRequest, + Error: err.Error()}) continue } n := new(storage.Needle) @@ -109,18 +113,45 @@ func (vs *VolumeServer) batchDeleteHandler(w http.ResponseWriter, r *http.Reques glog.V(4).Infoln("batch deleting", n) cookie := n.Cookie if _, err := vs.store.ReadVolumeNeedle(volumeId, n); err != nil { - ret = append(ret, operation.DeleteResult{Fid: fid, Error: err.Error()}) + ret = append(ret, operation.DeleteResult{ + Fid: fid, + Status: http.StatusNotFound, + Error: err.Error(), + }) continue } + + if n.IsChunkedManifest() { + //Don't allow delete manifest in batch delete mode + ret = append(ret, operation.DeleteResult{ + Fid: fid, + Status: http.StatusNotAcceptable, + Error: "ChunkManifest: not allow.", + }) + continue + } + if n.Cookie != cookie { - ret = append(ret, operation.DeleteResult{Fid: fid, Error: "File Random Cookie does not match."}) + ret = append(ret, operation.DeleteResult{ + Fid: fid, + Status: http.StatusBadRequest, + Error: "File Random Cookie does not match.", + }) glog.V(0).Infoln("deleting", fid, "with unmaching cookie from ", r.RemoteAddr, "agent", r.UserAgent()) return } if size, err := vs.store.Delete(volumeId, n); err != nil { - ret = append(ret, operation.DeleteResult{Fid: fid, Error: err.Error()}) + ret = append(ret, operation.DeleteResult{ + Fid: fid, + Status: http.StatusInternalServerError, + Error: err.Error()}, + ) } else { - ret = append(ret, operation.DeleteResult{Fid: fid, Size: int(size)}) + ret = append(ret, operation.DeleteResult{ + Fid: fid, + Status: http.StatusAccepted, + Size: int(size)}, + ) } } From 4fda5f8bfdd8973cc08e5f90c32e3389395ff066 Mon Sep 17 00:00:00 2001 From: tnextday Date: Thu, 3 Dec 2015 18:21:58 +0800 Subject: [PATCH 13/21] update --- go/weed/weed_server/volume_server_handlers_read.go | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/go/weed/weed_server/volume_server_handlers_read.go b/go/weed/weed_server/volume_server_handlers_read.go index 6b40194ff..ffbd717bb 100644 --- a/go/weed/weed_server/volume_server_handlers_read.go +++ b/go/weed/weed_server/volume_server_handlers_read.go @@ -181,7 +181,7 @@ func (vs *VolumeServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) 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(4).Infoln("response write error:", e) + glog.V(2).Infoln("response write error:", e) } return } @@ -276,7 +276,7 @@ func (vs *VolumeServer) tryHandleChunkedFile(n *storage.Needle, fileName string, if rangeReq == "" { w.Header().Set("Content-Length", strconv.FormatInt(chunkManifest.Size, 10)) if _, e = io.Copy(w, chunkedFileReader); e != nil { - glog.V(0).Infoln("response write error:", e) + glog.V(2).Infoln("response write error:", e) } return true } @@ -317,10 +317,10 @@ func (vs *VolumeServer) tryHandleChunkedFile(n *storage.Needle, fileName string, w.Header().Set("Content-Range", ra.contentRange(size)) w.WriteHeader(http.StatusPartialContent) if _, e = chunkedFileReader.Seek(ra.start, 0); e != nil { - glog.V(0).Infoln("chunkedFileReader Seek error:", e) + glog.V(2).Infoln("chunkedFileReader Seek error:", e) } if _, e = io.CopyN(w, chunkedFileReader, ra.length); e != nil { - glog.V(4).Infoln("response write error:", e) + glog.V(2).Infoln("response write error:", e) } return } @@ -345,7 +345,7 @@ func (vs *VolumeServer) tryHandleChunkedFile(n *storage.Needle, fileName string, return } if _, e = chunkedFileReader.Seek(ra.start, 0); e != nil { - glog.V(4).Infoln("response write error:", e) + glog.V(2).Infoln("response write error:", e) } if _, err = io.CopyN(part, chunkedFileReader, ra.length); err != nil { pw.CloseWithError(err) From 82e6d67ccca361a4841b9e40548c9d0d23981423 Mon Sep 17 00:00:00 2001 From: tnextday Date: Thu, 3 Dec 2015 21:35:33 +0800 Subject: [PATCH 14/21] update --- go/operation/chunked_file.go | 6 +++--- go/weed/weed_server/volume_server_handlers_write.go | 1 + 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/go/operation/chunked_file.go b/go/operation/chunked_file.go index dbc450fd8..f8266087b 100644 --- a/go/operation/chunked_file.go +++ b/go/operation/chunked_file.go @@ -87,7 +87,7 @@ func (cm *ChunkManifest) DeleteChunks(master string) error { // return nil //} -func httpRangeDownload(fileUrl string, w io.Writer, offset int64) (written int64, e error) { +func readChunkNeedle(fileUrl string, w io.Writer, offset int64) (written int64, e error) { req, err := http.NewRequest("GET", fileUrl, nil) if err != nil { return written, err @@ -112,7 +112,7 @@ func httpRangeDownload(fileUrl string, w io.Writer, offset int64) (written int64 case http.StatusPartialContent: break default: - return written, fmt.Errorf("Read Needle http error: [%d] %s", resp.StatusCode, fileUrl) + return written, fmt.Errorf("Read chunk needle error: [%d] %s", resp.StatusCode, fileUrl) } return io.Copy(w, resp.Body) @@ -158,7 +158,7 @@ func (cf *ChunkedFileReader) WriteTo(w io.Writer) (n int64, err error) { if lookupError != nil { return n, lookupError } - if wn, e := httpRangeDownload(fileUrl, w, chunkStartOffset); e != nil { + if wn, e := readChunkNeedle(fileUrl, w, chunkStartOffset); e != nil { return n, e } else { n += wn diff --git a/go/weed/weed_server/volume_server_handlers_write.go b/go/weed/weed_server/volume_server_handlers_write.go index cd4a4c673..1f0fa96dd 100644 --- a/go/weed/weed_server/volume_server_handlers_write.go +++ b/go/weed/weed_server/volume_server_handlers_write.go @@ -75,6 +75,7 @@ func (vs *VolumeServer) DeleteHandler(w http.ResponseWriter, r *http.Request) { writeJsonError(w, r, http.StatusInternalServerError, errors.New("Load chunks manifest error: "+e.Error())) return } + // make sure all chunks had deleted before delete manifest if e := chunkManifest.DeleteChunks(vs.GetMasterNode()); e != nil { writeJsonError(w, r, http.StatusInternalServerError, errors.New("Delete chunks error: "+e.Error())) return From f240c57f16e89a03be9ca583ab25703421e82e1e Mon Sep 17 00:00:00 2001 From: tnextday Date: Mon, 14 Dec 2015 22:01:30 +0800 Subject: [PATCH 15/21] 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 16/21] 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 17/21] `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 | 100 ++++++++++++++++++++++------------- 3 files changed, 66 insertions(+), 43 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 filename == "" { - filename = fid - } - 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 - } - 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 - } - } - } else { - ioutil.WriteFile(path.Join(*d.dir, filename), content, os.ModePerm) + if e := downloadToFile(*d.server, fid, *d.dir); e != nil { + fmt.Println("Download Error:", e) } } return true } -func fetchFileId(server string, fileId string) (filename string, content []byte, e error) { +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 + } + 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) + } + if err == nil && n < len(part) { + err = io.ErrShortWrite + } + if err != nil { + return err + } + } + } else { + if _, err = io.Copy(f, rc); err != nil { + return err + } + + } + return nil +} + +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 18/21] 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 19/21] 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 + if e := writeResponseContent(filename, mtype, bytes.NewReader(n.Data), w, r); e != nil { + glog.V(2).Infoln("response write error:", e) } - 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 - } - } - 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 20/21] 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 21/21] 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 != "" {