Merge pull request #33 from smkniazi/GH-14

Fixed default user/group information for new files create by a user t…

Author: smkniazi

Dir.go

@@ -14,6 +14,7 @@ import (
 	"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
@@ -206,7 +207,7 @@ func (dir *DirINode) LookupAttrs(name string, attrs *Attrs) error {
 	*attrs, err = dir.FileSystem.getDFSConnector().Stat(path.Join(dir.AbsolutePath(), name))
 	if err != nil {
 		// It is a warning as each time new file write tries to stat if the file exists
-		loginfo("Stat failed", Fields{Operation: Stat, Path: path.Join(dir.AbsolutePath(), name), Error: err})
+		loginfo("stat failed", Fields{Operation: Stat, Path: path.Join(dir.AbsolutePath(), name), Error: err})
 		return err
 	}
 
@@ -223,8 +224,16 @@ func (dir *DirINode) Mkdir(ctx context.Context, req *fuse.MkdirRequest) (fs.Node
 
 	err := dir.FileSystem.getDFSConnector().Mkdir(dir.AbsolutePathForChild(req.Name), req.Mode)
 	if err != nil {
+		loginfo("mkdir failed", Fields{Operation: Mkdir, Path: path.Join(dir.AbsolutePath(), req.Name), Error: err})
 		return nil, err
 	}
+	logdebug("mkdir successful", Fields{Operation: Mkdir, Path: path.Join(dir.AbsolutePath(), req.Name)})
+
+	err = dir.changeOwnership(dir.AbsolutePathForChild(req.Name), req.Uid, req.Gid)
+	if err != nil {
+		return nil, err
+	}
+
 	return dir.NodeFromAttrs(Attrs{Name: req.Name, Mode: req.Mode | os.ModeDir}), nil
 }
 
@@ -242,9 +251,29 @@ func (dir *DirINode) Create(ctx context.Context, req *fuse.CreateRequest, resp *
 		return nil, nil, err
 	}
 	file.AddHandle(handle)
+
+	err = dir.changeOwnership(dir.AbsolutePathForChild(req.Name), req.Uid, req.Gid)
+	if err != nil {
+		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)
+		group := ugcache.LookupGroupName(gid)
+		err := dir.FileSystem.getDFSConnector().Chown(path, user, group)
+		if err != nil {
+			dir.FileSystem.getDFSConnector().Remove(dir.AbsolutePathForChild(path))
+			return err
+		}
+		loginfo("Update ownership", Fields{Operation: Create, Path: path, User: user, Group: group})
+	}
+	return nil
+}
+
 // Responds on FUSE Remove request
 func (dir *DirINode) Remove(ctx context.Context, req *fuse.RemoveRequest) error {
 	dir.lockMutex()

FileSystem.go

@@ -148,6 +148,5 @@ func (filesystem *FileSystem) Statfs(ctx context.Context, req *fuse.StatfsReques
 func (filesystem *FileSystem) getDFSConnector() HdfsAccessor {
 	filesystem.hdfsAccessorsIndex = filesystem.hdfsAccessorsIndex + 1
 	index := filesystem.hdfsAccessorsIndex % len(filesystem.HdfsAccessors)
-	loginfo(fmt.Sprintf("Client index %d. len %d", index, len(filesystem.HdfsAccessors)), nil)
 	return filesystem.HdfsAccessors[index]
 }

HdfsAccessor.go

@@ -3,27 +3,24 @@
 package main
 
 import (
-	"errors"
 	"fmt"
 	"io"
 	"os"
-	"os/user"
-	"strconv"
 	"strings"
 	"sync"
 	"syscall"
 	"time"
 
 	"bazil.org/fuse"
 	"github.com/colinmarc/hdfs/v2"
-)
-
-const (
-	UGCacheTime = 3 * time.Second
+	"logicalclocks.com/hopsfs-mount/ugcache"
 )
 
 // Interface for accessing HDFS
 // Concurrency: thread safe: handles unlimited number of concurrent requests
+var hadoopUserName string = os.Getenv("HADOOP_USER_NAME")
+var hadoopUserID uint32 = 0
+
 type HdfsAccessor interface {
 	OpenRead(path string) (ReadSeekCloser, error) // Opens HDFS file for reading
 	CreateFile(path string,
@@ -49,18 +46,11 @@ type TLSConfig struct {
 }
 
 type hdfsAccessorImpl struct {
-	Clock               Clock                   // interface to get wall clock time
-	NameNodeAddresses   []string                // array of Address:port string for the name nodes
-	MetadataClient      *hdfs.Client            // HDFS client used for metadata operations
-	MetadataClientMutex sync.Mutex              // Serializing all metadata operations for simplicity (for now), TODO: allow N concurrent operations
-	UserNameToUidCache  map[string]UGCacheEntry // cache for converting usernames to UIDs
-	GroupNameToUidCache map[string]UGCacheEntry // cache for converting usernames to UIDs
-	TLSConfig           TLSConfig               // enable/disable using tls
-}
-
-type UGCacheEntry struct {
-	ID      uint32    // User/Group Id
-	Expires time.Time // Absolute time when this cache entry expires
+	Clock               Clock        // interface to get wall clock time
+	NameNodeAddresses   []string     // array of Address:port string for the name nodes
+	MetadataClient      *hdfs.Client // HDFS client used for metadata operations
+	MetadataClientMutex sync.Mutex   // Serializing all metadata operations for simplicity (for now), TODO: allow N concurrent operations
+	TLSConfig           TLSConfig    // enable/disable using tls
 }
 
 var _ HdfsAccessor = (*hdfsAccessorImpl)(nil) // ensure hdfsAccessorImpl implements HdfsAccessor
@@ -70,11 +60,9 @@ func NewHdfsAccessor(nameNodeAddresses string, clock Clock, tlsConfig TLSConfig)
 	nns := strings.Split(nameNodeAddresses, ",")
 
 	this := &hdfsAccessorImpl{
-		NameNodeAddresses:   nns,
-		Clock:               clock,
-		UserNameToUidCache:  make(map[string]UGCacheEntry),
-		GroupNameToUidCache: make(map[string]UGCacheEntry),
-		TLSConfig:           tlsConfig,
+		NameNodeAddresses: nns,
+		Clock:             clock,
+		TLSConfig:         tlsConfig,
 	}
 	return this, nil
 }
@@ -103,29 +91,29 @@ func (dfs *hdfsAccessorImpl) ConnectToNameNode() (*hdfs.Client, error) {
 	client, err := dfs.connectToNameNodeImpl()
 	if err != nil {
 		// Connection failed
-		return nil, errors.New(fmt.Sprintf("Fail to connect to name node with error: %s", err.Error()))
+		return nil, fmt.Errorf("fail to connect to name node with error: %s", err.Error())
 	}
 	return client, nil
 }
 
 // Performs an attempt to connect to the HDFS name
 func (dfs *hdfsAccessorImpl) connectToNameNodeImpl() (*hdfs.Client, error) {
-	hadoopUser := os.Getenv("HADOOP_USER_NAME")
-	if hadoopUser == "" {
-		u, err := user.Current()
+	if hadoopUserName == "" {
+		u, err := ugcache.CurrentUserName()
 		if err != nil {
-			return nil, fmt.Errorf("Couldn't determine user: %s", err)
+			return nil, fmt.Errorf("couldn't determine user: %s", err)
 		}
-		hadoopUser = u.Username
+		hadoopUserName = u
 	}
-	loginfo(fmt.Sprintf("Connecting as user: %s", hadoopUser), nil)
+	hadoopUserID = ugcache.LookupUId(hadoopUserName)
+	loginfo(fmt.Sprintf("Connecting as user: %s, UID: %d", hadoopUserName, hadoopUserID), nil)
 
 	// Performing an attempt to connect to the name node
 	// Colinmar's hdfs implementation has supported the multiple name node connection
 	hdfsOptions := hdfs.ClientOptions{
 		Addresses: dfs.NameNodeAddresses,
 		TLS:       dfs.TLSConfig.TLS,
-		User:      hadoopUser,
+		User:      hadoopUserName,
 	}
 
 	if dfs.TLSConfig.TLS {
@@ -278,11 +266,11 @@ func (dfs *hdfsAccessorImpl) AttrsFromFileInfo(fileInfo os.FileInfo) Attrs {
 		Name:   fileInfo.Name(),
 		Mode:   mode,
 		Size:   fi.Length(),
-		Uid:    dfs.LookupUid(fi.Owner()),
+		Uid:    ugcache.LookupUId(fi.Owner()),
 		Mtime:  modificationTime,
 		Ctime:  modificationTime,
 		Crtime: modificationTime,
-		Gid:    dfs.LookupGid(fi.OwnerGroup())}
+		Gid:    ugcache.LookupGid(fi.OwnerGroup())}
 }
 
 func (dfs *hdfsAccessorImpl) AttrsFromFsInfo(fsInfo hdfs.FsInfo) FsInfo {
@@ -296,69 +284,6 @@ func HadoopTimestampToTime(timestamp uint64) time.Time {
 	return time.Unix(int64(timestamp)/1000, 0)
 }
 
-// Performs a cache-assisted lookup of UID by username
-func (dfs *hdfsAccessorImpl) LookupUid(userName string) uint32 {
-	if userName == "" {
-		return 0
-	}
-	// Note: this method is called under MetadataClientMutex, so accessing the cache dirctionary is safe
-	cacheEntry, ok := dfs.UserNameToUidCache[userName]
-	if ok && dfs.Clock.Now().Before(cacheEntry.Expires) {
-		return cacheEntry.ID
-	}
-
-	u, err := user.Lookup(userName)
-	if u != nil {
-		var uid64 uint64
-		if err == nil {
-			// UID is returned as string, need to parse it
-			uid64, err = strconv.ParseUint(u.Uid, 10, 32)
-		}
-		if err != nil {
-			uid64 = (1 << 31) - 1
-		}
-		dfs.UserNameToUidCache[userName] = UGCacheEntry{
-			ID:      uint32(uid64),
-			Expires: dfs.Clock.Now().Add(UGCacheTime)}
-		return uint32(uid64)
-
-	} else {
-		return 0
-	}
-}
-
-// Performs a cache-assisted lookup of GID by grooupname
-func (dfs *hdfsAccessorImpl) LookupGid(groupName string) uint32 {
-	if groupName == "" {
-		return 0
-	}
-	// Note: this method is called under MetadataClientMutex, so accessing the cache dirctionary is safe
-	cacheEntry, ok := dfs.GroupNameToUidCache[groupName]
-	if ok && dfs.Clock.Now().Before(cacheEntry.Expires) {
-		return cacheEntry.ID
-	}
-
-	g, err := user.LookupGroup(groupName)
-	if g != nil {
-		var gid64 uint64
-		if err == nil {
-			// GID is returned as string, need to parse it
-			gid64, err = strconv.ParseUint(g.Gid, 10, 32)
-		}
-		if err != nil {
-			gid64 = (1 << 31) - 1
-		}
-		dfs.GroupNameToUidCache[groupName] = UGCacheEntry{
-			ID:      uint32(gid64),
-			Expires: dfs.Clock.Now().Add(UGCacheTime)}
-		return uint32(gid64)
-
-	} else {
-		logwarn(fmt.Sprintf("Group not found %s", groupName), nil)
-		return 0
-	}
-}
-
 // Returns true if err==nil or err is expected (benign) error which should be propagated directoy to the caller
 func IsSuccessOrNonRetriableError(err error) bool {
 	if err == nil {

Log.go

@@ -36,10 +36,12 @@ const (
 	Flush             = "flush"
 	Close             = "close"
 	Stat              = "stat"
+	Mkdir             = "mkdir"
 	StatFS            = "statfs"
 	UID               = "uid"
 	GID               = "gid"
 	User              = "user"
+	Group             = "group"
 	Holes             = "holes"
 	Seeks             = "seeks"
 	HardSeeks         = "hard_seeks"