From c5cf9bd29046877ed6b173e5d2723f865d06aa66 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 13 May 2018 23:56:16 -0700 Subject: [PATCH] properly working filer --- weed/command/filer_copy.go | 12 +-- weed/filer2/filer.go | 15 ++- weed/filer2/filer_structure.go | 9 ++ weed/filer2/memdb/memdb_store_test.go | 8 ++ weed/operation/filer/register.go | 4 +- weed/server/filer_grpc_server.go | 49 +++++---- weed/server/filer_server.go | 38 +++---- weed/server/filer_server_handlers_admin.go | 42 ++++---- weed/server/filer_server_handlers_read.go | 69 ++++++------ weed/server/filer_server_handlers_write.go | 118 ++++++++++----------- weed/server/filer_ui/templates.go | 23 ++-- 11 files changed, 202 insertions(+), 185 deletions(-) diff --git a/weed/command/filer_copy.go b/weed/command/filer_copy.go index da7fb43bb..2b286d3d8 100644 --- a/weed/command/filer_copy.go +++ b/weed/command/filer_copy.go @@ -91,14 +91,14 @@ func runCopy(cmd *Command, args []string) bool { func doEachCopy(fileOrDir string, host string, path string) bool { f, err := os.Open(fileOrDir) if err != nil { - fmt.Printf("Failed to open file %s: %v", fileOrDir, err) + fmt.Printf("Failed to open file %s: %v\n", fileOrDir, err) return false } defer f.Close() fi, err := f.Stat() if err != nil { - fmt.Printf("Failed to get stat for file %s: %v", fileOrDir, err) + fmt.Printf("Failed to get stat for file %s: %v\n", fileOrDir, err) return false } @@ -122,22 +122,22 @@ func doEachCopy(fileOrDir string, host string, path string) bool { parts, err := operation.NewFileParts([]string{fileOrDir}) if err != nil { - fmt.Printf("Failed to read file %s: %v", fileOrDir, err) + fmt.Printf("Failed to read file %s: %v\n", fileOrDir, err) } results, err := operation.SubmitFiles(*copy.master, parts, *copy.replication, *copy.collection, "", *copy.ttl, *copy.maxMB, copy.secret) if err != nil { - fmt.Printf("Failed to submit file %s: %v", fileOrDir, err) + fmt.Printf("Failed to submit file %s: %v\n", fileOrDir, err) } if strings.HasSuffix(path, "/") { path = path + fi.Name() } - if err = filer_operation.RegisterFile(host, path, results[0].Fid, copy.secret); err != nil { - fmt.Printf("Failed to register file %s on %s: %v", fileOrDir, host, err) + if err = filer_operation.RegisterFile(host, path, results[0].Fid, parts[0].FileSize, copy.secret); err != nil { + fmt.Printf("Failed to register file %s on %s: %v\n", fileOrDir, host, err) return false } diff --git a/weed/filer2/filer.go b/weed/filer2/filer.go index ad7a3d906..5f76d6fb0 100644 --- a/weed/filer2/filer.go +++ b/weed/filer2/filer.go @@ -114,11 +114,24 @@ func (f *Filer) FindEntry(p FullPath) (found bool, entry *Entry, err error) { } func (f *Filer) DeleteEntry(p FullPath) (fileEntry *Entry, err error) { + found, entry, err := f.FindEntry(p) + if err != nil || !found { + return nil, err + } + if entry.IsDirectory() { + entries, err := f.ListDirectoryEntries(p, "", false, 1) + if err != nil { + return nil, fmt.Errorf("list folder %s: %v", p, err) + } + if len(entries) > 0 { + return nil, fmt.Errorf("folder %s is not empty", p) + } + } return f.store.DeleteEntry(p) } func (f *Filer) ListDirectoryEntries(p FullPath, startFileName string, inclusive bool, limit int) ([]*Entry, error) { - if strings.HasSuffix(string(p), "/") { + if strings.HasSuffix(string(p), "/") && len(p) > 1 { p = p[0:len(p)-1] } return f.store.ListDirectoryEntries(p, startFileName, inclusive, limit) diff --git a/weed/filer2/filer_structure.go b/weed/filer2/filer_structure.go index c31d878cf..1cff96253 100644 --- a/weed/filer2/filer_structure.go +++ b/weed/filer2/filer_structure.go @@ -21,6 +21,11 @@ func (fp FullPath) DirAndName() (string, string) { return dir[:len(dir)-1], name } +func (fp FullPath) Name() (string) { + _, name := filepath.Split(string(fp)) + return name +} + type Attr struct { Mtime time.Time // time of last modification Crtime time.Time // time of creation (OS X only) @@ -29,6 +34,10 @@ type Attr struct { Gid uint32 // group gid } +func (attr Attr) IsDirectory() (bool) { + return attr.Mode & os.ModeDir > 0 +} + type Entry struct { FullPath diff --git a/weed/filer2/memdb/memdb_store_test.go b/weed/filer2/memdb/memdb_store_test.go index 3fefbf5cd..ab93a7665 100644 --- a/weed/filer2/memdb/memdb_store_test.go +++ b/weed/filer2/memdb/memdb_store_test.go @@ -108,6 +108,14 @@ func TestCreateFileAndList(t *testing.T) { return } + // checking root directory + entries, _ = filer.ListDirectoryEntries(filer2.FullPath("/"), "", false, 100) + if len(entries) != 1 { + t.Errorf("list entries count: %v", len(entries)) + return + } + + // add file3 file3Path := filer2.FullPath("/home/chris/this/is/file3.jpg") entry3 := &filer2.Entry{ FullPath: file3Path, diff --git a/weed/operation/filer/register.go b/weed/operation/filer/register.go index d45fd4f35..94e502165 100644 --- a/weed/operation/filer/register.go +++ b/weed/operation/filer/register.go @@ -6,6 +6,7 @@ import ( "github.com/chrislusf/seaweedfs/weed/security" "github.com/chrislusf/seaweedfs/weed/util" + "strconv" ) type SubmitResult struct { @@ -16,13 +17,14 @@ type SubmitResult struct { Error string `json:"error,omitempty"` } -func RegisterFile(filer string, path string, fileId string, secret security.Secret) error { +func RegisterFile(filer string, path string, fileId string, fileSize int64, secret security.Secret) error { // TODO: jwt need to be used _ = security.GenJwt(secret, fileId) values := make(url.Values) values.Add("path", path) values.Add("fileId", fileId) + values.Add("fileSize", strconv.FormatInt(fileSize, 10)) _, err := util.Post("http://"+filer+"/admin/register", values) if err != nil { return fmt.Errorf("Failed to register path %s on filer %s to file id %s : %v", path, filer, fileId, err) diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index b53c8c419..665ce2d9b 100644 --- a/weed/server/filer_grpc_server.go +++ b/weed/server/filer_grpc_server.go @@ -8,11 +8,14 @@ import ( "github.com/chrislusf/seaweedfs/weed/util" "github.com/chrislusf/seaweedfs/weed/pb/filer_pb" "fmt" + "github.com/chrislusf/seaweedfs/weed/filer2" + "path/filepath" + "github.com/chrislusf/seaweedfs/weed/glog" ) func (fs *FilerServer) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { - found, fileId, err := fs.filer.LookupDirectoryEntry(req.Directory, req.Name) + found, entry, err := fs.filer.FindEntry(filer2.FullPath(filepath.Join(req.Directory, req.Name))) if err != nil { return nil, err } @@ -20,10 +23,15 @@ func (fs *FilerServer) LookupDirectoryEntry(ctx context.Context, req *filer_pb.L return nil, fmt.Errorf("%s not found under %s", req.Name, req.Directory) } + var fileId string + if !entry.IsDirectory() && len(entry.Chunks) > 0 { + fileId = string(entry.Chunks[0].Fid) + } + return &filer_pb.LookupDirectoryEntryResponse{ Entry: &filer_pb.Entry{ Name: req.Name, - IsDirectory: fileId == "", + IsDirectory: entry.IsDirectory(), FileId: fileId, }, }, nil @@ -31,27 +39,21 @@ func (fs *FilerServer) LookupDirectoryEntry(ctx context.Context, req *filer_pb.L func (fs *FilerServer) ListEntries(ctx context.Context, req *filer_pb.ListEntriesRequest) (*filer_pb.ListEntriesResponse, error) { - directoryNames, err := fs.filer.ListDirectories(req.Directory) - if err != nil { - return nil, err - } - files, err := fs.filer.ListFiles(req.Directory, "", 1000) + entries, err := fs.filer.ListDirectoryEntries(filer2.FullPath(req.Directory), "", false, 1000) if err != nil { return nil, err } resp := &filer_pb.ListEntriesResponse{} - for _, dir := range directoryNames { - resp.Entries = append(resp.Entries, &filer_pb.Entry{ - Name: string(dir), - IsDirectory: true, - }) - } - for _, fileEntry := range files { + for _, entry := range entries { + var fileId string + if !entry.IsDirectory() && len(entry.Chunks) > 0 { + fileId = string(entry.Chunks[0].Fid) + } resp.Entries = append(resp.Entries, &filer_pb.Entry{ - Name: fileEntry.Name, - IsDirectory: false, - FileId: string(fileEntry.Id), + Name: entry.Name(), + IsDirectory: entry.IsDirectory(), + FileId: fileId, }) } @@ -97,12 +99,13 @@ func (fs *FilerServer) GetFileContent(ctx context.Context, req *filer_pb.GetFile } func (fs *FilerServer) DeleteEntry(ctx context.Context, req *filer_pb.DeleteEntryRequest) (resp *filer_pb.DeleteEntryResponse, err error) { - if req.IsDirectory { - err = fs.filer.DeleteDirectory(req.Directory+req.Name, false) - } else { - fid, err := fs.filer.DeleteFile(req.Directory + req.Name) - if err == nil && fid != "" { - err = operation.DeleteFile(fs.getMasterNode(), fid, fs.jwt(fid)) + entry, err := fs.filer.DeleteEntry(filer2.FullPath(filepath.Join(req.Directory, req.Name))) + if err == nil { + for _, chunk := range entry.Chunks { + fid := string(chunk.Fid) + if err = operation.DeleteFile(fs.getMasterNode(), fid, fs.jwt(fid)); err != nil { + glog.V(0).Infof("deleting file %s: %v", fid, err) + } } } return nil, err diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go index feaea78ae..8eeced9cb 100644 --- a/weed/server/filer_server.go +++ b/weed/server/filer_server.go @@ -9,21 +9,18 @@ import ( "sync" "time" - "github.com/chrislusf/seaweedfs/weed/filer" - "github.com/chrislusf/seaweedfs/weed/filer/cassandra_store" - "github.com/chrislusf/seaweedfs/weed/filer/embedded_filer" - "github.com/chrislusf/seaweedfs/weed/filer/flat_namespace" "github.com/chrislusf/seaweedfs/weed/filer/mysql_store" "github.com/chrislusf/seaweedfs/weed/filer/postgres_store" - "github.com/chrislusf/seaweedfs/weed/filer/redis_store" "github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/security" "github.com/chrislusf/seaweedfs/weed/storage" "github.com/chrislusf/seaweedfs/weed/util" + "github.com/chrislusf/seaweedfs/weed/filer2" + "github.com/chrislusf/seaweedfs/weed/filer2/memdb" ) type filerConf struct { - MysqlConf []mysql_store.MySqlConf `json:"mysql"` + MysqlConf []mysql_store.MySqlConf `json:"mysql"` mysql_store.ShardingConf PostgresConf *postgres_store.PostgresConf `json:"postgres"` } @@ -52,7 +49,7 @@ type FilerServer struct { redirectOnRead bool disableDirListing bool secret security.Secret - filer filer.Filer + filer *filer2.Filer maxMB int masterNodes *storage.MasterNodes } @@ -86,28 +83,31 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, ip string, port int, } if setting.MysqlConf != nil && len(setting.MysqlConf) != 0 { - mysql_store := mysql_store.NewMysqlStore(setting.MysqlConf, setting.IsSharding, setting.ShardCount) - fs.filer = flat_namespace.NewFlatNamespaceFiler(master, mysql_store) + // mysql_store := mysql_store.NewMysqlStore(setting.MysqlConf, setting.IsSharding, setting.ShardCount) + // fs.filer = flat_namespace.NewFlatNamespaceFiler(master, mysql_store) } else if setting.PostgresConf != nil { - fs.filer = postgres_store.NewPostgresStore(master, *setting.PostgresConf) + // fs.filer = postgres_store.NewPostgresStore(master, *setting.PostgresConf) } else if cassandra_server != "" { - cassandra_store, err := cassandra_store.NewCassandraStore(cassandra_keyspace, cassandra_server) - if err != nil { - glog.Fatalf("Can not connect to cassandra server %s with keyspace %s: %v", cassandra_server, cassandra_keyspace, err) - } - fs.filer = flat_namespace.NewFlatNamespaceFiler(master, cassandra_store) + // cassandra_store, err := cassandra_store.NewCassandraStore(cassandra_keyspace, cassandra_server) + // if err != nil { + // glog.Fatalf("Can not connect to cassandra server %s with keyspace %s: %v", cassandra_server, cassandra_keyspace, err) + // } + // fs.filer = flat_namespace.NewFlatNamespaceFiler(master, cassandra_store) } else if redis_server != "" { - redis_store := redis_store.NewRedisStore(redis_server, redis_password, redis_database) - fs.filer = flat_namespace.NewFlatNamespaceFiler(master, redis_store) + // redis_store := redis_store.NewRedisStore(redis_server, redis_password, redis_database) + // fs.filer = flat_namespace.NewFlatNamespaceFiler(master, redis_store) } else { + /* if fs.filer, err = embedded_filer.NewFilerEmbedded(master, dir); err != nil { glog.Fatalf("Can not start filer in dir %s : %v", dir, err) return } - - defaultMux.HandleFunc("/admin/mv", fs.moveHandler) + */ } + fs.filer = filer2.NewFiler(master) + fs.filer.SetStore(memdb.NewMemDbStore()) + defaultMux.HandleFunc("/admin/register", fs.registerHandler) defaultMux.HandleFunc("/", fs.filerHandler) if defaultMux != readonlyMux { diff --git a/weed/server/filer_server_handlers_admin.go b/weed/server/filer_server_handlers_admin.go index aa7c09986..5c1b23255 100644 --- a/weed/server/filer_server_handlers_admin.go +++ b/weed/server/filer_server_handlers_admin.go @@ -4,34 +4,30 @@ import ( "net/http" "github.com/chrislusf/seaweedfs/weed/glog" + "github.com/chrislusf/seaweedfs/weed/filer2" + "strconv" ) -/* -Move a folder or a file, with 4 Use cases: - mv fromDir toNewDir - mv fromDir toOldDir - mv fromFile toDir - mv fromFile toFile - -Wildcard is not supported. - -*/ -func (fs *FilerServer) moveHandler(w http.ResponseWriter, r *http.Request) { - from := r.FormValue("from") - to := r.FormValue("to") - err := fs.filer.Move(from, to) - if err != nil { - glog.V(4).Infoln("moving", from, "->", to, err.Error()) - writeJsonError(w, r, http.StatusInternalServerError, err) - } else { - w.WriteHeader(http.StatusOK) - } -} - func (fs *FilerServer) registerHandler(w http.ResponseWriter, r *http.Request) { path := r.FormValue("path") fileId := r.FormValue("fileId") - err := fs.filer.CreateFile(path, fileId) + fileSize, err := strconv.ParseUint(r.FormValue("fileSize"), 10, 64) + if err != nil { + glog.V(4).Infof("register %s to %s parse fileSize %s: %v", fileId, path, r.FormValue("fileSize"), err) + writeJsonError(w, r, http.StatusInternalServerError, err) + return + } + entry := &filer2.Entry{ + FullPath: filer2.FullPath(path), + Attr: filer2.Attr{ + Mode: 0660, + }, + Chunks: []filer2.FileChunk{{ + Fid: filer2.FileId(fileId), + Size: fileSize, + }}, + } + err = fs.filer.CreateEntry(entry) if err != nil { glog.V(4).Infof("register %s to %s error: %v", fileId, path, err) writeJsonError(w, r, http.StatusInternalServerError, err) diff --git a/weed/server/filer_server_handlers_read.go b/weed/server/filer_server_handlers_read.go index e95c7fcd0..93775f839 100644 --- a/weed/server/filer_server_handlers_read.go +++ b/weed/server/filer_server_handlers_read.go @@ -7,12 +7,11 @@ import ( "strconv" "strings" - "github.com/chrislusf/seaweedfs/weed/filer" "github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/operation" ui "github.com/chrislusf/seaweedfs/weed/server/filer_ui" "github.com/chrislusf/seaweedfs/weed/util" - "github.com/syndtr/goleveldb/leveldb" + "github.com/chrislusf/seaweedfs/weed/filer2" ) // listDirectoryHandler lists directories and folers under a directory @@ -20,56 +19,40 @@ import ( // sub directories are listed on the first page, when "lastFileName" // is empty. func (fs *FilerServer) listDirectoryHandler(w http.ResponseWriter, r *http.Request) { - if !strings.HasSuffix(r.URL.Path, "/") { - return + path := r.URL.Path + if strings.HasSuffix(path, "/") && len(path) > 1 { + path = path[:len(path)-1] } + limit, limit_err := strconv.Atoi(r.FormValue("limit")) if limit_err != nil { limit = 100 } lastFileName := r.FormValue("lastFileName") - files, err := fs.filer.ListFiles(r.URL.Path, lastFileName, limit) - if err == leveldb.ErrNotFound { - glog.V(0).Infof("Error %s", err) - w.WriteHeader(http.StatusNotFound) - return - } + entries, err := fs.filer.ListDirectoryEntries(filer2.FullPath(path), lastFileName, false, limit) - directories, err2 := fs.filer.ListDirectories(r.URL.Path) - if err2 == leveldb.ErrNotFound { - glog.V(0).Infof("Error %s", err) + if err != nil { + glog.V(0).Infof("listDirectory %s %s $d: %s", path, lastFileName, limit, err) w.WriteHeader(http.StatusNotFound) return } - shouldDisplayLoadMore := len(files) > 0 - - lastFileName = "" - if len(files) > 0 { - lastFileName = files[len(files)-1].Name - - files2, err3 := fs.filer.ListFiles(r.URL.Path, lastFileName, limit) - if err3 == leveldb.ErrNotFound { - glog.V(0).Infof("Error %s", err) - w.WriteHeader(http.StatusNotFound) - return - } - shouldDisplayLoadMore = len(files2) > 0 + shouldDisplayLoadMore := len(entries) == limit + if path == "/" { + path = "" } args := struct { Path string - Files interface{} - Directories interface{} + Entries interface{} Limit int LastFileName string ShouldDisplayLoadMore bool }{ - r.URL.Path, - files, - directories, + path, + entries, limit, lastFileName, shouldDisplayLoadMore, @@ -83,7 +66,19 @@ func (fs *FilerServer) listDirectoryHandler(w http.ResponseWriter, r *http.Reque } func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request, isGetMethod bool) { - if strings.HasSuffix(r.URL.Path, "/") { + path := r.URL.Path + if strings.HasSuffix(path, "/") && len(path) > 1 { + path = path[:len(path)-1] + } + + found, entry, err := fs.filer.FindEntry(filer2.FullPath(path)) + if !found || err != nil { + glog.V(3).Infof("Not found %s: %v", path, err) + w.WriteHeader(http.StatusNotFound) + return + } + + if entry.IsDirectory() { if fs.disableDirListing { w.WriteHeader(http.StatusMethodNotAllowed) return @@ -92,13 +87,15 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request, return } - fileId, err := fs.filer.FindFile(r.URL.Path) - if err == filer.ErrNotFound { - glog.V(3).Infoln("Not found in db", r.URL.Path) - w.WriteHeader(http.StatusNotFound) + if len(entry.Chunks) == 0 { + glog.V(3).Infof("Empty %s: %v", path) + w.WriteHeader(http.StatusNoContent) return } + // FIXME pick the right fid + fileId := string(entry.Chunks[0].Fid) + urlLocation, err := operation.LookupFileId(fs.getMasterNode(), fileId) if err != nil { glog.V(1).Infoln("operation LookupFileId %s failed, err is %s", fileId, err.Error()) diff --git a/weed/server/filer_server_handlers_write.go b/weed/server/filer_server_handlers_write.go index 07452cd77..1a4b62235 100644 --- a/weed/server/filer_server_handlers_write.go +++ b/weed/server/filer_server_handlers_write.go @@ -22,6 +22,7 @@ import ( "github.com/chrislusf/seaweedfs/weed/operation" "github.com/chrislusf/seaweedfs/weed/storage" "github.com/chrislusf/seaweedfs/weed/util" + "github.com/chrislusf/seaweedfs/weed/filer2" ) type FilerPostResult struct { @@ -73,16 +74,19 @@ func makeFormData(filename, mimeType string, content io.Reader) (formData io.Rea } func (fs *FilerServer) queryFileInfoByPath(w http.ResponseWriter, r *http.Request, path string) (fileId, urlLocation string, err error) { - if fileId, err = fs.filer.FindFile(path); err != nil && err != filer.ErrNotFound { + var found bool + var entry *filer2.Entry + if found, entry, err = fs.filer.FindEntry(filer2.FullPath(path)); err != nil { glog.V(0).Infoln("failing to find path in filer store", path, err.Error()) writeJsonError(w, r, http.StatusInternalServerError, err) - } else if fileId != "" && err == nil { + } else if found { + fileId = string(entry.Chunks[0].Fid) urlLocation, err = operation.LookupFileId(fs.getMasterNode(), fileId) if err != nil { glog.V(1).Infoln("operation LookupFileId %s failed, err is %s", fileId, err.Error()) w.WriteHeader(http.StatusNotFound) } - } else if fileId == "" && err == filer.ErrNotFound { + } else { w.WriteHeader(http.StatusNotFound) } return @@ -313,7 +317,8 @@ func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request) { // also delete the old fid unless PUT operation if r.Method != "PUT" { - if oldFid, err := fs.filer.FindFile(path); err == nil { + if found, entry, err := fs.filer.FindEntry(filer2.FullPath(path)); err == nil && found { + oldFid := string(entry.Chunks[0].Fid) operation.DeleteFile(fs.getMasterNode(), oldFid, fs.jwt(oldFid)) } else if err != nil && err != filer.ErrNotFound { glog.V(0).Infof("error %v occur when finding %s in filer store", err, path) @@ -321,7 +326,17 @@ func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request) { } glog.V(4).Infoln("saving", path, "=>", fileId) - if db_err := fs.filer.CreateFile(path, fileId); db_err != nil { + entry := &filer2.Entry{ + FullPath: filer2.FullPath(path), + Attr: filer2.Attr{ + Mode: 0660, + }, + Chunks: []filer2.FileChunk{{ + Fid: filer2.FileId(fileId), + Size: uint64(r.ContentLength), + }}, + } + if db_err := fs.filer.CreateEntry(entry); db_err != nil { operation.DeleteFile(fs.getMasterNode(), fileId, fs.jwt(fileId)) //clean up glog.V(0).Infof("failing to write %s to filer server : %v", path, db_err) writeJsonError(w, r, http.StatusInternalServerError, db_err) @@ -400,13 +415,7 @@ func (fs *FilerServer) doAutoChunk(w http.ResponseWriter, r *http.Request, conte fileName = path.Base(fileName) } - chunks := (int64(contentLength) / int64(chunkSize)) + 1 - cm := operation.ChunkManifest{ - Name: fileName, - Size: 0, // don't know yet - Mime: "application/octet-stream", - Chunks: make([]*operation.ChunkInfo, 0, chunks), - } + var fileChunks []filer2.FileChunk totalBytesRead := int64(0) tmpBufferSize := int32(1024 * 1024) @@ -438,18 +447,18 @@ func (fs *FilerServer) doAutoChunk(w http.ResponseWriter, r *http.Request, conte } // upload the chunk to the volume server - chunkName := fileName + "_chunk_" + strconv.FormatInt(int64(cm.Chunks.Len()+1), 10) + chunkName := fileName + "_chunk_" + strconv.FormatInt(int64(len(fileChunks)+1), 10) uploadErr := fs.doUpload(urlLocation, w, r, chunkBuf[0:chunkBufOffset], chunkName, "application/octet-stream", fileId) if uploadErr != nil { return nil, uploadErr } // Save to chunk manifest structure - cm.Chunks = append(cm.Chunks, - &operation.ChunkInfo{ + fileChunks = append(fileChunks, + filer2.FileChunk{ + Fid: filer2.FileId(fileId), Offset: chunkOffset, - Size: int64(chunkBufOffset), - Fid: fileId, + Size: uint64(chunkBufOffset), }, ) @@ -469,47 +478,30 @@ func (fs *FilerServer) doAutoChunk(w http.ResponseWriter, r *http.Request, conte } } - cm.Size = totalBytesRead - manifestBuf, marshalErr := cm.Marshal() - if marshalErr != nil { - return nil, marshalErr - } - - manifestStr := string(manifestBuf) - glog.V(4).Infoln("Generated chunk manifest: ", manifestStr) - - manifestFileId, manifestUrlLocation, manifestAssignmentErr := fs.assignNewFileInfo(w, r, replication, collection) - if manifestAssignmentErr != nil { - return nil, manifestAssignmentErr - } - glog.V(4).Infoln("Manifest uploaded to:", manifestUrlLocation, "Fid:", manifestFileId) - filerResult.Fid = manifestFileId - - u, _ := url.Parse(manifestUrlLocation) - q := u.Query() - q.Set("cm", "true") - u.RawQuery = q.Encode() - - manifestUploadErr := fs.doUpload(u.String(), w, r, manifestBuf, fileName+"_manifest", "application/json", manifestFileId) - if manifestUploadErr != nil { - return nil, manifestUploadErr - } - path := r.URL.Path // also delete the old fid unless PUT operation if r.Method != "PUT" { - if oldFid, err := fs.filer.FindFile(path); err == nil { - operation.DeleteFile(fs.getMasterNode(), oldFid, fs.jwt(oldFid)) - } else if err != nil && err != filer.ErrNotFound { + if found, entry, err := fs.filer.FindEntry(filer2.FullPath(path)); found && err == nil { + for _, chunk := range entry.Chunks { + oldFid := string(chunk.Fid) + operation.DeleteFile(fs.getMasterNode(), oldFid, fs.jwt(oldFid)) + } + } else if err != nil { glog.V(0).Infof("error %v occur when finding %s in filer store", err, path) } } - glog.V(4).Infoln("saving", path, "=>", manifestFileId) - if db_err := fs.filer.CreateFile(path, manifestFileId); db_err != nil { + glog.V(4).Infoln("saving", path) + entry := &filer2.Entry{ + FullPath: filer2.FullPath(path), + Attr: filer2.Attr{ + Mode: 0660, + }, + Chunks: fileChunks, + } + if db_err := fs.filer.CreateEntry(entry); db_err != nil { replyerr = db_err filerResult.Error = db_err.Error() - operation.DeleteFile(fs.getMasterNode(), manifestFileId, fs.jwt(manifestFileId)) //clean up glog.V(0).Infof("failing to write %s to filer server : %v", path, db_err) return } @@ -532,23 +524,21 @@ func (fs *FilerServer) doUpload(urlLocation string, w http.ResponseWriter, r *ht } // curl -X DELETE http://localhost:8888/path/to -// curl -X DELETE http://localhost:8888/path/to/?recursive=true func (fs *FilerServer) DeleteHandler(w http.ResponseWriter, r *http.Request) { - var err error - var fid string - if strings.HasSuffix(r.URL.Path, "/") { - isRecursive := r.FormValue("recursive") == "true" - err = fs.filer.DeleteDirectory(r.URL.Path, isRecursive) - } else { - fid, err = fs.filer.DeleteFile(r.URL.Path) - if err == nil && fid != "" { - err = operation.DeleteFile(fs.getMasterNode(), fid, fs.jwt(fid)) - } - } - if err == nil { - writeJsonQuiet(w, r, http.StatusAccepted, map[string]string{"error": ""}) - } else { + + entry, err := fs.filer.DeleteEntry(filer2.FullPath(r.URL.Path)) + if err != nil { glog.V(4).Infoln("deleting", r.URL.Path, ":", err.Error()) writeJsonError(w, r, http.StatusInternalServerError, err) + return + } + + if entry != nil && !entry.IsDirectory() { + for _, chunk := range entry.Chunks { + oldFid := string(chunk.Fid) + operation.DeleteFile(fs.getMasterNode(), oldFid, fs.jwt(oldFid)) + } } + + writeJsonQuiet(w, r, http.StatusAccepted, map[string]string{"error": ""}) } diff --git a/weed/server/filer_ui/templates.go b/weed/server/filer_ui/templates.go index 508a4d9aa..bba60ff1a 100644 --- a/weed/server/filer_ui/templates.go +++ b/weed/server/filer_ui/templates.go @@ -26,22 +26,21 @@ var StatusTpl = template.Must(template.New("status").Parse(`