change filer API to gRPC

weed/filesys/dir.go

@@ -10,6 +10,7 @@ import (
 	"bazil.org/fuse"
 	"github.com/chrislusf/seaweedfs/weed/filer"
 	"sync"
+	"github.com/chrislusf/seaweedfs/weed/glog"
 )
 
 type Dir struct {
@@ -20,10 +21,22 @@ type Dir struct {
 }
 
 func (dir *Dir) Attr(context context.Context, attr *fuse.Attr) error {
-	attr.Mode = os.ModeDir | 0555
+	attr.Mode = os.ModeDir | 0777
 	return nil
 }
 
+func (dir *Dir) Mkdir(ctx context.Context, req *fuse.MkdirRequest) (fs.Node, error) {
+	dir.NodeMapLock.Lock()
+	defer dir.NodeMapLock.Unlock()
+
+	fmt.Printf("mkdir %+v\n", req)
+
+	node := &Dir{Path: path.Join(dir.Path, req.Name), wfs: dir.wfs}
+	dir.NodeMap[req.Name] = node
+
+	return node, nil
+}
+
 func (dir *Dir) Lookup(ctx context.Context, name string) (node fs.Node, err error) {
 
 	dir.NodeMapLock.Lock()
@@ -37,37 +50,66 @@ func (dir *Dir) Lookup(ctx context.Context, name string) (node fs.Node, err erro
 		return node, nil
 	}
 
-	if entry, err := filer.LookupDirectoryEntry(dir.wfs.filer, dir.Path, name); err == nil {
-		if !entry.Found {
-			return nil, fuse.ENOENT
+	var entry *filer.Entry
+	err = dir.wfs.withFilerClient(func(client filer.SeaweedFilerClient) error {
+
+		request := &filer.LookupDirectoryEntryRequest{
+			Directory: dir.Path,
+			Name:      name,
 		}
-		if entry.FileId != "" {
-			node = &File{FileId: filer.FileId(entry.FileId), Name: name, wfs: dir.wfs}
-		} else {
+
+		glog.V(1).Infof("lookup directory entry: %v", request)
+		resp, err := client.LookupDirectoryEntry(ctx, request)
+		if err != nil {
+			return err
+		}
+
+		entry = resp.Entry
+
+		return nil
+	})
+
+	if entry != nil {
+		if entry.IsDirectory {
 			node = &Dir{Path: path.Join(dir.Path, name), wfs: dir.wfs}
+		} else {
+			node = &File{FileId: filer.FileId(entry.FileId), Name: name, wfs: dir.wfs}
 		}
 		dir.NodeMap[name] = node
 		return node, nil
 	}
 
-	return nil, fuse.ENOENT
+	return nil, err
 }
 
-func (dir *Dir) ReadDirAll(ctx context.Context) ([]fuse.Dirent, error) {
-	var ret []fuse.Dirent
-	if dirs, e := filer.ListDirectories(dir.wfs.filer, dir.Path); e == nil {
-		for _, d := range dirs.Directories {
-			dirent := fuse.Dirent{Name: string(d), Type: fuse.DT_Dir}
-			ret = append(ret, dirent)
+func (dir *Dir) ReadDirAll(ctx context.Context) (ret []fuse.Dirent, err error) {
+
+	err = dir.wfs.withFilerClient(func(client filer.SeaweedFilerClient) error {
+
+		request := &filer.ListEntriesRequest{
+			Directory: dir.Path,
 		}
-	}
-	if files, e := filer.ListFiles(dir.wfs.filer, dir.Path, ""); e == nil {
-		for _, f := range files.Files {
-			dirent := fuse.Dirent{Name: f.Name, Type: fuse.DT_File}
-			ret = append(ret, dirent)
+
+		glog.V(1).Infof("read directory: %v", request)
+		resp, err := client.ListEntries(ctx, request)
+		if err != nil {
+			return err
 		}
-	}
-	return ret, nil
+
+		for _, entry := range resp.Entries {
+			if entry.IsDirectory {
+				dirent := fuse.Dirent{Name: entry.Name, Type: fuse.DT_Dir}
+				ret = append(ret, dirent)
+			} else {
+				dirent := fuse.Dirent{Name: entry.Name, Type: fuse.DT_File}
+				ret = append(ret, dirent)
+			}
+		}
+
+		return nil
+	})
+
+	return ret, err
 }
 
 func (dir *Dir) Remove(ctx context.Context, req *fuse.RemoveRequest) error {
@@ -75,13 +117,23 @@ func (dir *Dir) Remove(ctx context.Context, req *fuse.RemoveRequest) error {
 	dir.NodeMapLock.Lock()
 	defer dir.NodeMapLock.Unlock()
 
-	name := path.Join(dir.Path, req.Name)
-	err := filer.DeleteDirectoryOrFile(dir.wfs.filer, name, req.Dir)
-	if err != nil {
-		fmt.Printf("Delete file %s [ERROR] %s\n", name, err)
-	} else {
-		delete(dir.NodeMap, req.Name)
-	}
+	return dir.wfs.withFilerClient(func(client filer.SeaweedFilerClient) error {
+
+		request := &filer.DeleteEntryRequest{
+			Directory:   dir.Path,
+			Name:        req.Name,
+			IsDirectory: req.Dir,
+		}
+
+		glog.V(1).Infof("remove directory entry: %v", request)
+		_, err := client.DeleteEntry(ctx, request)
+		if err != nil {
+			return err
+		}
+
+		delete(dir.NodeMap, req.Name)
+
+		return nil
+	})
 
-	return err
 }

weed/filesys/file.go

@@ -6,8 +6,18 @@ import (
 
 	"bazil.org/fuse"
 	"github.com/chrislusf/seaweedfs/weed/filer"
+	"bazil.org/fuse/fs"
+	"github.com/chrislusf/seaweedfs/weed/glog"
 )
 
+var _ = fs.Node(&File{})
+// var _ = fs.NodeOpener(&File{})
+// var _ = fs.NodeFsyncer(&File{})
+var _ = fs.Handle(&File{})
+var _ = fs.HandleReadAller(&File{})
+// var _ = fs.HandleReader(&File{})
+var _ = fs.HandleWriter(&File{})
+
 type File struct {
 	FileId filer.FileId
 	Name   string
@@ -16,20 +26,49 @@ type File struct {
 
 func (file *File) Attr(context context.Context, attr *fuse.Attr) error {
 	attr.Mode = 0444
-	ret, err := filer.GetFileSize(file.wfs.filer, string(file.FileId))
-	if err == nil {
-		attr.Size = ret.Size
-	} else {
-		fmt.Printf("Get file %s attr [ERROR] %s\n", file.Name, err)
-	}
-	return err
+	return file.wfs.withFilerClient(func(client filer.SeaweedFilerClient) error {
+
+		request := &filer.GetFileAttributesRequest{
+			Name:      file.Name,
+			ParentDir: "", //TODO add parent folder
+			FileId:    string(file.FileId),
+		}
+
+		glog.V(1).Infof("read file size: %v", request)
+		resp, err := client.GetFileAttributes(context, request)
+		if err != nil {
+			return err
+		}
+
+		attr.Size = resp.Attributes.FileSize
+
+		return nil
+	})
 }
 
-func (file *File) ReadAll(ctx context.Context) ([]byte, error) {
-	ret, err := filer.GetFileContent(file.wfs.filer, string(file.FileId))
-	if err == nil {
-		return ret.Content, nil
-	}
-	fmt.Printf("Get file %s content [ERROR] %s\n", file.Name, err)
-	return nil, err
+func (file *File) ReadAll(ctx context.Context) (content []byte, err error) {
+
+	err = file.wfs.withFilerClient(func(client filer.SeaweedFilerClient) error {
+
+		request := &filer.GetFileContentRequest{
+			FileId: string(file.FileId),
+		}
+
+		glog.V(1).Infof("read file content: %v", request)
+		resp, err := client.GetFileContent(ctx, request)
+		if err != nil {
+			return err
+		}
+
+		content = resp.Content
+
+		return nil
+	})
+
+	return content, err
+}
+
+func (file *File) Write(ctx context.Context, req *fuse.WriteRequest, resp *fuse.WriteResponse) error {
+	fmt.Printf("write file %+v\n", req)
+	return nil
 }

weed/filesys/wfs.go

@@ -1,6 +1,11 @@
 package filesys
 
-import "bazil.org/fuse/fs"
+import (
+	"bazil.org/fuse/fs"
+	"fmt"
+	"google.golang.org/grpc"
+	"github.com/chrislusf/seaweedfs/weed/filer"
+)
 
 type WFS struct {
 	filer string
@@ -15,3 +20,16 @@ func NewSeaweedFileSystem(filer string) *WFS {
 func (wfs *WFS) Root() (fs.Node, error) {
 	return &Dir{Path: "/", wfs: wfs}, nil
 }
+
+func (wfs *WFS) withFilerClient(fn func(filer.SeaweedFilerClient) error) error {
+
+	grpcConnection, err := grpc.Dial(wfs.filer, grpc.WithInsecure())
+	if err != nil {
+		return fmt.Errorf("fail to dial %s: %v", wfs.filer, err)
+	}
+	defer grpcConnection.Close()
+
+	client := filer.NewSeaweedFilerClient(grpcConnection)
+
+	return fn(client)
+}