aboutsummaryrefslogtreecommitdiff
path: root/go/operation
diff options
context:
space:
mode:
Diffstat (limited to 'go/operation')
-rw-r--r--go/operation/chunked_file.go213
-rw-r--r--go/operation/compress.go59
-rw-r--r--go/operation/delete_content.go16
-rw-r--r--go/operation/submit.go48
4 files changed, 322 insertions, 14 deletions
diff --git a/go/operation/chunked_file.go b/go/operation/chunked_file.go
new file mode 100644
index 000000000..70564cbd2
--- /dev/null
+++ b/go/operation/chunked_file.go
@@ -0,0 +1,213 @@
+package operation
+
+import (
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "net/http"
+ "sort"
+
+ "sync"
+
+ "github.com/chrislusf/seaweedfs/go/glog"
+ "github.com/chrislusf/seaweedfs/go/util"
+)
+
+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"`
+ Offset int64 `json:"offset"`
+ Size int64 `json:"size"`
+}
+
+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"`
+}
+
+// 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 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
+ }
+ sort.Sort(cm.Chunks)
+ return &cm, nil
+}
+
+func (cm *ChunkManifest) Marshal() ([]byte, error) {
+ return json.Marshal(cm)
+}
+
+func (cm *ChunkManifest) DeleteChunks(master string) error {
+ deleteError := 0
+ for _, ci := range cm.Chunks {
+ if e := DeleteFile(master, ci.Fid, ""); e != nil {
+ deleteError++
+ glog.V(0).Infof("Delete %s error: %s, master: %s", ci.Fid, e.Error(), master)
+ }
+ }
+ if deleteError > 0 {
+ return errors.New("Not all chunks deleted.")
+ }
+ 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 {
+ return written, err
+ }
+ if offset > 0 {
+ req.Header.Set("Range", fmt.Sprintf("bytes=%d-", offset))
+ }
+
+ resp, err := util.Do(req)
+ if err != nil {
+ return written, err
+ }
+ defer resp.Body.Close()
+
+ 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 chunk needle error: [%d] %s", resp.StatusCode, fileUrl)
+
+ }
+ return io.Copy(w, resp.Body)
+}
+
+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 cm.Chunks {
+ if cf.pos >= ci.Offset && cf.pos < ci.Offset+ci.Size {
+ chunkIndex = i
+ chunkStartOffset = cf.pos - ci.Offset
+ break
+ }
+ }
+ if chunkIndex < 0 {
+ return n, ErrInvalidRange
+ }
+ 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 n, lookupError
+ }
+ if wn, e := readChunkNeedle(fileUrl, w, chunkStartOffset); e != nil {
+ return n, e
+ } else {
+ n += wn
+ cf.pos += wn
+ }
+
+ chunkStartOffset = 0
+ }
+ return n, nil
+}
+
+func (cf *ChunkedFileReader) ReadAt(p []byte, off int64) (n int, err error) {
+ cf.Seek(off, 0)
+ return cf.Read(p)
+}
+
+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/operation/compress.go b/go/operation/compress.go
new file mode 100644
index 000000000..b1105ba4b
--- /dev/null
+++ b/go/operation/compress.go
@@ -0,0 +1,59 @@
+package operation
+
+import (
+ "bytes"
+ "compress/flate"
+ "compress/gzip"
+ "io/ioutil"
+ "strings"
+
+ "github.com/chrislusf/seaweedfs/go/glog"
+)
+
+/*
+* Default more not to gzip since gzip can be done on client side.
+ */
+func IsGzippable(ext, mtype string) bool {
+ if strings.HasPrefix(mtype, "text/") {
+ return true
+ }
+ switch ext {
+ case ".zip", ".rar", ".gz", ".bz2", ".xz":
+ return false
+ case ".pdf", ".txt", ".html", ".htm", ".css", ".js", ".json":
+ return true
+ }
+ if strings.HasPrefix(mtype, "application/") {
+ if strings.HasSuffix(mtype, "xml") {
+ return true
+ }
+ if strings.HasSuffix(mtype, "script") {
+ return true
+ }
+ }
+ return false
+}
+
+func GzipData(input []byte) ([]byte, error) {
+ buf := new(bytes.Buffer)
+ w, _ := gzip.NewWriterLevel(buf, flate.BestCompression)
+ if _, err := w.Write(input); err != nil {
+ glog.V(2).Infoln("error compressing data:", err)
+ return nil, err
+ }
+ if err := w.Close(); err != nil {
+ glog.V(2).Infoln("error closing compressed data:", err)
+ return nil, err
+ }
+ return buf.Bytes(), nil
+}
+func UnGzipData(input []byte) ([]byte, error) {
+ buf := bytes.NewBuffer(input)
+ r, _ := gzip.NewReader(buf)
+ defer r.Close()
+ output, err := ioutil.ReadAll(r)
+ if err != nil {
+ glog.V(2).Infoln("error uncompressing data:", err)
+ }
+ return output, err
+}
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/operation/submit.go b/go/operation/submit.go
index fec5d3801..d996d63f0 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"
)
@@ -114,25 +116,44 @@ 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
- var fids []string
+ cm := ChunkManifest{
+ Name: baseName,
+ 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),
+ baseName+"-"+strconv.FormatInt(i+1, 10),
io.LimitReader(fi.Reader, chunkSize),
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)
+ ret, e := Upload(fileUrl, baseName, fi.Reader, fi.IsGzipped, fi.MimeType, jwt)
if e != nil {
return 0, e
}
@@ -158,10 +179,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.Marshal()
+ 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, "application/json", jwt)
return e
}
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-21 23:36:38 +0000