Merge pull request #43 from smkniazi/GH-41

Gh 41

Author: smkniazi

Attrs.go

@@ -31,7 +31,7 @@ type FsInfo struct {
 }
 
 // Converts Attrs datastructure into FUSE represnetation
-func (attrs *Attrs) Attr(a *fuse.Attr) error {
+func (attrs *Attrs) ConvertAttrToFuse(a *fuse.Attr) error {
 	a.Inode = attrs.Inode
 	a.Mode = attrs.Mode
 	if (a.Mode & os.ModeDir) == 0 {

Dir.go

@@ -5,16 +5,15 @@ package main
 import (
 	"fmt"
 	"os"
-	"os/user"
 	"path"
+	"path/filepath"
 	"strings"
 	"sync"
 	"time"
 
 	"bazil.org/fuse"
 	"bazil.org/fuse/fs"
 	"golang.org/x/net/context"
-	"logicalclocks.com/hopsfs-mount/ugcache"
 )
 
 // Encapsulates state and operations for directory node on the HDFS file system
@@ -49,7 +48,7 @@ func (dir *DirINode) AbsolutePathForChild(name string) string {
 	if path != "/" {
 		path = path + "/"
 	}
-	return path + name
+	return path + filepath.Base(name)
 }
 
 // Responds on FUSE request to get directory attributes
@@ -63,7 +62,7 @@ func (dir *DirINode) Attr(ctx context.Context, a *fuse.Attr) error {
 		}
 
 	}
-	return dir.Attrs.Attr(a)
+	return dir.Attrs.ConvertAttrToFuse(a)
 }
 
 func (dir *DirINode) EntriesGet(name string) *fs.Node {
@@ -229,15 +228,16 @@ func (dir *DirINode) Mkdir(ctx context.Context, req *fuse.MkdirRequest) (fs.Node
 	}
 	logdebug("mkdir successful", Fields{Operation: Mkdir, Path: path.Join(dir.AbsolutePath(), req.Name)})
 
-	err = dir.changeOwnership(dir.AbsolutePathForChild(req.Name), req.Uid, req.Gid)
+	err = ChownOp(&dir.Attrs, dir.FileSystem, dir.AbsolutePathForChild(req.Name), req.Uid, req.Gid)
 	if err != nil {
-		logwarn("Unable to change ownership of new dir", Fields{Operation: Create, Path: dir.AbsolutePathForChild(req.Name), UID: req.Uid, GID: req.Gid})
+		logwarn("Unable to change ownership of new dir", Fields{Operation: Create, Path: dir.AbsolutePathForChild(req.Name),
+			UID: req.Uid, GID: req.Gid, Error: err})
 		//unable to change the ownership of the directory. so delete it as the operation as a whole failed
 		dir.FileSystem.getDFSConnector().Remove(dir.AbsolutePathForChild(req.Name))
 		return nil, err
 	}
 
-	return dir.NodeFromAttrs(Attrs{Name: req.Name, Mode: req.Mode | os.ModeDir}), nil
+	return dir.NodeFromAttrs(Attrs{Name: req.Name, Mode: req.Mode | os.ModeDir, Uid: req.Uid, Gid: req.Gid}), nil
 }
 
 // Responds on FUSE Create request
@@ -253,36 +253,24 @@ func (dir *DirINode) Create(ctx context.Context, req *fuse.CreateRequest, resp *
 		//TODO remove the entry from the cache
 		return nil, nil, err
 	}
-	file.AddHandle(handle)
 
-	err = dir.changeOwnership(dir.AbsolutePathForChild(req.Name), req.Uid, req.Gid)
+	file.AddHandle(handle)
+	err = ChownOp(&dir.Attrs, dir.FileSystem, dir.AbsolutePathForChild(req.Name), req.Uid, req.Gid)
 	if err != nil {
-		logwarn("Unable to change ownership of new file", Fields{Operation: Create, Path: dir.AbsolutePathForChild(req.Name), UID: req.Uid, GID: req.Gid})
+		logwarn("Unable to change ownership of new file", Fields{Operation: Create, Path: dir.AbsolutePathForChild(req.Name),
+			UID: req.Uid, GID: req.Gid, Error: err})
 		//unable to change the ownership of the file. so delete it as the operation as a whole failed
 		dir.FileSystem.getDFSConnector().Remove(dir.AbsolutePathForChild(req.Name))
 		return nil, nil, err
 	}
 
-	return file, handle, nil
-}
-
-func (dir *DirINode) changeOwnership(path string, uid uint32, gid uint32) error {
-	if hadoopUserID != uid { // the file is created by an other user, so change the ownership information
-		user := ugcache.LookupUserName(uid)
-		if user == "" {
-			logwarn("Unable to find user information", Fields{Operation: Chown, Path: path, UID: uid, GID: gid})
-		}
-		group := ugcache.LookupGroupName(gid)
-		if group == "" {
-			logwarn("Unable to find group information", Fields{Operation: Chown, Path: path, UID: uid, GID: gid})
-		}
-		err := dir.FileSystem.getDFSConnector().Chown(path, user, group)
-		if err != nil {
-			return err
-		}
-		loginfo("Updated ownership", Fields{Operation: Create, Path: path, User: user, Group: group})
+	//update the attributes of the file now
+	err = dir.LookupAttrs(file.Attrs.Name, &file.Attrs)
+	if err != nil {
+		return nil, nil, err
 	}
-	return nil
+
+	return file, handle, nil
 }
 
 // Responds on FUSE Remove request
@@ -332,59 +320,33 @@ func (dir *DirINode) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp
 	dir.lockMutex()
 	defer dir.unlockMutex()
 
-	// Get the filepath, so chmod in hdfs can work
+	if req.Valid.Size() {
+		return fmt.Errorf("unsupported operation. Can not set size of a directory")
+	}
+
 	path := dir.AbsolutePath()
-	var err error
 
 	if req.Valid.Mode() {
-		loginfo("Setting attributes", Fields{Operation: Chmod, Path: path, Mode: req.Mode})
-		(func() {
-			err = dir.FileSystem.getDFSConnector().Chmod(path, req.Mode)
-			if err != nil {
-				return
-			}
-		})()
-
-		if err != nil {
-			logerror("Failed to set attributes", Fields{Operation: Chmod, Path: path, Mode: req.Mode, Error: err})
-		} else {
-			dir.Attrs.Mode = req.Mode
+		if err := ChmodOp(&dir.Attrs, dir.FileSystem, path, req, resp); err != nil {
+			logwarn("Setattr (chmod) failed. ", Fields{Operation: Chmod, Path: path, Mode: req.Mode})
+			return err
 		}
 	}
 
-	if req.Valid.Uid() {
-		u, err := user.LookupId(fmt.Sprint(req.Uid))
-		owner := fmt.Sprint(req.Uid)
-		group := fmt.Sprint(req.Gid)
-		if err != nil {
-			logerror(fmt.Sprintf("Chown: username for uid %d not found, use uid/gid instead", req.Uid),
-				Fields{Operation: Chown, Path: path, User: u, UID: owner, GID: group, Error: err})
-		} else {
-			owner = u.Username
-			group = owner // hardcoded the group same as owner until LookupGroupId available
+	if req.Valid.Uid() || req.Valid.Gid() {
+		if err := SetAttrChownOp(&dir.Attrs, dir.FileSystem, path, req, resp); err != nil {
+			logwarn("Setattr (chown/chgrp )failed", Fields{Operation: Chmod, Path: path, UID: req.Uid, GID: req.Gid})
+			return err
 		}
+	}
 
-		loginfo("Setting attributes", Fields{Operation: Chown, Path: path, User: u, UID: owner, GID: group})
-		(func() {
-			err = dir.FileSystem.getDFSConnector().Chown(path, owner, group)
-			if err != nil {
-				return
-			}
-		})()
-
-		if err != nil {
-			logerror("Failed to set attributes", Fields{Operation: Chown, Path: path, User: u, UID: owner, GID: group, Error: err})
-		} else {
-			dir.Attrs.Uid = req.Uid
-			dir.Attrs.Gid = req.Gid
-		}
+	if err := UpdateTS(&dir.Attrs, dir.FileSystem, path, req, resp); err != nil {
+		return err
 	}
 
-	return err
+	return nil
 }
 
-var dirLockTime time.Time = time.Time{}
-
 func (dir *DirINode) lockMutex() {
 	dir.mutex.Lock()
 }

Dir_test.go

@@ -131,32 +131,36 @@ func TestLookupWithFiltering(t *testing.T) {
 
 // Testing Mkdir
 func TestMkdir(t *testing.T) {
+	dir := "/foo"
 	mockCtrl := gomock.NewController(t)
 	mockClock := &MockClock{}
 	hdfsAccessor := NewMockHdfsAccessor(mockCtrl)
+	hdfsAccessor.EXPECT().Chown(dir, gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 	fs, _ := NewFileSystem([]HdfsAccessor{hdfsAccessor}, "/", []string{"foo", "bar"}, false, false, NewDefaultRetryPolicy(mockClock), mockClock)
 	root, _ := fs.Root()
-	hdfsAccessor.EXPECT().Mkdir("/foo", os.FileMode(0757)|os.ModeDir).Return(nil)
+	hdfsAccessor.EXPECT().Mkdir(dir, os.FileMode(0757)|os.ModeDir).Return(nil)
 	node, err := root.(*DirINode).Mkdir(nil, &fuse.MkdirRequest{Name: "foo", Mode: os.FileMode(0757) | os.ModeDir})
 	assert.Nil(t, err)
 	assert.Equal(t, "foo", node.(*DirINode).Attrs.Name)
 }
 
 // Testing Chmod and Chown
 func TestSetattr(t *testing.T) {
+	dir := "/foo"
 	mockCtrl := gomock.NewController(t)
 	mockClock := &MockClock{}
 	hdfsAccessor := NewMockHdfsAccessor(mockCtrl)
+	hdfsAccessor.EXPECT().Chown(dir, gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 	fs, _ := NewFileSystem([]HdfsAccessor{hdfsAccessor}, "/", []string{"foo", "bar"}, false, false, NewDefaultRetryPolicy(mockClock), mockClock)
 	root, _ := fs.Root()
-	hdfsAccessor.EXPECT().Mkdir("/foo", os.FileMode(0757)|os.ModeDir).Return(nil)
+	hdfsAccessor.EXPECT().Mkdir(dir, os.FileMode(0757)|os.ModeDir).Return(nil)
 	node, _ := root.(*DirINode).Mkdir(nil, &fuse.MkdirRequest{Name: "foo", Mode: os.FileMode(0757) | os.ModeDir})
-	hdfsAccessor.EXPECT().Chmod("/foo", os.FileMode(0777)).Return(nil)
+	hdfsAccessor.EXPECT().Chmod(dir, os.FileMode(0777)).Return(nil).AnyTimes()
 	err := node.(*DirINode).Setattr(nil, &fuse.SetattrRequest{Mode: os.FileMode(0777), Valid: fuse.SetattrMode}, &fuse.SetattrResponse{})
 	assert.Nil(t, err)
 	assert.Equal(t, os.FileMode(0777), node.(*DirINode).Attrs.Mode)
 
-	hdfsAccessor.EXPECT().Chown("/foo", "root", "root").Return(nil)
+	hdfsAccessor.EXPECT().Chown(dir, "root", gomock.Any()).Return(nil).AnyTimes()
 	err = node.(*DirINode).Setattr(nil, &fuse.SetattrRequest{Uid: 0, Valid: fuse.SetattrUid}, &fuse.SetattrResponse{})
 	assert.Nil(t, err)
 	assert.Equal(t, uint32(0), node.(*DirINode).Attrs.Uid)

File.go

@@ -8,7 +8,6 @@ import (
 	"io/ioutil"
 	"math/rand"
 	"os"
-	"os/user"
 	"path"
 	"sync"
 	"syscall"
@@ -50,13 +49,29 @@ func (file *FileINode) Attr(ctx context.Context, a *fuse.Attr) error {
 	file.lockFile()
 	defer file.unlockFile()
 
-	if file.FileSystem.Clock.Now().After(file.Attrs.Expires) {
-		err := file.Parent.LookupAttrs(file.Attrs.Name, &file.Attrs)
+	// if the file is open for writing then update the file length and mtime
+	// from the straging file.
+	// Otherwise read the stats from the cache if it is valid.
+
+	if lrwfp, ok := file.handle.(*LocalRWFileProxy); ok {
+		fileInfo, err := lrwfp.localFile.Stat()
 		if err != nil {
+			logwarn("stat failed on staging file", Fields{Operation: Stat, Path: file.AbsolutePath(), Error: err})
 			return err
 		}
+		// update the local cache
+		file.Attrs.Size = uint64(fileInfo.Size())
+		file.Attrs.Mtime = fileInfo.ModTime()
+	} else {
+		if file.FileSystem.Clock.Now().After(file.Attrs.Expires) {
+			err := file.Parent.LookupAttrs(file.Attrs.Name, &file.Attrs)
+			if err != nil {
+				return err
+			}
+		}
 	}
-	return file.Attrs.Attr(a)
+	return file.Attrs.ConvertAttrToFuse(a)
+
 }
 
 // Responds to the FUSE file open request (creates new file handle)
@@ -111,7 +126,6 @@ func (file *FileINode) RemoveHandle(handle *FileHandle) {
 	//close the staging file if it is the last handle
 	if len(file.activeHandles) == 0 {
 		file.closeStaging()
-		logdebug("Staging file is closed.", file.logInfo(Fields{Operation: Close}))
 	} else {
 		logtrace("Staging file is not closed.", file.logInfo(Fields{Operation: Close}))
 	}
@@ -156,67 +170,37 @@ func (file *FileINode) Setattr(ctx context.Context, req *fuse.SetattrRequest, re
 	defer file.unlockFile()
 
 	if req.Valid.Size() {
-		var retErr error
+		var err error = nil
 		for _, handle := range file.activeHandles {
-			if handle.dataChanged() { // to only write enabled handles
-				err := handle.Truncate(int64(req.Size))
-				if err != nil {
-					retErr = err
-				}
+			err := handle.Truncate(int64(req.Size))
+			if err != nil {
+				err = err
 			}
+			resp.Attr.Size = req.Size
+			file.Attrs.Size = req.Size
 		}
-		return retErr
+		return err
 	}
 
-	// Get the filepath, so chmod in hdfs can work
 	path := file.AbsolutePath()
-	var err error
 
 	if req.Valid.Mode() {
-		loginfo("Setting attributes", Fields{Operation: Chmod, Path: path, Mode: req.Mode})
-		(func() {
-			err = file.FileSystem.getDFSConnector().Chmod(path, req.Mode)
-			if err != nil {
-				return
-			}
-		})()
-
-		if err != nil {
-			logerror("Failed to set attributes", Fields{Operation: Chmod, Path: path, Mode: req.Mode, Error: err})
-		} else {
-			file.Attrs.Mode = req.Mode
+		if err := ChmodOp(&file.Attrs, file.FileSystem, path, req, resp); err != nil {
+			return err
 		}
 	}
 
-	if req.Valid.Uid() {
-		u, err := user.LookupId(fmt.Sprint(req.Uid))
-		owner := fmt.Sprint(req.Uid)
-		group := fmt.Sprint(req.Gid)
-		if err != nil {
-			logerror(fmt.Sprintf("Chown: username for uid %d not found, use uid/gid instead", req.Uid),
-				Fields{Operation: Chown, Path: path, User: u, UID: owner, GID: group, Error: err})
-		} else {
-			owner = u.Username
-			group = owner // hardcoded the group same as owner
+	if req.Valid.Uid() || req.Valid.Gid() {
+		if err := SetAttrChownOp(&file.Attrs, file.FileSystem, path, req, resp); err != nil {
+			return err
 		}
+	}
 
-		loginfo("Setting attributes", Fields{Operation: Chown, Path: path, User: u, UID: owner, GID: group})
-		(func() {
-			err = file.FileSystem.getDFSConnector().Chown(path, fmt.Sprint(req.Uid), fmt.Sprint(req.Gid))
-			if err != nil {
-				return
-			}
-		})()
-
-		if err != nil {
-			logerror("Failed to set attributes", Fields{Operation: Chown, Path: path, User: u, UID: owner, GID: group, Error: err})
-		} else {
-			file.Attrs.Uid = req.Uid
-			file.Attrs.Gid = req.Gid
-		}
+	if err := UpdateTS(&file.Attrs, file.FileSystem, path, req, resp); err != nil {
+		return err
 	}
 
-	return err
+	return nil
 }
 
 func (file *FileINode) countActiveHandles() int {
@@ -326,6 +310,7 @@ func (file *FileINode) NewFileHandle(existsInDFS bool, flags fuse.OpenFlags) (*F
 			loginfo("Opened file, RO handle", fh.logInfo(Fields{Operation: operation, Flags: fh.fileFlags}))
 		}
 	}
+
 	return fh, nil
 }
 

FileHandleWriter_test.go

@@ -27,6 +27,7 @@ func TestReadWriteFile(t *testing.T) {
 	hdfswriter.EXPECT().Close().Return(nil).AnyTimes()
 	hdfsAccessor.EXPECT().CreateFile(fileName, os.FileMode(0757), gomock.Any()).Return(hdfswriter, nil).AnyTimes()
 	hdfsAccessor.EXPECT().Stat(fileName).Return(Attrs{Name: fileName}, nil).AnyTimes()
+	hdfsAccessor.EXPECT().Chown(fileName, gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 	hdfswriter.EXPECT().Close().Return(nil).AnyTimes()
 
 	root, _ := fs.Root()
@@ -64,14 +65,14 @@ func TestFaultTolerantWriteFile(t *testing.T) {
 	hdfswriter := NewMockHdfsWriter(mockCtrl)
 
 	hdfswriter.EXPECT().Close().Return(nil).AnyTimes()
-	hdfsAccessor.EXPECT().Stat(fileName).Return(Attrs{Name: fileName}, nil).AnyTimes()
+	hdfsAccessor.EXPECT().Stat(fileName).Return(Attrs{Name: fileName, Mode: os.FileMode(0757)}, nil).AnyTimes()
+	hdfsAccessor.EXPECT().Chown(fileName, gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 	hdfswriter.EXPECT().Close().Return(nil).AnyTimes()
 
 	hdfsAccessor.EXPECT().StatFs().Return(FsInfo{capacity: uint64(100), used: uint64(20), remaining: uint64(80)}, nil).AnyTimes()
 	hdfsAccessor.EXPECT().Remove("/testWriteFile_1").Return(nil).AnyTimes()
 	hdfsAccessor.EXPECT().CreateFile(fileName, os.FileMode(0757), gomock.Any()).DoAndReturn(func(path string,
 		mode os.FileMode, overwrite bool) (HdfsWriter, error) {
-		// fmt.Println("-.....> ")
 		return hdfswriter, nil
 	}).AnyTimes()