Merge pull request #27 from smkniazi/GH-26

Gh 26

Author: smkniazi

Dir.go

@@ -208,9 +208,6 @@ func (dir *Dir) LookupAttrs(name string, attrs *Attrs) error {
 	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})
-		if pathError, ok := err.(*os.PathError); ok && (pathError.Err == os.ErrNotExist) {
-			return fuse.ENOENT
-		}
 		return err
 	}
 

HdfsAccessor.go

@@ -11,6 +11,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"syscall"
 	"time"
 
 	"bazil.org/fuse"
@@ -166,7 +167,7 @@ func (dfs *hdfsAccessorImpl) OpenRead(path string) (ReadSeekCloser, error) {
 	}
 	reader, err := dfs.MetadataClient.Open(path)
 	if err != nil {
-		return nil, err
+		return nil, unwrapAndTranslateError(err)
 	}
 	return NewHdfsReader(reader), nil
 }
@@ -182,7 +183,7 @@ func (dfs *hdfsAccessorImpl) CreateFile(path string, mode os.FileMode, overwrite
 	}
 	writer, err := dfs.MetadataClient.CreateFile(path, 3, 64*1024*1024, mode, overwrite)
 	if err != nil {
-		return nil, err
+		return nil, unwrapAndTranslateError(err)
 	}
 
 	return NewHdfsWriter(writer), nil
@@ -201,12 +202,12 @@ func (dfs *hdfsAccessorImpl) ReadDir(path string) ([]Attrs, error) {
 	if err != nil {
 		if IsSuccessOrNonRetriableError(err) {
 			// benign error (e.g. path not found)
-			return nil, err
+			return nil, unwrapAndTranslateError(err)
 		}
 		// We've got error from this client, setting to nil, so we try another one next time
 		dfs.MetadataClient = nil
 		// TODO: attempt to gracefully close the conenction
-		return nil, err
+		return nil, unwrapAndTranslateError(err)
 	}
 	allAttrs := make([]Attrs, len(files))
 	for i, fileInfo := range files {
@@ -230,12 +231,12 @@ func (dfs *hdfsAccessorImpl) Stat(path string) (Attrs, error) {
 	if err != nil {
 		if IsSuccessOrNonRetriableError(err) {
 			// benign error (e.g. path not found)
-			return Attrs{}, err
+			return Attrs{}, unwrapAndTranslateError(err)
 		}
 		// We've got error from this client, setting to nil, so we try another one next time
 		dfs.MetadataClient = nil
 		// TODO: attempt to gracefully close the conenction
-		return Attrs{}, err
+		return Attrs{}, unwrapAndTranslateError(err)
 	}
 	return dfs.AttrsFromFileInfo(fileInfo), nil
 }
@@ -254,10 +255,10 @@ func (dfs *hdfsAccessorImpl) StatFs() (FsInfo, error) {
 	fsInfo, err := dfs.MetadataClient.StatFs()
 	if err != nil {
 		if IsSuccessOrNonRetriableError(err) {
-			return FsInfo{}, err
+			return FsInfo{}, unwrapAndTranslateError(err)
 		}
 		dfs.MetadataClient = nil
-		return FsInfo{}, err
+		return FsInfo{}, unwrapAndTranslateError(err)
 	}
 	return dfs.AttrsFromFsInfo(fsInfo), nil
 }
@@ -360,10 +361,49 @@ func (dfs *hdfsAccessorImpl) LookupGid(groupName string) uint32 {
 
 // 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 || err == io.EOF || err == fuse.EEXIST {
+	if err == nil {
 		return true
 	}
-	if pathError, ok := err.(*os.PathError); ok && (pathError.Err == os.ErrNotExist || pathError.Err == os.ErrPermission) {
+
+	return isNonRetriableError(unwrapAndTranslateError(err))
+}
+
+func unwrapAndTranslateError(err error) error {
+	var e error
+	pathError, ok := err.(*os.PathError)
+	if ok {
+		e = pathError.Err
+	} else {
+		e = err
+	}
+
+	if e == os.ErrNotExist {
+		return syscall.ENOENT
+	}
+	if e == os.ErrPermission {
+		return syscall.EPERM
+	}
+	if e == os.ErrExist {
+		return syscall.EEXIST
+	}
+
+	return e
+}
+
+func isNonRetriableError(err error) bool {
+	if err == io.EOF ||
+		err == fuse.EEXIST ||
+		err == syscall.ENOENT ||
+		err == syscall.EACCES ||
+		err == syscall.ENOTEMPTY ||
+		err == syscall.EEXIST ||
+		err == syscall.EROFS ||
+		err == syscall.EDQUOT ||
+		err == syscall.ENOLINK ||
+		err == os.ErrNotExist ||
+		err == os.ErrPermission ||
+		err == os.ErrExist ||
+		err == os.ErrClosed {
 		return true
 	} else {
 		return false

go.mod

@@ -14,7 +14,7 @@ require (
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0 // indirect
 )
 
-replace github.com/colinmarc/hdfs/v2 v2.2.0 => github.com/logicalclocks/hopsfs-go-client/v2 v2.4.1
+replace github.com/colinmarc/hdfs/v2 v2.2.0 => github.com/logicalclocks/hopsfs-go-client/v2 v2.4.2
 
 //replace github.com/colinmarc/hdfs/v2 v2.2.0 => /home/salman/code/hops/hopsfs-go/hopsfs-go-client
 //replace bazil.org/fuse v0.0.0-20200524192727-fb710f7dfd05 => /home/salman/code/hops/hopsfs-go/fuse

main.go

@@ -29,6 +29,7 @@ var allowedPrefixesString *string
 var expandZips *bool
 var readOnly *bool
 var tls *bool
+var maxAttempts int
 
 func main() {
 
@@ -44,7 +45,7 @@ func main() {
 
 	allowedPrefixes := strings.Split(*allowedPrefixesString, ",")
 
-	retryPolicy.MaxAttempts += 1 // converting # of retry attempts to total # of attempts
+	retryPolicy.MaxAttempts = maxAttempts
 
 	tlsConfig := TLSConfig{
 		TLS:               *tls,
@@ -134,7 +135,7 @@ var Usage = func() {
 func parseArgsAndInitLogger(retryPolicy *RetryPolicy) {
 	lazyMount = flag.Bool("lazy", false, "Allows to mount HopsFS filesystem before HopsFS is available")
 	flag.DurationVar(&retryPolicy.TimeLimit, "retryTimeLimit", 5*time.Minute, "time limit for all retry attempts for failed operations")
-	flag.IntVar(&retryPolicy.MaxAttempts, "retryMaxAttempts", 99999999, "Maxumum retry attempts for failed operations")
+	flag.IntVar(&maxAttempts, "retryMaxAttempts", 10, "Maxumum retry attempts for failed operations")
 	flag.DurationVar(&retryPolicy.MinDelay, "retryMinDelay", 1*time.Second, "minimum delay between retries (note, first retry always happens immediatelly)")
 	flag.DurationVar(&retryPolicy.MaxDelay, "retryMaxDelay", 60*time.Second, "maximum delay between retries")
 	allowedPrefixesString = flag.String("allowedPrefixes", "*", "Comma-separated list of allowed path prefixes on the remote file system, if specified the mount point will expose access to those prefixes only")
@@ -163,7 +164,7 @@ func parseArgsAndInitLogger(retryPolicy *RetryPolicy) {
 	}
 	initLogger(logLevel, false, logFile)
 
-	loginfo(fmt.Sprintf("Staging dir is:%s, Using TLS: %v, LogFile: %s", stagingDir, *tls, logFile), nil)
+	loginfo(fmt.Sprintf("Staging dir is:%s, Using TLS: %v, RetryAttempts: %d,  LogFile: %s", stagingDir, *tls, retryPolicy.MaxAttempts, logFile), nil)
 	loginfo(fmt.Sprintf("hopsfs-mount: current head GITCommit: %s Built time: %s Built by: %s ", GITCOMMIT, BUILDTIME, HOSTNAME), nil)
 }