diff --git a/weed/command/webdav.go b/weed/command/webdav.go index 71933d885..9ab2ce9de 100644 --- a/weed/command/webdav.go +++ b/weed/command/webdav.go @@ -3,6 +3,8 @@ package command import ( "fmt" "net/http" + "os/user" + "strconv" "time" "github.com/chrislusf/seaweedfs/weed/glog" @@ -19,6 +21,7 @@ var ( type WebDavOption struct { filer *string port *int + collection *string tlsPrivateKey *string tlsCertificate *string } @@ -27,6 +30,7 @@ func init() { cmdWebDav.Run = runWebDav // break init cycle webDavStandaloneOptions.filer = cmdWebDav.Flag.String("filer", "localhost:8888", "filer server address") webDavStandaloneOptions.port = cmdWebDav.Flag.Int("port", 7333, "webdav server http listen port") + webDavStandaloneOptions.collection = cmdWebDav.Flag.String("collection", "", "collection to create the files") webDavStandaloneOptions.tlsPrivateKey = cmdWebDav.Flag.String("key.file", "", "path to the TLS private key file") webDavStandaloneOptions.tlsCertificate = cmdWebDav.Flag.String("cert.file", "", "path to the TLS certificate file") } @@ -57,10 +61,24 @@ func (wo *WebDavOption) startWebDav() bool { return false } + // detect current user + uid, gid := uint32(0), uint32(0) + if u, err := user.Current(); err == nil { + if parsedId, pe := strconv.ParseUint(u.Uid, 10, 32); pe == nil { + uid = uint32(parsedId) + } + if parsedId, pe := strconv.ParseUint(u.Gid, 10, 32); pe == nil { + gid = uint32(parsedId) + } + } + ws, webdavServer_err := weed_server.NewWebDavServer(&weed_server.WebDavOption{ Filer: *wo.filer, FilerGrpcAddress: filerGrpcAddress, GrpcDialOption: security.LoadClientTLS(viper.Sub("grpc"), "client"), + Collection: *wo.collection, + Uid: uid, + Gid: gid, }) if webdavServer_err != nil { glog.Fatalf("WebDav Server startup error: %v", webdavServer_err) diff --git a/weed/filer2/filer_client_util.go b/weed/filer2/filer_client_util.go new file mode 100644 index 000000000..71daa9ee1 --- /dev/null +++ b/weed/filer2/filer_client_util.go @@ -0,0 +1,163 @@ +package filer2 + +import ( + "context" + "fmt" + "strings" + "sync" + + "github.com/chrislusf/seaweedfs/weed/glog" + "github.com/chrislusf/seaweedfs/weed/pb/filer_pb" + "github.com/chrislusf/seaweedfs/weed/util" +) + +func VolumeId(fileId string) string { + lastCommaIndex := strings.LastIndex(fileId, ",") + if lastCommaIndex > 0 { + return fileId[:lastCommaIndex] + } + return fileId +} + +type FilerClient interface { + WithFilerClient(ctx context.Context, fn func(filer_pb.SeaweedFilerClient) error) error +} + +func ReadIntoBuffer(ctx context.Context, filerClient FilerClient, fullFilePath string, buff []byte, chunkViews []*ChunkView, baseOffset int64) (totalRead int64, err error) { + var vids []string + for _, chunkView := range chunkViews { + vids = append(vids, VolumeId(chunkView.FileId)) + } + + vid2Locations := make(map[string]*filer_pb.Locations) + + err = filerClient.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { + + glog.V(4).Infof("read fh lookup volume id locations: %v", vids) + resp, err := client.LookupVolume(ctx, &filer_pb.LookupVolumeRequest{ + VolumeIds: vids, + }) + if err != nil { + return err + } + + vid2Locations = resp.LocationsMap + + return nil + }) + + if err != nil { + return 0, fmt.Errorf("failed to lookup volume ids %v: %v", vids, err) + } + + var wg sync.WaitGroup + for _, chunkView := range chunkViews { + wg.Add(1) + go func(chunkView *ChunkView) { + defer wg.Done() + + glog.V(4).Infof("read fh reading chunk: %+v", chunkView) + + locations := vid2Locations[VolumeId(chunkView.FileId)] + if locations == nil || len(locations.Locations) == 0 { + glog.V(0).Infof("failed to locate %s", chunkView.FileId) + err = fmt.Errorf("failed to locate %s", chunkView.FileId) + return + } + + var n int64 + n, err = util.ReadUrl( + fmt.Sprintf("http://%s/%s", locations.Locations[0].Url, chunkView.FileId), + chunkView.Offset, + int(chunkView.Size), + buff[chunkView.LogicOffset-baseOffset:chunkView.LogicOffset-baseOffset+int64(chunkView.Size)], + !chunkView.IsFullChunk) + + if err != nil { + + glog.V(0).Infof("%v read http://%s/%v %v bytes: %v", fullFilePath, locations.Locations[0].Url, chunkView.FileId, n, err) + + err = fmt.Errorf("failed to read http://%s/%s: %v", + locations.Locations[0].Url, chunkView.FileId, err) + return + } + + glog.V(4).Infof("read fh read %d bytes: %+v", n, chunkView) + totalRead += n + + }(chunkView) + } + wg.Wait() + return +} + +func GetEntry(ctx context.Context, filerClient FilerClient, fullFilePath string) (entry *filer_pb.Entry, err error) { + + dir, name := FullPath(fullFilePath).DirAndName() + + err = filerClient.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { + + request := &filer_pb.LookupDirectoryEntryRequest{ + Directory: dir, + Name: name, + } + + glog.V(1).Infof("read %s request: %v", fullFilePath, request) + resp, err := client.LookupDirectoryEntry(ctx, request) + if err != nil { + if err == ErrNotFound { + return nil + } + glog.V(0).Infof("read %s attr %v: %v", fullFilePath, request, err) + return err + } + + if resp.Entry != nil { + entry = resp.Entry + } + + return nil + }) + + return +} + +func ReadDirAllEntries(ctx context.Context, filerClient FilerClient, fullDirPath string, fn func(entry *filer_pb.Entry)) (err error) { + + err = filerClient.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { + + paginationLimit := 1024 + + lastEntryName := "" + + for { + + request := &filer_pb.ListEntriesRequest{ + Directory: fullDirPath, + StartFromFileName: lastEntryName, + Limit: uint32(paginationLimit), + } + + glog.V(4).Infof("read directory: %v", request) + resp, err := client.ListEntries(ctx, request) + if err != nil { + return fmt.Errorf("list %s: %v", fullDirPath, err) + } + + for _, entry := range resp.Entries { + fn(entry) + lastEntryName = entry.Name + } + + if len(resp.Entries) < paginationLimit { + break + } + + } + + return nil + + }) + + return +} diff --git a/weed/filesys/dir.go b/weed/filesys/dir.go index 767212103..7b6cf2000 100644 --- a/weed/filesys/dir.go +++ b/weed/filesys/dir.go @@ -4,7 +4,6 @@ import ( "context" "os" "path" - "path/filepath" "time" "github.com/chrislusf/seaweedfs/weed/filer2" @@ -54,39 +53,12 @@ func (dir *Dir) Attr(ctx context.Context, attr *fuse.Attr) error { return nil } - parent, name := filepath.Split(dir.Path) - - err := dir.wfs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { - - request := &filer_pb.LookupDirectoryEntryRequest{ - Directory: parent, - Name: name, - } - - glog.V(1).Infof("read dir %s request: %v", dir.Path, request) - resp, err := client.LookupDirectoryEntry(ctx, request) - if err != nil { - if err == filer2.ErrNotFound { - return nil - } - glog.V(0).Infof("read dir %s attr %v: %v", dir.Path, request, err) - return err - } - - if resp.Entry != nil { - dir.attributes = resp.Entry.Attributes - } - glog.V(2).Infof("read dir %s attr: %v", dir.Path, dir.attributes) - - // dir.wfs.listDirectoryEntriesCache.Set(dir.Path, resp.Entry, dir.wfs.option.EntryCacheTtl) - - return nil - }) - + entry, err := filer2.GetEntry(ctx, dir.wfs, dir.Path) if err != nil { glog.V(2).Infof("read dir %s attr: %v, error: %v", dir.Path, dir.attributes, err) return err } + dir.attributes = entry.Attributes glog.V(2).Infof("dir %s: %v perm: %v", dir.Path, dir.attributes, os.FileMode(dir.attributes.FileMode)) @@ -133,7 +105,7 @@ func (dir *Dir) Create(ctx context.Context, req *fuse.CreateRequest, glog.V(1).Infof("create: %v", request) if request.Entry.IsDirectory { - if err := dir.wfs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { + if err := dir.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { if _, err := client.CreateEntry(ctx, request); err != nil { glog.V(0).Infof("create %s/%s: %v", dir.Path, req.Name, err) return fuse.EIO @@ -156,7 +128,7 @@ func (dir *Dir) Create(ctx context.Context, req *fuse.CreateRequest, func (dir *Dir) Mkdir(ctx context.Context, req *fuse.MkdirRequest) (fs.Node, error) { - err := dir.wfs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { + err := dir.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { request := &filer_pb.CreateEntryRequest{ Directory: dir.Path, @@ -193,33 +165,18 @@ func (dir *Dir) Mkdir(ctx context.Context, req *fuse.MkdirRequest) (fs.Node, err func (dir *Dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.LookupResponse) (node fs.Node, err error) { var entry *filer_pb.Entry + fullFilePath := path.Join(dir.Path, req.Name) - item := dir.wfs.listDirectoryEntriesCache.Get(path.Join(dir.Path, req.Name)) + item := dir.wfs.listDirectoryEntriesCache.Get(fullFilePath) if item != nil && !item.Expired() { entry = item.Value().(*filer_pb.Entry) } if entry == nil { - err = dir.wfs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { - - request := &filer_pb.LookupDirectoryEntryRequest{ - Directory: dir.Path, - Name: req.Name, - } - - glog.V(4).Infof("lookup directory entry: %v", request) - resp, err := client.LookupDirectoryEntry(ctx, request) - if err != nil { - // glog.V(0).Infof("lookup %s/%s: %v", dir.Path, name, err) - return fuse.ENOENT - } - - entry = resp.Entry - - // dir.wfs.listDirectoryEntriesCache.Set(path.Join(dir.Path, entry.Name), entry, dir.wfs.option.EntryCacheTtl) - - return nil - }) + entry, err = filer2.GetEntry(ctx, dir.wfs, fullFilePath) + if err != nil { + return nil, err + } } if entry != nil { @@ -244,7 +201,7 @@ func (dir *Dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse. func (dir *Dir) ReadDirAll(ctx context.Context) (ret []fuse.Dirent, err error) { - err = dir.wfs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { + err = dir.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { paginationLimit := 1024 remaining := dir.wfs.option.DirListingLimit @@ -306,33 +263,14 @@ func (dir *Dir) Remove(ctx context.Context, req *fuse.RemoveRequest) error { func (dir *Dir) removeOneFile(ctx context.Context, req *fuse.RemoveRequest) error { - var entry *filer_pb.Entry - err := dir.wfs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { - - request := &filer_pb.LookupDirectoryEntryRequest{ - Directory: dir.Path, - Name: req.Name, - } - - glog.V(4).Infof("lookup to-be-removed entry: %v", request) - resp, err := client.LookupDirectoryEntry(ctx, request) - if err != nil { - // glog.V(0).Infof("lookup %s/%s: %v", dir.Path, name, err) - return fuse.ENOENT - } - - entry = resp.Entry - - return nil - }) - + entry, err := filer2.GetEntry(ctx, dir.wfs, path.Join(dir.Path, req.Name)) if err != nil { return err } dir.wfs.deleteFileChunks(ctx, entry.Chunks) - return dir.wfs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { + return dir.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { request := &filer_pb.DeleteEntryRequest{ Directory: dir.Path, @@ -356,7 +294,7 @@ func (dir *Dir) removeOneFile(ctx context.Context, req *fuse.RemoveRequest) erro func (dir *Dir) removeFolder(ctx context.Context, req *fuse.RemoveRequest) error { - return dir.wfs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { + return dir.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { request := &filer_pb.DeleteEntryRequest{ Directory: dir.Path, @@ -402,7 +340,7 @@ func (dir *Dir) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fus } parentDir, name := filer2.FullPath(dir.Path).DirAndName() - return dir.wfs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { + return dir.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { request := &filer_pb.UpdateEntryRequest{ Directory: parentDir, diff --git a/weed/filesys/dir_link.go b/weed/filesys/dir_link.go index 4f631bc88..92cf04d58 100644 --- a/weed/filesys/dir_link.go +++ b/weed/filesys/dir_link.go @@ -35,7 +35,7 @@ func (dir *Dir) Symlink(ctx context.Context, req *fuse.SymlinkRequest) (fs.Node, }, } - err := dir.wfs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { + err := dir.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { if _, err := client.CreateEntry(ctx, request); err != nil { glog.V(0).Infof("symlink %s/%s: %v", dir.Path, req.NewName, err) return fuse.EIO diff --git a/weed/filesys/dir_rename.go b/weed/filesys/dir_rename.go index 7a415ff82..e72a15758 100644 --- a/weed/filesys/dir_rename.go +++ b/weed/filesys/dir_rename.go @@ -12,7 +12,7 @@ func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDirector newDir := newDirectory.(*Dir) - return dir.wfs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { + return dir.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { request := &filer_pb.AtomicRenameEntryRequest{ OldDirectory: dir.Path, diff --git a/weed/filesys/dirty_page.go b/weed/filesys/dirty_page.go index 0044cfd87..5a7d51a91 100644 --- a/weed/filesys/dirty_page.go +++ b/weed/filesys/dirty_page.go @@ -167,7 +167,7 @@ func (pages *ContinuousDirtyPages) saveToStorage(ctx context.Context, buf []byte var fileId, host string var auth security.EncodedJwt - if err := pages.f.wfs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { + if err := pages.f.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { request := &filer_pb.AssignVolumeRequest{ Count: 1, diff --git a/weed/filesys/file.go b/weed/filesys/file.go index eb4b03f64..3015354a6 100644 --- a/weed/filesys/file.go +++ b/weed/filesys/file.go @@ -109,7 +109,7 @@ func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *f return nil } - return file.wfs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { + return file.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { request := &filer_pb.UpdateEntryRequest{ Directory: file.dir.Path, @@ -144,7 +144,7 @@ func (file *File) maybeLoadAttributes(ctx context.Context) error { file.setEntry(entry) // glog.V(1).Infof("file attr read cached %v attributes", file.Name) } else { - err := file.wfs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { + err := file.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { request := &filer_pb.LookupDirectoryEntryRequest{ Name: file.Name, diff --git a/weed/filesys/filehandle.go b/weed/filesys/filehandle.go index e87e0608e..feb19f525 100644 --- a/weed/filesys/filehandle.go +++ b/weed/filesys/filehandle.go @@ -3,20 +3,18 @@ package filesys import ( "context" "fmt" + "mime" + "path" + "time" + "github.com/chrislusf/seaweedfs/weed/filer2" "github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/operation" "github.com/chrislusf/seaweedfs/weed/pb/filer_pb" - "github.com/chrislusf/seaweedfs/weed/util" "github.com/gabriel-vasile/mimetype" "github.com/seaweedfs/fuse" "github.com/seaweedfs/fuse/fs" "google.golang.org/grpc" - "mime" - "path" - "strings" - "sync" - "time" ) type FileHandle struct { @@ -68,72 +66,7 @@ func (fh *FileHandle) Read(ctx context.Context, req *fuse.ReadRequest, resp *fus chunkViews := filer2.ViewFromVisibleIntervals(fh.f.entryViewCache, req.Offset, req.Size) - var vids []string - for _, chunkView := range chunkViews { - vids = append(vids, volumeId(chunkView.FileId)) - } - - vid2Locations := make(map[string]*filer_pb.Locations) - - err := fh.f.wfs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { - - glog.V(4).Infof("read fh lookup volume id locations: %v", vids) - resp, err := client.LookupVolume(ctx, &filer_pb.LookupVolumeRequest{ - VolumeIds: vids, - }) - if err != nil { - return err - } - - vid2Locations = resp.LocationsMap - - return nil - }) - - if err != nil { - glog.V(4).Infof("%v/%v read fh lookup volume ids: %v", fh.f.dir.Path, fh.f.Name, err) - return fmt.Errorf("failed to lookup volume ids %v: %v", vids, err) - } - - var totalRead int64 - var wg sync.WaitGroup - for _, chunkView := range chunkViews { - wg.Add(1) - go func(chunkView *filer2.ChunkView) { - defer wg.Done() - - glog.V(4).Infof("read fh reading chunk: %+v", chunkView) - - locations := vid2Locations[volumeId(chunkView.FileId)] - if locations == nil || len(locations.Locations) == 0 { - glog.V(0).Infof("failed to locate %s", chunkView.FileId) - err = fmt.Errorf("failed to locate %s", chunkView.FileId) - return - } - - var n int64 - n, err = util.ReadUrl( - fmt.Sprintf("http://%s/%s", locations.Locations[0].Url, chunkView.FileId), - chunkView.Offset, - int(chunkView.Size), - buff[chunkView.LogicOffset-req.Offset:chunkView.LogicOffset-req.Offset+int64(chunkView.Size)], - !chunkView.IsFullChunk) - - if err != nil { - - glog.V(0).Infof("%v/%v read http://%s/%v %v bytes: %v", fh.f.dir.Path, fh.f.Name, locations.Locations[0].Url, chunkView.FileId, n, err) - - err = fmt.Errorf("failed to read http://%s/%s: %v", - locations.Locations[0].Url, chunkView.FileId, err) - return - } - - glog.V(4).Infof("read fh read %d bytes: %+v", n, chunkView) - totalRead += n - - }(chunkView) - } - wg.Wait() + totalRead, err := filer2.ReadIntoBuffer(ctx, fh.f.wfs, fh.f.fullpath(), buff, chunkViews, req.Offset) resp.Data = buff[:totalRead] @@ -205,7 +138,7 @@ func (fh *FileHandle) Flush(ctx context.Context, req *fuse.FlushRequest) error { return nil } - return fh.f.wfs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { + return fh.f.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { if fh.f.entry.Attributes != nil { fh.f.entry.Attributes.Mime = fh.contentType @@ -243,7 +176,7 @@ func deleteFileIds(ctx context.Context, grpcDialOption grpc.DialOption, client f var vids []string for _, fileId := range fileIds { - vids = append(vids, volumeId(fileId)) + vids = append(vids, filer2.VolumeId(fileId)) } lookupFunc := func(vids []string) (map[string]operation.LookupResult, error) { @@ -280,11 +213,3 @@ func deleteFileIds(ctx context.Context, grpcDialOption grpc.DialOption, client f return err } - -func volumeId(fileId string) string { - lastCommaIndex := strings.LastIndex(fileId, ",") - if lastCommaIndex > 0 { - return fileId[:lastCommaIndex] - } - return fileId -} diff --git a/weed/filesys/wfs.go b/weed/filesys/wfs.go index f8be24e5e..b4d1b0608 100644 --- a/weed/filesys/wfs.go +++ b/weed/filesys/wfs.go @@ -73,7 +73,7 @@ func (wfs *WFS) Root() (fs.Node, error) { return &Dir{Path: wfs.option.FilerMountRootPath, wfs: wfs}, nil } -func (wfs *WFS) withFilerClient(ctx context.Context, fn func(filer_pb.SeaweedFilerClient) error) error { +func (wfs *WFS) WithFilerClient(ctx context.Context, fn func(filer_pb.SeaweedFilerClient) error) error { return util.WithCachedGrpcClient(ctx, func(grpcConnection *grpc.ClientConn) error { client := filer_pb.NewSeaweedFilerClient(grpcConnection) @@ -133,7 +133,7 @@ func (wfs *WFS) Statfs(ctx context.Context, req *fuse.StatfsRequest, resp *fuse. if wfs.stats.lastChecked < time.Now().Unix()-20 { - err := wfs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { + err := wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { request := &filer_pb.StatisticsRequest{ Collection: wfs.option.Collection, diff --git a/weed/filesys/wfs_deletion.go b/weed/filesys/wfs_deletion.go index 16f8af594..dd7992816 100644 --- a/weed/filesys/wfs_deletion.go +++ b/weed/filesys/wfs_deletion.go @@ -15,7 +15,7 @@ func (wfs *WFS) deleteFileChunks(ctx context.Context, chunks []*filer_pb.FileChu fileIds = append(fileIds, chunk.FileId) } - wfs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { + wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { deleteFileIds(ctx, wfs.option.GrpcDialOption, client, fileIds) return nil }) diff --git a/weed/server/webdav_server.go b/weed/server/webdav_server.go index b06c055e2..5930e99be 100644 --- a/weed/server/webdav_server.go +++ b/weed/server/webdav_server.go @@ -2,6 +2,7 @@ package weed_server import ( "context" + "fmt" "io" "net/http" "net/url" @@ -10,6 +11,8 @@ import ( "strings" "time" + "github.com/chrislusf/seaweedfs/weed/pb/filer_pb" + "github.com/chrislusf/seaweedfs/weed/util" "golang.org/x/net/webdav" "google.golang.org/grpc" @@ -25,6 +28,9 @@ type WebDavOption struct { DomainName string BucketsPath string GrpcDialOption grpc.DialOption + Collection string + Uid uint32 + Gid uint32 } type WebDavServer struct { @@ -37,7 +43,7 @@ type WebDavServer struct { func NewWebDavServer(option *WebDavOption) (ws *WebDavServer, err error) { - fs, _ := NewWebDavFileSystem() + fs, _ := NewWebDavFileSystem(option) ws = &WebDavServer{ option: option, @@ -78,31 +84,49 @@ func NewWebDavServer(option *WebDavOption) (ws *WebDavServer, err error) { // adapted from https://github.com/mattn/davfs/blob/master/plugin/mysql/mysql.go type WebDavFileSystem struct { + option *WebDavOption + secret security.SigningKey + filer *filer2.Filer + grpcDialOption grpc.DialOption } type FileInfo struct { - name string - size int64 - mode os.FileMode - mod_time time.Time + name string + size int64 + mode os.FileMode + modifiledTime time.Time + isDirectory bool } func (fi *FileInfo) Name() string { return fi.name } func (fi *FileInfo) Size() int64 { return fi.size } func (fi *FileInfo) Mode() os.FileMode { return fi.mode } -func (fi *FileInfo) ModTime() time.Time { return fi.mod_time } -func (fi *FileInfo) IsDir() bool { return fi.mode.IsDir() } +func (fi *FileInfo) ModTime() time.Time { return fi.modifiledTime } +func (fi *FileInfo) IsDir() bool { return fi.isDirectory } func (fi *FileInfo) Sys() interface{} { return nil } type WebDavFile struct { - fs *WebDavFileSystem - name string - isDirectory bool - off int64 + fs *WebDavFileSystem + name string + isDirectory bool + off int64 + entry *filer_pb.Entry + entryViewCache []filer2.VisibleInterval } -func NewWebDavFileSystem() (webdav.FileSystem, error) { - return &WebDavFileSystem{}, nil +func NewWebDavFileSystem(option *WebDavOption) (webdav.FileSystem, error) { + return &WebDavFileSystem{ + option: option, + }, nil +} + +func (fs *WebDavFileSystem) WithFilerClient(ctx context.Context, fn func(filer_pb.SeaweedFilerClient) error) error { + + return util.WithCachedGrpcClient(ctx, func(grpcConnection *grpc.ClientConn) error { + client := filer_pb.NewSeaweedFilerClient(grpcConnection) + return fn(client) + }, fs.option.FilerGrpcAddress, fs.option.GrpcDialOption) + } func clearName(name string) (string, error) { @@ -117,32 +141,55 @@ func clearName(name string) (string, error) { return name, nil } -func (fs *WebDavFileSystem) Mkdir(ctx context.Context, name string, perm os.FileMode) error { +func (fs *WebDavFileSystem) Mkdir(ctx context.Context, fullDirPath string, perm os.FileMode) error { - glog.V(2).Infof("WebDavFileSystem.Mkdir %v", name) + glog.V(2).Infof("WebDavFileSystem.Mkdir %v", fullDirPath) - if !strings.HasSuffix(name, "/") { - name += "/" + if !strings.HasSuffix(fullDirPath, "/") { + fullDirPath += "/" } var err error - if name, err = clearName(name); err != nil { + if fullDirPath, err = clearName(fullDirPath); err != nil { return err } - _, err = fs.stat(name) + _, err = fs.stat(ctx, fullDirPath) if err == nil { return os.ErrExist } base := "/" - for _, elem := range strings.Split(strings.Trim(name, "/"), "/") { + for _, elem := range strings.Split(strings.Trim(fullDirPath, "/"), "/") { base += elem + "/" - _, err = fs.stat(base) + _, err = fs.stat(ctx, base) if err != os.ErrNotExist { return err } - // _, err = fs.db.Exec(`insert into filesystem(name, content, mode, mod_time) values(?, '', ?, now())`, base, perm.Perm()|os.ModeDir) + err = fs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { + dir, name := filer2.FullPath(base).DirAndName() + request := &filer_pb.CreateEntryRequest{ + Directory: dir, + Entry: &filer_pb.Entry{ + Name: name, + IsDirectory: true, + Attributes: &filer_pb.FuseAttributes{ + Mtime: time.Now().Unix(), + Crtime: time.Now().Unix(), + FileMode: uint32(perm), + Uid: fs.option.Uid, + Gid: fs.option.Gid, + }, + }, + } + + glog.V(1).Infof("mkdir: %v", request) + if _, err := client.CreateEntry(ctx, request); err != nil { + return fmt.Errorf("mkdir %s/%s: %v", dir, name, err) + } + + return nil + }) if err != nil { return err } @@ -150,68 +197,116 @@ func (fs *WebDavFileSystem) Mkdir(ctx context.Context, name string, perm os.File return nil } -func (fs *WebDavFileSystem) OpenFile(ctx context.Context, name string, flag int, perm os.FileMode) (webdav.File, error) { +func (fs *WebDavFileSystem) OpenFile(ctx context.Context, fullFilePath string, flag int, perm os.FileMode) (webdav.File, error) { - glog.V(2).Infof("WebDavFileSystem.OpenFile %v", name) + glog.V(2).Infof("WebDavFileSystem.OpenFile %v", fullFilePath) var err error - if name, err = clearName(name); err != nil { + if fullFilePath, err = clearName(fullFilePath); err != nil { return nil, err } if flag&os.O_CREATE != 0 { // file should not have / suffix. - if strings.HasSuffix(name, "/") { + if strings.HasSuffix(fullFilePath, "/") { return nil, os.ErrInvalid } // based directory should be exists. - dir, _ := path.Split(name) - _, err := fs.stat(dir) + dir, _ := path.Split(fullFilePath) + _, err := fs.stat(ctx, dir) if err != nil { return nil, os.ErrInvalid } - _, err = fs.stat(name) + _, err = fs.stat(ctx, fullFilePath) if err == nil { if flag&os.O_EXCL != 0 { return nil, os.ErrExist } - fs.removeAll(name) + fs.removeAll(ctx, fullFilePath) } - // _, err = fs.db.Exec(`insert into filesystem(name, content, mode, mod_time) values(?, '', ?, now())`, name, perm.Perm()) + + dir, name := filer2.FullPath(fullFilePath).DirAndName() + err = fs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { + if _, err := client.CreateEntry(ctx, &filer_pb.CreateEntryRequest{ + Directory: dir, + Entry: &filer_pb.Entry{ + Name: name, + IsDirectory: perm&os.ModeDir > 0, + Attributes: &filer_pb.FuseAttributes{ + Mtime: time.Now().Unix(), + Crtime: time.Now().Unix(), + FileMode: uint32(perm), + Uid: fs.option.Uid, + Gid: fs.option.Gid, + Collection: fs.option.Collection, + Replication: "000", + TtlSec: 0, + }, + }, + }); err != nil { + return fmt.Errorf("create %s: %v", fullFilePath, err) + } + return nil + }) if err != nil { return nil, err } - return &WebDavFile{fs, name, false, 0}, nil + return &WebDavFile{ + fs: fs, + name: fullFilePath, + isDirectory: false, + }, nil } - fi, err := fs.stat(name) + fi, err := fs.stat(ctx, fullFilePath) if err != nil { return nil, os.ErrNotExist } - if !strings.HasSuffix(name, "/") && fi.IsDir() { - name += "/" + if !strings.HasSuffix(fullFilePath, "/") && fi.IsDir() { + fullFilePath += "/" } - return &WebDavFile{fs, name, true, 0}, nil + return &WebDavFile{ + fs: fs, + name: fullFilePath, + isDirectory: false, + }, nil } -func (fs *WebDavFileSystem) removeAll(name string) error { +func (fs *WebDavFileSystem) removeAll(ctx context.Context, fullFilePath string) error { var err error - if name, err = clearName(name); err != nil { + if fullFilePath, err = clearName(fullFilePath); err != nil { return err } - fi, err := fs.stat(name) + fi, err := fs.stat(ctx, fullFilePath) if err != nil { return err } if fi.IsDir() { - //_, err = fs.db.Exec(`delete from filesystem where name like $1 escape '\'`, strings.Replace(name, `%`, `\%`, -1)+`%`) + //_, err = fs.db.Exec(`delete from filesystem where fullFilePath like $1 escape '\'`, strings.Replace(fullFilePath, `%`, `\%`, -1)+`%`) } else { - //_, err = fs.db.Exec(`delete from filesystem where name = ?`, name) + //_, err = fs.db.Exec(`delete from filesystem where fullFilePath = ?`, fullFilePath) } + dir, name := filer2.FullPath(fullFilePath).DirAndName() + err = fs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { + + request := &filer_pb.DeleteEntryRequest{ + Directory: dir, + Name: name, + IsDeleteData: true, + } + + glog.V(3).Infof("removing entry: %v", request) + _, err := client.DeleteEntry(ctx, request) + if err != nil { + return fmt.Errorf("remove %s: %v", fullFilePath, err) + } + + return nil + }) return err } @@ -219,7 +314,7 @@ func (fs *WebDavFileSystem) RemoveAll(ctx context.Context, name string) error { glog.V(2).Infof("WebDavFileSystem.RemoveAll %v", name) - return fs.removeAll(name) + return fs.removeAll(ctx, name) } func (fs *WebDavFileSystem) Rename(ctx context.Context, oldName, newName string) error { @@ -234,7 +329,7 @@ func (fs *WebDavFileSystem) Rename(ctx context.Context, oldName, newName string) return err } - of, err := fs.stat(oldName) + of, err := fs.stat(ctx, oldName) if err != nil { return os.ErrExist } @@ -243,34 +338,55 @@ func (fs *WebDavFileSystem) Rename(ctx context.Context, oldName, newName string) newName += "/" } - _, err = fs.stat(newName) + _, err = fs.stat(ctx, newName) if err == nil { return os.ErrExist } - //_, err = fs.db.Exec(`update filesystem set name = ? where name = ?`, newName, oldName) - return err + oldDir, oldBaseName := filer2.FullPath(oldName).DirAndName() + newDir, newBaseName := filer2.FullPath(newName).DirAndName() + + return fs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error { + + request := &filer_pb.AtomicRenameEntryRequest{ + OldDirectory: oldDir, + OldName: oldBaseName, + NewDirectory: newDir, + NewName: newBaseName, + } + + _, err := client.AtomicRenameEntry(ctx, request) + if err != nil { + return fmt.Errorf("renaming %s/%s => %s/%s: %v", oldDir, oldBaseName, newDir, newBaseName, err) + } + + return nil + + }) } -func (fs *WebDavFileSystem) stat(name string) (os.FileInfo, error) { +func (fs *WebDavFileSystem) stat(ctx context.Context, fullFilePath string) (os.FileInfo, error) { var err error - if name, err = clearName(name); err != nil { + if fullFilePath, err = clearName(fullFilePath); err != nil { return nil, err } - //rows, err := fs.db.Query(`select name, format(length(content)/2, 0), mode, mod_time from filesystem where name = ?`, name) - if err != nil { - return nil, err - } var fi FileInfo - // err = rows.Scan(&fi.name, &fi.size, &fi.mode, &fi.mod_time) + entry, err := filer2.GetEntry(ctx, fs, fullFilePath) if err != nil { return nil, err } + fi.size = int64(filer2.TotalSize(entry.GetChunks())) + fi.name = fullFilePath + fi.mode = os.FileMode(entry.Attributes.FileMode) + fi.modifiledTime = time.Unix(entry.Attributes.Mtime, 0) + fi.isDirectory = entry.IsDirectory + _, fi.name = path.Split(path.Clean(fi.name)) if fi.name == "" { fi.name = "/" - fi.mod_time = time.Now() + fi.modifiledTime = time.Now() + fi.isDirectory = true } return &fi, nil } @@ -279,7 +395,7 @@ func (fs *WebDavFileSystem) Stat(ctx context.Context, name string) (os.FileInfo, glog.V(2).Infof("WebDavFileSystem.Stat %v", name) - return fs.stat(name) + return fs.stat(ctx, name) } func (f *WebDavFile) Write(p []byte) (int, error) { @@ -299,41 +415,105 @@ func (f *WebDavFile) Close() error { glog.V(2).Infof("WebDavFileSystem.Close %v", f.name) + if f.entry != nil { + f.entry = nil + f.entryViewCache = nil + } + return nil } -func (f *WebDavFile) Read(p []byte) (int, error) { +func (f *WebDavFile) Read(p []byte) (readSize int, err error) { glog.V(2).Infof("WebDavFileSystem.Read %v", f.name) + ctx := context.Background() - var err error - //rows, err := f.fs.db.Query(`select mode, substr(content, ?, ?) from filesystem where name = ?`, 1+f.off*2, len(p)*2, f.name) + if f.entry == nil { + f.entry, err = filer2.GetEntry(ctx, f.fs, f.name) + } if err != nil { return 0, err } - //defer rows.Close() + if len(f.entry.Chunks) == 0 { + return 0, io.EOF + } + if f.entryViewCache == nil { + f.entryViewCache = filer2.NonOverlappingVisibleIntervals(f.entry.Chunks) + } + chunkViews := filer2.ViewFromVisibleIntervals(f.entryViewCache, f.off, len(p)) - return 0, io.EOF + totalRead, err := filer2.ReadIntoBuffer(ctx, f.fs, f.name, p, chunkViews, f.off) + if err != nil { + return 0, err + } + readSize = int(totalRead) + + f.off += totalRead + if readSize == 0 { + return 0, io.EOF + } + return } -func (f *WebDavFile) Readdir(count int) ([]os.FileInfo, error) { +func (f *WebDavFile) Readdir(count int) (ret []os.FileInfo, err error) { - glog.V(2).Infof("WebDavFileSystem.Readdir %v", f.name) + glog.V(2).Infof("WebDavFileSystem.Readdir %v count %d", f.name, count) + ctx := context.Background() - // return f.children[old:f.off], nil - return nil, nil + dir := f.name + if dir != "/" && strings.HasSuffix(dir, "/") { + dir = dir[:len(dir)-1] + } + + err = filer2.ReadDirAllEntries(ctx, f.fs, dir, func(entry *filer_pb.Entry) { + fi := FileInfo{ + size: int64(filer2.TotalSize(entry.GetChunks())), + name: entry.Name, + mode: os.FileMode(entry.Attributes.FileMode), + modifiledTime: time.Unix(entry.Attributes.Mtime, 0), + isDirectory: entry.IsDirectory, + } + + if !strings.HasSuffix(fi.name, "/") && fi.IsDir() { + fi.name += "/" + } + glog.V(4).Infof("entry: %v", fi.name) + ret = append(ret, &fi) + }) + + + old := f.off + if old >= int64(len(ret)) { + if count > 0 { + return nil, io.EOF + } + return nil, nil + } + if count > 0 { + f.off += int64(count) + if f.off > int64(len(ret)) { + f.off = int64(len(ret)) + } + } else { + f.off = int64(len(ret)) + old = 0 + } + + return ret[old:f.off], nil } func (f *WebDavFile) Seek(offset int64, whence int) (int64, error) { glog.V(2).Infof("WebDavFile.Seek %v %v %v", f.name, offset, whence) + ctx := context.Background() + var err error switch whence { case 0: f.off = 0 case 2: - if fi, err := f.fs.stat(f.name); err != nil { + if fi, err := f.fs.stat(ctx, f.name); err != nil { return 0, err } else { f.off = fi.Size() @@ -347,5 +527,7 @@ func (f *WebDavFile) Stat() (os.FileInfo, error) { glog.V(2).Infof("WebDavFile.Stat %v", f.name) - return f.fs.stat(f.name) + ctx := context.Background() + + return f.fs.stat(ctx, f.name) }