2.40

otherwise, on mac, during large directory deletion, if some ReaDirAll happens, the lib seems confused about the directories, and some child directories are not deleted.

.github/workflows/release.yml

@@ -60,5 +60,5 @@ jobs:
           ldflags: -extldflags -static -X github.com/chrislusf/seaweedfs/weed/util.COMMIT=${{github.sha}}
           # Where to run `go build .`
           project_path: weed
-          binary_name: weed-
+          binary_name: weed
           asset_name: "weed-${{ env.BUILD_TIME }}-${{ matrix.goos }}-${{ matrix.goarch }}"

docker/compose/local-dev-compose.yml

@@ -26,9 +26,10 @@ services:
   filer:
     image: chrislusf/seaweedfs:local
     ports:
+      - 8111:8111
       - 8888:8888
       - 18888:18888
-    command: '-v=1 filer -master="master:9333"'
+    command: '-v=1 filer -master="master:9333" -iam'
     depends_on:
       - master
       - volume

go.mod

@@ -42,6 +42,7 @@ require (
 	github.com/grpc-ecosystem/go-grpc-middleware v1.0.1-0.20190118093823-f849b5445de4
 	github.com/grpc-ecosystem/grpc-gateway v1.11.0 // indirect
 	github.com/jcmturner/gofork v1.0.0 // indirect
+	github.com/jinzhu/copier v0.2.8
 	github.com/json-iterator/go v1.1.10
 	github.com/karlseguin/ccache v2.0.3+incompatible // indirect
 	github.com/karlseguin/ccache/v2 v2.0.7
@@ -61,7 +62,7 @@ require (
 	github.com/prometheus/client_golang v1.3.0
 	github.com/rakyll/statik v0.1.7
 	github.com/rcrowley/go-metrics v0.0.0-20190826022208-cac0b30c2563 // indirect
-	github.com/seaweedfs/fuse v1.1.3
+	github.com/seaweedfs/fuse v1.1.4
 	github.com/seaweedfs/goexif v1.0.2
 	github.com/skip2/go-qrcode v0.0.0-20200617195104-da1b6568686e
 	github.com/spaolacci/murmur3 v1.1.0 // indirect

go.sum

@@ -439,6 +439,8 @@ github.com/jcmturner/gofork v0.0.0-20190328161633-dc7c13fece03 h1:FUwcHNlEqkqLjL
 github.com/jcmturner/gofork v0.0.0-20190328161633-dc7c13fece03/go.mod h1:MK8+TM0La+2rjBD4jE12Kj1pCCxK7d2LK/UM3ncEo0o=
 github.com/jcmturner/gofork v1.0.0 h1:J7uCkflzTEhUZ64xqKnkDxq3kzc96ajM1Gli5ktUem8=
 github.com/jcmturner/gofork v1.0.0/go.mod h1:MK8+TM0La+2rjBD4jE12Kj1pCCxK7d2LK/UM3ncEo0o=
+github.com/jinzhu/copier v0.2.8 h1:N8MbL5niMwE3P4dOwurJixz5rMkKfujmMRFmAanSzWE=
+github.com/jinzhu/copier v0.2.8/go.mod h1:24xnZezI2Yqac9J61UC6/dG/k76ttpq0DdJI3QmUvro=
 github.com/jmespath/go-jmespath v0.0.0-20160202185014-0b12d6b521d8/go.mod h1:Nht3zPeWKUH0NzdCt2Blrr5ys8VGpn0CEB0cQHVjt7k=
 github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af h1:pmfjZENx5imkbgOkpRUYLnmbU7UEFbjtDA2hxJ1ichM=
 github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af/go.mod h1:Nht3zPeWKUH0NzdCt2Blrr5ys8VGpn0CEB0cQHVjt7k=
@@ -691,6 +693,8 @@ github.com/seaweedfs/fuse v1.1.1 h1:WD51YFJcBViOx8I89jeqPD+vAKl4EowzBy9GUw0plb0=
 github.com/seaweedfs/fuse v1.1.1/go.mod h1:+PP6WlkrRUG6KPE+Th2EX5To/PjHaFsvqg/UgQ39aj8=
 github.com/seaweedfs/fuse v1.1.3 h1:0DddotXwSRGbYG2kynoJyr8GHCy30Z2SpdhP3vdyijY=
 github.com/seaweedfs/fuse v1.1.3/go.mod h1:+PP6WlkrRUG6KPE+Th2EX5To/PjHaFsvqg/UgQ39aj8=
+github.com/seaweedfs/fuse v1.1.4 h1:YYqkK86agMhXRSwR+wFbRI8ikMgk3kL6PNTna1MAHyQ=
+github.com/seaweedfs/fuse v1.1.4/go.mod h1:+PP6WlkrRUG6KPE+Th2EX5To/PjHaFsvqg/UgQ39aj8=
 github.com/seaweedfs/goexif v1.0.2 h1:p+rTXYdQ2mgxd+1JaTrQ9N8DvYuw9UH9xgYmJ+Bb29E=
 github.com/seaweedfs/goexif v1.0.2/go.mod h1:MrKs5LK0HXdffrdCZrW3OIMegL2xXpC6ThLyXMyjdrk=
 github.com/secsy/goftp v0.0.0-20190720192957-f31499d7c79a h1:C6IhVTxNkhlb0tlCB6JfHOUv1f0xHPK7V8X4HlJZEJw=

k8s/seaweedfs/Chart.yaml

@@ -1,5 +1,5 @@
 apiVersion: v1
 description: SeaweedFS
 name: seaweedfs
-appVersion: "2.39"
-version: 2.39
+appVersion: "2.40"
+version: 2.40

k8s/seaweedfs/templates/volume-service.yaml

@@ -23,6 +23,6 @@ spec:
     targetPort: {{ .Values.volume.metricsPort }}
     protocol: TCP
 {{- end }}
-selector:
+  selector:
     app: {{ template "seaweedfs.name" . }}
     component: volume

k8s/seaweedfs/values.yaml

@@ -4,7 +4,7 @@ global:
   registry: ""
   repository: ""
   imageName: chrislusf/seaweedfs
-  # imageTag: "2.39" - started using {.Chart.appVersion}
+  # imageTag: "2.40" - started using {.Chart.appVersion}
   imagePullPolicy: IfNotPresent
   imagePullSecrets: imagepullsecret
   restartPolicy: Always

weed/command/export.go

@@ -215,7 +215,7 @@ func runExport(cmd *Command, args []string) bool {
 
 	err = storage.ScanVolumeFile(util.ResolvePath(*export.dir), *export.collection, vid, storage.NeedleMapInMemory, volumeFileScanner)
 	if err != nil && err != io.EOF {
-		glog.Fatalf("Export Volume File [ERROR] %s\n", err)
+		glog.Errorf("Export Volume File [ERROR] %s\n", err)
 	}
 	return true
 }

weed/command/mount_std.go

@@ -167,6 +167,9 @@ func RunMount(option *MountOptions, umask os.FileMode) bool {
 	if *option.nonempty {
 		options = append(options, fuse.AllowNonEmptyMount())
 	}
+	if *option.readOnly {
+		options = append(options, fuse.ReadOnly())
+	}
 
 	// find mount point
 	mountRoot := filerMountRootPath
@@ -191,7 +194,6 @@ func RunMount(option *MountOptions, umask os.FileMode) bool {
 		CacheDir:           *option.cacheDir,
 		CacheSizeMB:        *option.cacheSizeMB,
 		DataCenter:         *option.dataCenter,
-		EntryCacheTtl:      3 * time.Second,
 		MountUid:           uid,
 		MountGid:           gid,
 		MountMode:          mountMode,
@@ -201,7 +203,6 @@ func RunMount(option *MountOptions, umask os.FileMode) bool {
 		VolumeServerAccess: *mountOptions.volumeServerAccess,
 		Cipher:             cipher,
 		UidGidMapper:       uidGidMapper,
-		ReadOnly:           *option.readOnly,
 	})
 
 	// mount

weed/filer/filer_conf.go

@@ -18,6 +18,7 @@ const (
 	FilerConfName         = "filer.conf"
 	IamConfigDirecotry    = "/etc/iam"
 	IamIdentityFile       = "identity.json"
+	IamPoliciesFile       = "policies.json"
 )
 
 type FilerConf struct {

weed/filer/filerstore_wrapper.go

@@ -149,8 +149,8 @@ func (fsw *FilerStoreWrapper) FindEntry(ctx context.Context, fp util.FullPath) (
 		stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "find").Observe(time.Since(start).Seconds())
 	}()
 
-	glog.V(4).Infof("FindEntry %s", fp)
 	entry, err = actualStore.FindEntry(ctx, fp)
+	glog.V(4).Infof("FindEntry %s: %v", fp, err)
 	if err != nil {
 		return nil, err
 	}

weed/filesys/dir.go

@@ -24,9 +24,11 @@ type Dir struct {
 	wfs    *WFS
 	entry  *filer_pb.Entry
 	parent *Dir
+	id     uint64
 }
 
 var _ = fs.Node(&Dir{})
+//var _ = fs.NodeIdentifier(&Dir{})
 var _ = fs.NodeCreater(&Dir{})
 var _ = fs.NodeMknoder(&Dir{})
 var _ = fs.NodeMkdirer(&Dir{})
@@ -42,6 +44,10 @@ var _ = fs.NodeRemovexattrer(&Dir{})
 var _ = fs.NodeListxattrer(&Dir{})
 var _ = fs.NodeForgetter(&Dir{})
 
+func (dir *Dir) xId() uint64 {
+	return dir.id
+}
+
 func (dir *Dir) Attr(ctx context.Context, attr *fuse.Attr) error {
 
 	// https://github.com/bazil/fuse/issues/196
@@ -53,17 +59,18 @@ func (dir *Dir) Attr(ctx context.Context, attr *fuse.Attr) error {
 		return nil
 	}
 
-	if err := dir.maybeLoadEntry(); err != nil {
+	entry, err := dir.maybeLoadEntry()
+	if err != nil {
 		glog.V(3).Infof("dir Attr %s,err: %+v", dir.FullPath(), err)
 		return err
 	}
 
-	// attr.Inode = util.FullPath(dir.FullPath()).AsInode()
-	attr.Mode = os.FileMode(dir.entry.Attributes.FileMode) | os.ModeDir
-	attr.Mtime = time.Unix(dir.entry.Attributes.Mtime, 0)
-	attr.Crtime = time.Unix(dir.entry.Attributes.Crtime, 0)
-	attr.Gid = dir.entry.Attributes.Gid
-	attr.Uid = dir.entry.Attributes.Uid
+	// attr.Inode = dir.Id()
+	attr.Mode = os.FileMode(entry.Attributes.FileMode) | os.ModeDir
+	attr.Mtime = time.Unix(entry.Attributes.Mtime, 0)
+	attr.Crtime = time.Unix(entry.Attributes.Crtime, 0)
+	attr.Gid = entry.Attributes.Gid
+	attr.Uid = entry.Attributes.Uid
 
 	glog.V(4).Infof("dir Attr %s, attr: %+v", dir.FullPath(), attr)
 
@@ -74,16 +81,18 @@ func (dir *Dir) Getxattr(ctx context.Context, req *fuse.GetxattrRequest, resp *f
 
 	glog.V(4).Infof("dir Getxattr %s", dir.FullPath())
 
-	if err := dir.maybeLoadEntry(); err != nil {
+	entry, err := dir.maybeLoadEntry()
+	if err != nil {
 		return err
 	}
 
-	return getxattr(dir.entry, req, resp)
+	return getxattr(entry, req, resp)
 }
 
 func (dir *Dir) setRootDirAttributes(attr *fuse.Attr) {
 	// attr.Inode = 1 // filer2.FullPath(dir.Path).AsInode()
 	attr.Valid = time.Second
+	attr.Inode = 1 // dir.Id()
 	attr.Uid = dir.wfs.option.MountUid
 	attr.Gid = dir.wfs.option.MountGid
 	attr.Mode = dir.wfs.option.MountMode
@@ -102,36 +111,35 @@ func (dir *Dir) Fsync(ctx context.Context, req *fuse.FsyncRequest) error {
 	return nil
 }
 
-func (dir *Dir) newFile(name string, entry *filer_pb.Entry) fs.Node {
-	f := dir.wfs.fsNodeCache.EnsureFsNode(util.NewFullPath(dir.FullPath(), name), func() fs.Node {
-		return &File{
-			Name:           name,
-			dir:            dir,
-			wfs:            dir.wfs,
-			entry:          entry,
-			entryViewCache: nil,
-		}
-	})
-	f.(*File).dir = dir // in case dir node was created later
-	return f
+func (dir *Dir) newFile(name string) fs.Node {
+
+	fileFullPath := util.NewFullPath(dir.FullPath(), name)
+	fileId := fileFullPath.AsInode()
+	dir.wfs.handlesLock.Lock()
+	existingHandle, found := dir.wfs.handles[fileId]
+	dir.wfs.handlesLock.Unlock()
+
+	if found {
+		glog.V(4).Infof("newFile found opened file handle: %+v", fileFullPath)
+		return existingHandle.f
+	}
+	return &File{
+		Name: name,
+		dir:  dir,
+		wfs:  dir.wfs,
+		id:   fileId,
+	}
 }
 
-func (dir *Dir) newDirectory(fullpath util.FullPath, entry *filer_pb.Entry) fs.Node {
+func (dir *Dir) newDirectory(fullpath util.FullPath) fs.Node {
+
+	return &Dir{name: fullpath.Name(), wfs: dir.wfs, parent: dir, id: fullpath.AsInode()}
 
-	d := dir.wfs.fsNodeCache.EnsureFsNode(fullpath, func() fs.Node {
-		return &Dir{name: entry.Name, wfs: dir.wfs, entry: entry, parent: dir}
-	})
-	d.(*Dir).parent = dir // in case dir node was created later
-	return d
 }
 
 func (dir *Dir) Create(ctx context.Context, req *fuse.CreateRequest,
 	resp *fuse.CreateResponse) (fs.Node, fs.Handle, error) {
 
-	if dir.wfs.option.ReadOnly {
-		return nil, nil, fuse.EPERM
-	}
-
 	request, err := dir.doCreateEntry(req.Name, req.Mode, req.Uid, req.Gid, req.Flags&fuse.OpenExclusive != 0)
 
 	if err != nil {
@@ -139,11 +147,11 @@ func (dir *Dir) Create(ctx context.Context, req *fuse.CreateRequest,
 	}
 	var node fs.Node
 	if request.Entry.IsDirectory {
-		node = dir.newDirectory(util.NewFullPath(dir.FullPath(), req.Name), request.Entry)
+		node = dir.newDirectory(util.NewFullPath(dir.FullPath(), req.Name))
 		return node, nil, nil
 	}
 
-	node = dir.newFile(req.Name, request.Entry)
+	node = dir.newFile(req.Name)
 	file := node.(*File)
 	fh := dir.wfs.AcquireHandle(file, req.Uid, req.Gid)
 	return file, fh, nil
@@ -152,23 +160,20 @@ func (dir *Dir) Create(ctx context.Context, req *fuse.CreateRequest,
 
 func (dir *Dir) Mknod(ctx context.Context, req *fuse.MknodRequest) (fs.Node, error) {
 
-	if dir.wfs.option.ReadOnly {
-		return nil, fuse.EPERM
-	}
-
-	request, err := dir.doCreateEntry(req.Name, req.Mode, req.Uid, req.Gid, false)
+	_, err := dir.doCreateEntry(req.Name, req.Mode, req.Uid, req.Gid, false)
 
 	if err != nil {
 		return nil, err
 	}
 	var node fs.Node
-	node = dir.newFile(req.Name, request.Entry)
+	node = dir.newFile(req.Name)
 	return node, nil
 }
 
-func (dir *Dir) doCreateEntry(name string, mode os.FileMode, uid, gid uint32, exlusive bool) (*filer_pb.CreateEntryRequest, error) {
+func (dir *Dir) doCreateEntry(name string, mode os.FileMode, uid, gid uint32, exclusive bool) (*filer_pb.CreateEntryRequest, error) {
+	dirFullPath := dir.FullPath()
 	request := &filer_pb.CreateEntryRequest{
-		Directory: dir.FullPath(),
+		Directory: dirFullPath,
 		Entry: &filer_pb.Entry{
 			Name:        name,
 			IsDirectory: mode&os.ModeDir > 0,
@@ -183,10 +188,10 @@ func (dir *Dir) doCreateEntry(name string, mode os.FileMode, uid, gid uint32, ex
 				TtlSec:      dir.wfs.option.TtlSec,
 			},
 		},
-		OExcl:      exlusive,
+		OExcl:      exclusive,
 		Signatures: []int32{dir.wfs.signature},
 	}
-	glog.V(1).Infof("create %s/%s", dir.FullPath(), name)
+	glog.V(1).Infof("create %s/%s", dirFullPath, name)
 
 	err := dir.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
 
@@ -197,11 +202,14 @@ func (dir *Dir) doCreateEntry(name string, mode os.FileMode, uid, gid uint32, ex
 			if strings.Contains(err.Error(), "EEXIST") {
 				return fuse.EEXIST
 			}
-			glog.V(0).Infof("create %s/%s: %v", dir.FullPath(), name, err)
+			glog.V(0).Infof("create %s/%s: %v", dirFullPath, name, err)
 			return fuse.EIO
 		}
 
-		dir.wfs.metaCache.InsertEntry(context.Background(), filer.FromPbEntry(request.Directory, request.Entry))
+		if err := dir.wfs.metaCache.InsertEntry(context.Background(), filer.FromPbEntry(request.Directory, request.Entry)); err != nil {
+			glog.Errorf("local InsertEntry dir %s/%s: %v", dirFullPath, name, err)
+			return fuse.EIO
+		}
 
 		return nil
 	})
@@ -210,10 +218,6 @@ func (dir *Dir) doCreateEntry(name string, mode os.FileMode, uid, gid uint32, ex
 
 func (dir *Dir) Mkdir(ctx context.Context, req *fuse.MkdirRequest) (fs.Node, error) {
 
-	if dir.wfs.option.ReadOnly {
-		return nil, fuse.EPERM
-	}
-
 	glog.V(4).Infof("mkdir %s: %s", dir.FullPath(), req.Name)
 
 	newEntry := &filer_pb.Entry{
@@ -228,35 +232,40 @@ func (dir *Dir) Mkdir(ctx context.Context, req *fuse.MkdirRequest) (fs.Node, err
 		},
 	}
 
+	dirFullPath := dir.FullPath()
+
 	err := dir.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
 
 		dir.wfs.mapPbIdFromLocalToFiler(newEntry)
 		defer dir.wfs.mapPbIdFromFilerToLocal(newEntry)
 
 		request := &filer_pb.CreateEntryRequest{
-			Directory:  dir.FullPath(),
+			Directory:  dirFullPath,
 			Entry:      newEntry,
 			Signatures: []int32{dir.wfs.signature},
 		}
 
 		glog.V(1).Infof("mkdir: %v", request)
 		if err := filer_pb.CreateEntry(client, request); err != nil {
-			glog.V(0).Infof("mkdir %s/%s: %v", dir.FullPath(), req.Name, err)
+			glog.V(0).Infof("mkdir %s/%s: %v", dirFullPath, req.Name, err)
 			return err
 		}
 
-		dir.wfs.metaCache.InsertEntry(context.Background(), filer.FromPbEntry(request.Directory, request.Entry))
+		if err := dir.wfs.metaCache.InsertEntry(context.Background(), filer.FromPbEntry(request.Directory, request.Entry)); err != nil {
+			glog.Errorf("local mkdir dir %s/%s: %v", dirFullPath, req.Name, err)
+			return fuse.EIO
+		}
 
 		return nil
 	})
 
 	if err == nil {
-		node := dir.newDirectory(util.NewFullPath(dir.FullPath(), req.Name), newEntry)
+		node := dir.newDirectory(util.NewFullPath(dirFullPath, req.Name))
 
 		return node, nil
 	}
 
-	glog.V(0).Infof("mkdir %s/%s: %v", dir.FullPath(), req.Name, err)
+	glog.V(0).Infof("mkdir %s/%s: %v", dirFullPath, req.Name, err)
 
 	return nil, fuse.EIO
 }
@@ -272,40 +281,40 @@ func (dir *Dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.
 		glog.Errorf("dir Lookup %s: %v", dirPath, visitErr)
 		return nil, fuse.EIO
 	}
-	cachedEntry, cacheErr := dir.wfs.metaCache.FindEntry(context.Background(), fullFilePath)
+	localEntry, cacheErr := dir.wfs.metaCache.FindEntry(context.Background(), fullFilePath)
 	if cacheErr == filer_pb.ErrNotFound {
 		return nil, fuse.ENOENT
 	}
-	entry := cachedEntry.ToProtoEntry()
 
-	if entry == nil {
+	if localEntry == nil {
 		// glog.V(3).Infof("dir Lookup cache miss %s", fullFilePath)
-		entry, err = filer_pb.GetEntry(dir.wfs, fullFilePath)
+		entry, err := filer_pb.GetEntry(dir.wfs, fullFilePath)
 		if err != nil {
 			glog.V(1).Infof("dir GetEntry %s: %v", fullFilePath, err)
 			return nil, fuse.ENOENT
 		}
+		localEntry = filer.FromPbEntry(string(dirPath), entry)
 	} else {
 		glog.V(4).Infof("dir Lookup cache hit %s", fullFilePath)
 	}
 
-	if entry != nil {
-		if entry.IsDirectory {
-			node = dir.newDirectory(fullFilePath, entry)
+	if localEntry != nil {
+		if localEntry.IsDirectory() {
+			node = dir.newDirectory(fullFilePath)
 		} else {
-			node = dir.newFile(req.Name, entry)
+			node = dir.newFile(req.Name)
 		}
 
 		// resp.EntryValid = time.Second
-		// resp.Attr.Inode = fullFilePath.AsInode()
+		resp.Attr.Inode = fullFilePath.AsInode()
 		resp.Attr.Valid = time.Second
-		resp.Attr.Mtime = time.Unix(entry.Attributes.Mtime, 0)
-		resp.Attr.Crtime = time.Unix(entry.Attributes.Crtime, 0)
-		resp.Attr.Mode = os.FileMode(entry.Attributes.FileMode)
-		resp.Attr.Gid = entry.Attributes.Gid
-		resp.Attr.Uid = entry.Attributes.Uid
-		if entry.HardLinkCounter > 0 {
-			resp.Attr.Nlink = uint32(entry.HardLinkCounter)
+		resp.Attr.Mtime = localEntry.Attr.Mtime
+		resp.Attr.Crtime = localEntry.Attr.Crtime
+		resp.Attr.Mode = localEntry.Attr.Mode
+		resp.Attr.Gid = localEntry.Attr.Gid
+		resp.Attr.Uid = localEntry.Attr.Uid
+		if localEntry.HardLinkCounter > 0 {
+			resp.Attr.Nlink = uint32(localEntry.HardLinkCounter)
 		}
 
 		return node, nil
@@ -320,15 +329,14 @@ func (dir *Dir) ReadDirAll(ctx context.Context) (ret []fuse.Dirent, err error) {
 	dirPath := util.FullPath(dir.FullPath())
 	glog.V(4).Infof("dir ReadDirAll %s", dirPath)
 
-	processEachEntryFn := func(entry *filer_pb.Entry, isLast bool) error {
-		if entry.IsDirectory {
-			dirent := fuse.Dirent{Name: entry.Name, Type: fuse.DT_Dir}
+	processEachEntryFn := func(entry *filer.Entry, isLast bool) {
+		if entry.IsDirectory() {
+			dirent := fuse.Dirent{Name: entry.Name(), Type: fuse.DT_Dir, Inode: dirPath.Child(entry.Name()).AsInode()}
 			ret = append(ret, dirent)
 		} else {
-			dirent := fuse.Dirent{Name: entry.Name, Type: findFileType(uint16(entry.Attributes.FileMode))}
+			dirent := fuse.Dirent{Name: entry.Name(), Type: findFileType(uint16(entry.Attr.Mode)), Inode: dirPath.Child(entry.Name()).AsInode()}
 			ret = append(ret, dirent)
 		}
-		return nil
 	}
 
 	if err = meta_cache.EnsureVisited(dir.wfs.metaCache, dir.wfs, dirPath); err != nil {
@@ -336,7 +344,7 @@ func (dir *Dir) ReadDirAll(ctx context.Context) (ret []fuse.Dirent, err error) {
 		return nil, fuse.EIO
 	}
 	listErr := dir.wfs.metaCache.ListDirectoryEntries(context.Background(), dirPath, "", false, int64(math.MaxInt32), func(entry *filer.Entry) bool {
-		processEachEntryFn(entry.ToProtoEntry(), false)
+		processEachEntryFn(entry, false)
 		return true
 	})
 	if listErr != nil {
@@ -368,10 +376,6 @@ func findFileType(mode uint16) fuse.DirentType {
 
 func (dir *Dir) Remove(ctx context.Context, req *fuse.RemoveRequest) error {
 
-	if dir.wfs.option.ReadOnly {
-		return fuse.EPERM
-	}
-
 	if !req.Dir {
 		return dir.removeOneFile(req)
 	}
@@ -382,40 +386,32 @@ func (dir *Dir) Remove(ctx context.Context, req *fuse.RemoveRequest) error {
 
 func (dir *Dir) removeOneFile(req *fuse.RemoveRequest) error {
 
-	filePath := util.NewFullPath(dir.FullPath(), req.Name)
+	dirFullPath := dir.FullPath()
+	filePath := util.NewFullPath(dirFullPath, req.Name)
 	entry, err := filer_pb.GetEntry(dir.wfs, filePath)
 	if err != nil {
 		return err
 	}
-	if entry == nil {
-		return nil
-	}
 
 	// first, ensure the filer store can correctly delete
 	glog.V(3).Infof("remove file: %v", req)
-	isDeleteData := entry.HardLinkCounter <= 1
-	err = filer_pb.Remove(dir.wfs, dir.FullPath(), req.Name, isDeleteData, false, false, false, []int32{dir.wfs.signature})
+	isDeleteData := entry != nil && entry.HardLinkCounter <= 1
+	err = filer_pb.Remove(dir.wfs, dirFullPath, req.Name, isDeleteData, false, false, false, []int32{dir.wfs.signature})
 	if err != nil {
-		glog.V(3).Infof("not found remove file %s/%s: %v", dir.FullPath(), req.Name, err)
+		glog.V(3).Infof("not found remove file %s: %v", filePath, err)
 		return fuse.ENOENT
 	}
 
 	// then, delete meta cache and fsNode cache
-	dir.wfs.metaCache.DeleteEntry(context.Background(), filePath)
-
-	// clear entry inside the file
-	fsNode := dir.wfs.fsNodeCache.GetFsNode(filePath)
-	dir.wfs.fsNodeCache.DeleteFsNode(filePath)
-	if fsNode != nil {
-		if file, ok := fsNode.(*File); ok {
-			file.clearEntry()
-		}
+	if err = dir.wfs.metaCache.DeleteEntry(context.Background(), filePath); err != nil {
+		glog.V(3).Infof("local DeleteEntry %s: %v", filePath, err)
+		return fuse.ESTALE
 	}
 
 	// remove current file handle if any
 	dir.wfs.handlesLock.Lock()
 	defer dir.wfs.handlesLock.Unlock()
-	inodeId := util.NewFullPath(dir.FullPath(), req.Name).AsInode()
+	inodeId := filePath.AsInode()
 	delete(dir.wfs.handles, inodeId)
 
 	return nil
@@ -424,20 +420,20 @@ func (dir *Dir) removeOneFile(req *fuse.RemoveRequest) error {
 
 func (dir *Dir) removeFolder(req *fuse.RemoveRequest) error {
 
+	dirFullPath := dir.FullPath()
 	glog.V(3).Infof("remove directory entry: %v", req)
 	ignoreRecursiveErr := true // ignore recursion error since the OS should manage it
-	err := filer_pb.Remove(dir.wfs, dir.FullPath(), req.Name, true, false, ignoreRecursiveErr, false, []int32{dir.wfs.signature})
+	err := filer_pb.Remove(dir.wfs, dirFullPath, req.Name, true, true, ignoreRecursiveErr, false, []int32{dir.wfs.signature})
 	if err != nil {
-		glog.V(0).Infof("remove %s/%s: %v", dir.FullPath(), req.Name, err)
+		glog.V(0).Infof("remove %s/%s: %v", dirFullPath, req.Name, err)
 		if strings.Contains(err.Error(), "non-empty") {
 			return fuse.EEXIST
 		}
 		return fuse.ENOENT
 	}
 
-	t := util.NewFullPath(dir.FullPath(), req.Name)
+	t := util.NewFullPath(dirFullPath, req.Name)
 	dir.wfs.metaCache.DeleteEntry(context.Background(), t)
-	dir.wfs.fsNodeCache.DeleteFsNode(t)
 
 	return nil
 
@@ -445,73 +441,64 @@ func (dir *Dir) removeFolder(req *fuse.RemoveRequest) error {
 
 func (dir *Dir) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fuse.SetattrResponse) error {
 
-	if dir.wfs.option.ReadOnly {
-		return fuse.EPERM
-	}
-
 	glog.V(4).Infof("%v dir setattr %+v", dir.FullPath(), req)
 
-	if err := dir.maybeLoadEntry(); err != nil {
+	entry, err := dir.maybeLoadEntry()
+	if err != nil {
 		return err
 	}
 
 	if req.Valid.Mode() {
-		dir.entry.Attributes.FileMode = uint32(req.Mode)
+		entry.Attributes.FileMode = uint32(req.Mode)
 	}
 
 	if req.Valid.Uid() {
-		dir.entry.Attributes.Uid = req.Uid
+		entry.Attributes.Uid = req.Uid
 	}
 
 	if req.Valid.Gid() {
-		dir.entry.Attributes.Gid = req.Gid
+		entry.Attributes.Gid = req.Gid
 	}
 
 	if req.Valid.Mtime() {
-		dir.entry.Attributes.Mtime = req.Mtime.Unix()
+		entry.Attributes.Mtime = req.Mtime.Unix()
 	}
 
-	return dir.saveEntry()
+	return dir.saveEntry(entry)
 
 }
 
 func (dir *Dir) Setxattr(ctx context.Context, req *fuse.SetxattrRequest) error {
 
-	if dir.wfs.option.ReadOnly {
-		return fuse.EPERM
-	}
-
 	glog.V(4).Infof("dir Setxattr %s: %s", dir.FullPath(), req.Name)
 
-	if err := dir.maybeLoadEntry(); err != nil {
+	entry, err := dir.maybeLoadEntry()
+	if err != nil {
 		return err
 	}
 
-	if err := setxattr(dir.entry, req); err != nil {
+	if err := setxattr(entry, req); err != nil {
 		return err
 	}
 
-	return dir.saveEntry()
+	return dir.saveEntry(entry)
 
 }
 
 func (dir *Dir) Removexattr(ctx context.Context, req *fuse.RemovexattrRequest) error {
 
-	if dir.wfs.option.ReadOnly {
-		return fuse.EPERM
-	}
-
 	glog.V(4).Infof("dir Removexattr %s: %s", dir.FullPath(), req.Name)
 
-	if err := dir.maybeLoadEntry(); err != nil {
+	entry, err := dir.maybeLoadEntry()
+	if err != nil {
 		return err
 	}
 
-	if err := removexattr(dir.entry, req); err != nil {
+	if err := removexattr(entry, req); err != nil {
 		return err
 	}
 
-	return dir.saveEntry()
+	return dir.saveEntry(entry)
 
 }
 
@@ -519,11 +506,12 @@ func (dir *Dir) Listxattr(ctx context.Context, req *fuse.ListxattrRequest, resp
 
 	glog.V(4).Infof("dir Listxattr %s", dir.FullPath())
 
-	if err := dir.maybeLoadEntry(); err != nil {
+	entry, err := dir.maybeLoadEntry()
+	if err != nil {
 		return err
 	}
 
-	if err := listxattr(dir.entry, req, resp); err != nil {
+	if err := listxattr(entry, req, resp); err != nil {
 		return err
 	}
 
@@ -533,34 +521,25 @@ func (dir *Dir) Listxattr(ctx context.Context, req *fuse.ListxattrRequest, resp
 
 func (dir *Dir) Forget() {
 	glog.V(4).Infof("Forget dir %s", dir.FullPath())
-
-	dir.wfs.fsNodeCache.DeleteFsNode(util.FullPath(dir.FullPath()))
 }
 
-func (dir *Dir) maybeLoadEntry() error {
-	if dir.entry == nil {
-		parentDirPath, name := util.FullPath(dir.FullPath()).DirAndName()
-		entry, err := dir.wfs.maybeLoadEntry(parentDirPath, name)
-		if err != nil {
-			return err
-		}
-		dir.entry = entry
-	}
-	return nil
+func (dir *Dir) maybeLoadEntry() (*filer_pb.Entry, error) {
+	parentDirPath, name := util.FullPath(dir.FullPath()).DirAndName()
+	return dir.wfs.maybeLoadEntry(parentDirPath, name)
 }
 
-func (dir *Dir) saveEntry() error {
+func (dir *Dir) saveEntry(entry *filer_pb.Entry) error {
 
 	parentDir, name := util.FullPath(dir.FullPath()).DirAndName()
 
 	return dir.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
 
-		dir.wfs.mapPbIdFromLocalToFiler(dir.entry)
-		defer dir.wfs.mapPbIdFromFilerToLocal(dir.entry)
+		dir.wfs.mapPbIdFromLocalToFiler(entry)
+		defer dir.wfs.mapPbIdFromFilerToLocal(entry)
 
 		request := &filer_pb.UpdateEntryRequest{
 			Directory:  parentDir,
-			Entry:      dir.entry,
+			Entry:      entry,
 			Signatures: []int32{dir.wfs.signature},
 		}
 
@@ -571,7 +550,10 @@ func (dir *Dir) saveEntry() error {
 			return fuse.EIO
 		}
 
-		dir.wfs.metaCache.UpdateEntry(context.Background(), filer.FromPbEntry(request.Directory, request.Entry))
+		if err := dir.wfs.metaCache.UpdateEntry(context.Background(), filer.FromPbEntry(request.Directory, request.Entry)); err != nil {
+			glog.Errorf("UpdateEntry dir %s/%s: %v", parentDir, name, err)
+			return fuse.ESTALE
+		}
 
 		return nil
 	})

weed/filesys/dir_link.go

@@ -24,10 +24,6 @@ const (
 
 func (dir *Dir) Link(ctx context.Context, req *fuse.LinkRequest, old fs.Node) (fs.Node, error) {
 
-	if dir.wfs.option.ReadOnly {
-		return nil, fuse.EPERM
-	}
-
 	oldFile, ok := old.(*File)
 	if !ok {
 		glog.Errorf("old node is not a file: %+v", old)
@@ -35,11 +31,11 @@ func (dir *Dir) Link(ctx context.Context, req *fuse.LinkRequest, old fs.Node) (f
 
 	glog.V(4).Infof("Link: %v/%v -> %v/%v", oldFile.dir.FullPath(), oldFile.Name, dir.FullPath(), req.NewName)
 
-	if _, err := oldFile.maybeLoadEntry(ctx); err != nil {
+	oldEntry, err := oldFile.maybeLoadEntry(ctx)
+	if err != nil {
 		return nil, err
 	}
 
-	oldEntry := oldFile.getEntry()
 	if oldEntry == nil {
 		return nil, fuse.EIO
 	}
@@ -72,7 +68,7 @@ func (dir *Dir) Link(ctx context.Context, req *fuse.LinkRequest, old fs.Node) (f
 	}
 
 	// apply changes to the filer, and also apply to local metaCache
-	err := dir.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+	err = dir.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
 
 		dir.wfs.mapPbIdFromLocalToFiler(request.Entry)
 		defer dir.wfs.mapPbIdFromFilerToLocal(request.Entry)
@@ -97,11 +93,8 @@ func (dir *Dir) Link(ctx context.Context, req *fuse.LinkRequest, old fs.Node) (f
 	}
 
 	// create new file node
-	newNode := dir.newFile(req.NewName, request.Entry)
+	newNode := dir.newFile(req.NewName)
 	newFile := newNode.(*File)
-	if _, err := newFile.maybeLoadEntry(ctx); err != nil {
-		return nil, err
-	}
 
 	return newFile, err
 
@@ -109,10 +102,6 @@ func (dir *Dir) Link(ctx context.Context, req *fuse.LinkRequest, old fs.Node) (f
 
 func (dir *Dir) Symlink(ctx context.Context, req *fuse.SymlinkRequest) (fs.Node, error) {
 
-	if dir.wfs.option.ReadOnly {
-		return nil, fuse.EPERM
-	}
-
 	glog.V(4).Infof("Symlink: %v/%v to %v", dir.FullPath(), req.NewName, req.Target)
 
 	request := &filer_pb.CreateEntryRequest{
@@ -147,7 +136,7 @@ func (dir *Dir) Symlink(ctx context.Context, req *fuse.SymlinkRequest) (fs.Node,
 		return nil
 	})
 
-	symlink := dir.newFile(req.NewName, request.Entry)
+	symlink := dir.newFile(req.NewName)
 
 	return symlink, err
 

weed/filesys/dir_rename.go

@@ -13,10 +13,6 @@ import (
 
 func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDirectory fs.Node) error {
 
-	if dir.wfs.option.ReadOnly {
-		return fuse.EPERM
-	}
-
 	newDir := newDirectory.(*Dir)
 
 	newPath := util.NewFullPath(newDir.FullPath(), req.NewName)
@@ -68,19 +64,28 @@ func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDirector
 		return fuse.EIO
 	}
 
-	// fmt.Printf("rename path: %v => %v\n", oldPath, newPath)
-	dir.wfs.fsNodeCache.Move(oldPath, newPath)
+	oldFsNode := NodeWithId(oldPath.AsInode())
+	newFsNode := NodeWithId(newPath.AsInode())
+	dir.wfs.Server.InvalidateInternalNode(oldFsNode, newFsNode, func(internalNode fs.Node) {
+		if file, ok := internalNode.(*File); ok {
+			glog.V(4).Infof("internal node %s", file.Name)
+			file.Name = req.NewName
+			file.id = uint64(newFsNode)
+		}
+	})
 
 	// change file handle
 	dir.wfs.handlesLock.Lock()
 	defer dir.wfs.handlesLock.Unlock()
 	inodeId := oldPath.AsInode()
 	existingHandle, found := dir.wfs.handles[inodeId]
+	glog.V(4).Infof("has open filehandle %s: %v", oldPath, found)
 	if !found || existingHandle == nil {
-		return err
+		return nil
 	}
+	glog.V(4).Infof("opened filehandle %s => %s", oldPath, newPath)
 	delete(dir.wfs.handles, inodeId)
 	dir.wfs.handles[newPath.AsInode()] = existingHandle
 
-	return err
+	return nil
 }

weed/filesys/file.go

@@ -2,10 +2,8 @@ package filesys
 
 import (
 	"context"
-	"io"
 	"os"
 	"sort"
-	"sync"
 	"time"
 
 	"github.com/seaweedfs/fuse"
@@ -20,6 +18,7 @@ import (
 const blockSize = 512
 
 var _ = fs.Node(&File{})
+var _ = fs.NodeIdentifier(&File{})
 var _ = fs.NodeOpener(&File{})
 var _ = fs.NodeFsyncer(&File{})
 var _ = fs.NodeSetattrer(&File{})
@@ -30,37 +29,37 @@ var _ = fs.NodeListxattrer(&File{})
 var _ = fs.NodeForgetter(&File{})
 
 type File struct {
-	Name           string
-	dir            *Dir
-	wfs            *WFS
-	entry          *filer_pb.Entry
-	entryLock      sync.RWMutex
-	entryViewCache []filer.VisibleInterval
-	isOpen         int
-	reader         io.ReaderAt
-	dirtyMetadata  bool
+	Name          string
+	dir           *Dir
+	wfs           *WFS
+	entry         *filer_pb.Entry
+	isOpen        int
+	dirtyMetadata bool
+	id            uint64
 }
 
 func (file *File) fullpath() util.FullPath {
 	return util.NewFullPath(file.dir.FullPath(), file.Name)
 }
 
+func (file *File) Id() uint64 {
+	return file.id
+}
+
 func (file *File) Attr(ctx context.Context, attr *fuse.Attr) (err error) {
 
 	glog.V(4).Infof("file Attr %s, open:%v existing:%v", file.fullpath(), file.isOpen, attr)
 
-	entry := file.getEntry()
-	if file.isOpen <= 0 || entry == nil {
-		if entry, err = file.maybeLoadEntry(ctx); err != nil {
-			return err
-		}
+	entry, err := file.maybeLoadEntry(ctx)
+	if err != nil {
+		return err
 	}
 
 	if entry == nil {
 		return fuse.ENOENT
 	}
 
-	// attr.Inode = file.fullpath().AsInode()
+	attr.Inode = file.Id()
 	attr.Valid = time.Second
 	attr.Mode = os.FileMode(entry.Attributes.FileMode)
 	attr.Size = filer.FileSize(entry)
@@ -110,10 +109,6 @@ func (file *File) Open(ctx context.Context, req *fuse.OpenRequest, resp *fuse.Op
 
 func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fuse.SetattrResponse) error {
 
-	if file.wfs.option.ReadOnly {
-		return fuse.EPERM
-	}
-
 	glog.V(4).Infof("%v file setattr %+v", file.fullpath(), req)
 
 	entry, err := file.maybeLoadEntry(ctx)
@@ -122,7 +117,7 @@ func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *f
 	}
 	if file.isOpen > 0 {
 		file.wfs.handlesLock.Lock()
-		fileHandle := file.wfs.handles[file.fullpath().AsInode()]
+		fileHandle := file.wfs.handles[file.Id()]
 		file.wfs.handlesLock.Unlock()
 
 		if fileHandle != nil {
@@ -154,8 +149,6 @@ func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *f
 				}
 			}
 			entry.Chunks = chunks
-			file.entryViewCache, _ = filer.NonOverlappingVisibleIntervals(file.wfs.LookupFn(), chunks)
-			file.setReader(nil)
 		}
 		entry.Attributes.FileSize = req.Size
 		file.dirtyMetadata = true
@@ -204,10 +197,6 @@ func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *f
 
 func (file *File) Setxattr(ctx context.Context, req *fuse.SetxattrRequest) error {
 
-	if file.wfs.option.ReadOnly {
-		return fuse.EPERM
-	}
-
 	glog.V(4).Infof("file Setxattr %s: %s", file.fullpath(), req.Name)
 
 	entry, err := file.maybeLoadEntry(ctx)
@@ -225,10 +214,6 @@ func (file *File) Setxattr(ctx context.Context, req *fuse.SetxattrRequest) error
 
 func (file *File) Removexattr(ctx context.Context, req *fuse.RemovexattrRequest) error {
 
-	if file.wfs.option.ReadOnly {
-		return fuse.EPERM
-	}
-
 	glog.V(4).Infof("file Removexattr %s: %s", file.fullpath(), req.Name)
 
 	entry, err := file.maybeLoadEntry(ctx)
@@ -272,16 +257,19 @@ func (file *File) Fsync(ctx context.Context, req *fuse.FsyncRequest) error {
 func (file *File) Forget() {
 	t := util.NewFullPath(file.dir.FullPath(), file.Name)
 	glog.V(4).Infof("Forget file %s", t)
-	file.wfs.fsNodeCache.DeleteFsNode(t)
 	file.wfs.ReleaseHandle(t, 0)
-	file.setReader(nil)
 }
 
 func (file *File) maybeLoadEntry(ctx context.Context) (entry *filer_pb.Entry, err error) {
-	entry = file.getEntry()
-	if file.isOpen > 0 {
-		return entry, nil
+
+	file.wfs.handlesLock.Lock()
+	handle, found := file.wfs.handles[file.Id()]
+	file.wfs.handlesLock.Unlock()
+	if found {
+		glog.V(4).Infof("maybeLoadEntry found opened file %s/%s: %v %v", file.dir.FullPath(), file.Name, handle.f.entry, entry)
+		entry = handle.f.entry
 	}
+
 	if entry != nil {
 		if len(entry.HardLinkId) == 0 {
 			// only always reload hard link
@@ -294,7 +282,7 @@ func (file *File) maybeLoadEntry(ctx context.Context) (entry *filer_pb.Entry, er
 		return entry, err
 	}
 	if entry != nil {
-		file.setEntry(entry)
+		// file.entry = entry
 	} else {
 		glog.Warningf("maybeLoadEntry not found entry %s/%s: %v", file.dir.FullPath(), file.Name, err)
 	}
@@ -336,44 +324,11 @@ func (file *File) addChunks(chunks []*filer_pb.FileChunk) {
 		return lessThan(chunks[i], chunks[j])
 	})
 
-	// add to entry view cache
-	for _, chunk := range chunks {
-		file.entryViewCache = filer.MergeIntoVisibles(file.entryViewCache, chunk)
-	}
-
-	file.setReader(nil)
-
 	glog.V(4).Infof("%s existing %d chunks adds %d more", file.fullpath(), len(entry.Chunks), len(chunks))
 
 	entry.Chunks = append(entry.Chunks, newChunks...)
 }
 
-func (file *File) setReader(reader io.ReaderAt) {
-	r := file.reader
-	if r != nil {
-		if closer, ok := r.(io.Closer); ok {
-			closer.Close()
-		}
-	}
-	file.reader = reader
-}
-
-func (file *File) setEntry(entry *filer_pb.Entry) {
-	file.entryLock.Lock()
-	defer file.entryLock.Unlock()
-	file.entry = entry
-	file.entryViewCache, _ = filer.NonOverlappingVisibleIntervals(file.wfs.LookupFn(), entry.Chunks)
-	file.setReader(nil)
-}
-
-func (file *File) clearEntry() {
-	file.entryLock.Lock()
-	defer file.entryLock.Unlock()
-	file.entry = nil
-	file.entryViewCache = nil
-	file.setReader(nil)
-}
-
 func (file *File) saveEntry(entry *filer_pb.Entry) error {
 	return file.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
 
@@ -400,7 +355,5 @@ func (file *File) saveEntry(entry *filer_pb.Entry) error {
 }
 
 func (file *File) getEntry() *filer_pb.Entry {
-	file.entryLock.RLock()
-	defer file.entryLock.RUnlock()
 	return file.entry
 }

weed/filesys/filehandle.go

@@ -20,9 +20,11 @@ import (
 
 type FileHandle struct {
 	// cache file has been written to
-	dirtyPages  *ContinuousDirtyPages
-	contentType string
-	handle      uint64
+	dirtyPages     *ContinuousDirtyPages
+	entryViewCache []filer.VisibleInterval
+	reader         io.ReaderAt
+	contentType    string
+	handle         uint64
 	sync.Mutex
 
 	f         *File
@@ -125,20 +127,20 @@ func (fh *FileHandle) readFromChunks(buff []byte, offset int64) (int64, error) {
 	}
 
 	var chunkResolveErr error
-	if fh.f.entryViewCache == nil {
-		fh.f.entryViewCache, chunkResolveErr = filer.NonOverlappingVisibleIntervals(fh.f.wfs.LookupFn(), entry.Chunks)
+	if fh.entryViewCache == nil {
+		fh.entryViewCache, chunkResolveErr = filer.NonOverlappingVisibleIntervals(fh.f.wfs.LookupFn(), entry.Chunks)
 		if chunkResolveErr != nil {
 			return 0, fmt.Errorf("fail to resolve chunk manifest: %v", chunkResolveErr)
 		}
-		fh.f.setReader(nil)
+		fh.reader = nil
 	}
 
-	reader := fh.f.reader
+	reader := fh.reader
 	if reader == nil {
-		chunkViews := filer.ViewFromVisibleIntervals(fh.f.entryViewCache, 0, math.MaxInt64)
+		chunkViews := filer.ViewFromVisibleIntervals(fh.entryViewCache, 0, math.MaxInt64)
 		reader = filer.NewChunkReaderAtFromClient(fh.f.wfs.LookupFn(), chunkViews, fh.f.wfs.chunkCache, fileSize)
 	}
-	fh.f.setReader(reader)
+	fh.reader = reader
 
 	totalRead, err := reader.ReadAt(buff, offset)
 
@@ -154,10 +156,6 @@ func (fh *FileHandle) readFromChunks(buff []byte, offset int64) (int64, error) {
 // Write to the file handle
 func (fh *FileHandle) Write(ctx context.Context, req *fuse.WriteRequest, resp *fuse.WriteResponse) error {
 
-	if fh.f.wfs.option.ReadOnly {
-		return fuse.EPERM
-	}
-
 	fh.Lock()
 	defer fh.Unlock()
 
@@ -195,25 +193,25 @@ func (fh *FileHandle) Write(ctx context.Context, req *fuse.WriteRequest, resp *f
 
 func (fh *FileHandle) Release(ctx context.Context, req *fuse.ReleaseRequest) error {
 
-	glog.V(4).Infof("Release %v fh %d", fh.f.fullpath(), fh.handle)
+	glog.V(4).Infof("Release %v fh %d open=%d", fh.f.fullpath(), fh.handle, fh.f.isOpen)
 
 	fh.Lock()
 	defer fh.Unlock()
 
-	fh.f.entryViewCache = nil
+	fh.f.isOpen--
 
-	if fh.f.isOpen <= 0 {
+	if fh.f.isOpen < 0 {
 		glog.V(0).Infof("Release reset %s open count %d => %d", fh.f.Name, fh.f.isOpen, 0)
 		fh.f.isOpen = 0
 		return nil
 	}
 
-	if fh.f.isOpen == 1 {
-
-		fh.f.isOpen--
+	if fh.f.isOpen == 0 {
+		fh.f.entry = nil
+		fh.entryViewCache = nil
+		fh.reader = nil
 
 		fh.f.wfs.ReleaseHandle(fh.f.fullpath(), fuse.HandleID(fh.handle))
-		fh.f.setReader(nil)
 	}
 
 	return nil

weed/filesys/wfs.go

@@ -41,9 +41,7 @@ type Option struct {
 	CacheDir           string
 	CacheSizeMB        int64
 	DataCenter         string
-	EntryCacheTtl      time.Duration
 	Umask              os.FileMode
-	ReadOnly           bool
 
 	MountUid   uint32
 	MountGid   uint32
@@ -105,24 +103,16 @@ func NewSeaweedFileSystem(option *Option) *WFS {
 	}
 
 	wfs.metaCache = meta_cache.NewMetaCache(path.Join(cacheDir, "meta"), util.FullPath(option.FilerMountRootPath), option.UidGidMapper, func(filePath util.FullPath) {
-		fsNode := wfs.fsNodeCache.GetFsNode(filePath)
-		if fsNode != nil {
-			if file, ok := fsNode.(*File); ok {
-				if err := wfs.Server.InvalidateNodeData(file); err != nil {
-					glog.V(4).Infof("InvalidateNodeData %s : %v", filePath, err)
-				}
-				file.clearEntry()
-			}
+
+		fsNode := NodeWithId(filePath.AsInode())
+		if err := wfs.Server.InvalidateNodeData(fsNode); err != nil {
+			glog.V(4).Infof("InvalidateNodeData %s : %v", filePath, err)
 		}
+
 		dir, name := filePath.DirAndName()
-		parent := wfs.root
-		if dir != "/" {
-			parent = wfs.fsNodeCache.GetFsNode(util.FullPath(dir))
-		}
-		if parent != nil {
-			if err := wfs.Server.InvalidateEntry(parent, name); err != nil {
-				glog.V(4).Infof("InvalidateEntry %s : %v", filePath, err)
-			}
+		parent := NodeWithId(util.FullPath(dir).AsInode())
+		if err := wfs.Server.InvalidateEntry(parent, name); err != nil {
+			glog.V(4).Infof("InvalidateEntry %s : %v", filePath, err)
 		}
 	})
 	startTime := time.Now()
@@ -131,8 +121,7 @@ func NewSeaweedFileSystem(option *Option) *WFS {
 		wfs.metaCache.Shutdown()
 	})
 
-	entry, _ := filer_pb.GetEntry(wfs, util.FullPath(wfs.option.FilerMountRootPath))
-	wfs.root = &Dir{name: wfs.option.FilerMountRootPath, wfs: wfs, entry: entry}
+	wfs.root = &Dir{name: wfs.option.FilerMountRootPath, wfs: wfs}
 	wfs.fsNodeCache = newFsCache(wfs.root)
 
 	if wfs.option.ConcurrentWriters > 0 {
@@ -151,25 +140,28 @@ func (wfs *WFS) AcquireHandle(file *File, uid, gid uint32) (fileHandle *FileHand
 	fullpath := file.fullpath()
 	glog.V(4).Infof("AcquireHandle %s uid=%d gid=%d", fullpath, uid, gid)
 
-	wfs.handlesLock.Lock()
-	defer wfs.handlesLock.Unlock()
+	inodeId := file.Id()
 
-	inodeId := file.fullpath().AsInode()
-	if file.isOpen > 0 {
-		existingHandle, found := wfs.handles[inodeId]
-		if found && existingHandle != nil {
-			file.isOpen++
-			return existingHandle
-		}
+	wfs.handlesLock.Lock()
+	existingHandle, found := wfs.handles[inodeId]
+	wfs.handlesLock.Unlock()
+	if found && existingHandle != nil {
+		existingHandle.f.isOpen++
+		glog.V(4).Infof("Acquired Handle %s open %d", fullpath, existingHandle.f.isOpen)
+		return existingHandle
 	}
 
+	entry, _ := file.maybeLoadEntry(context.Background())
+	file.entry = entry
 	fileHandle = newFileHandle(file, uid, gid)
-	file.maybeLoadEntry(context.Background())
 	file.isOpen++
 
+	wfs.handlesLock.Lock()
 	wfs.handles[inodeId] = fileHandle
+	wfs.handlesLock.Unlock()
 	fileHandle.handle = inodeId
 
+	glog.V(4).Infof("Acquired new Handle %s open %d", fullpath, file.isOpen)
 	return
 }
 
@@ -177,9 +169,9 @@ func (wfs *WFS) ReleaseHandle(fullpath util.FullPath, handleId fuse.HandleID) {
 	wfs.handlesLock.Lock()
 	defer wfs.handlesLock.Unlock()
 
-	glog.V(4).Infof("%s ReleaseHandle id %d current handles length %d", fullpath, handleId, len(wfs.handles))
+	glog.V(4).Infof("ReleaseHandle %s id %d current handles length %d", fullpath, handleId, len(wfs.handles))
 
-	delete(wfs.handles, fullpath.AsInode())
+	delete(wfs.handles, uint64(handleId))
 
 	return
 }
@@ -269,3 +261,11 @@ func (wfs *WFS) LookupFn() wdclient.LookupFileIdFunctionType {
 	return filer.LookupFn(wfs)
 
 }
+
+type NodeWithId uint64
+func (n NodeWithId) Id() uint64 {
+	return uint64(n)
+}
+func (n NodeWithId) Attr(ctx context.Context, attr *fuse.Attr) error {
+	return nil
+}

weed/glog/glog.go

@@ -398,7 +398,7 @@ type flushSyncWriter interface {
 func init() {
 	flag.BoolVar(&logging.toStderr, "logtostderr", false, "log to standard error instead of files")
 	flag.BoolVar(&logging.alsoToStderr, "alsologtostderr", true, "log to standard error as well as files")
-	flag.Var(&logging.verbosity, "v", "log level for V logs")
+	flag.Var(&logging.verbosity, "v", "log levels [0|1|2|3|4], default to 0")
 	flag.Var(&logging.stderrThreshold, "stderrthreshold", "logs at or above this threshold go to stderr")
 	flag.Var(&logging.vmodule, "vmodule", "comma-separated list of pattern=N settings for file-filtered logging")
 	flag.Var(&logging.traceLocation, "log_backtrace_at", "when logging hits line file:N, emit a stack trace")

weed/iamapi/iamapi_handlers.go

@@ -50,20 +50,25 @@ func writeErrorResponse(w http.ResponseWriter, errorCode s3err.ErrorCode, reqURL
 	writeResponse(w, apiError.HTTPStatusCode, encodedErrorResponse, mimeXML)
 }
 
-func writeIamErrorResponse(w http.ResponseWriter, err error, object string, value string) {
+func writeIamErrorResponse(w http.ResponseWriter, err error, object string, value string, msg error) {
 	errCode := err.Error()
 	errorResp := ErrorResponse{}
 	errorResp.Error.Type = "Sender"
 	errorResp.Error.Code = &errCode
+	if msg != nil {
+		errMsg := msg.Error()
+		errorResp.Error.Message = &errMsg
+	}
 	glog.Errorf("Response %+v", err)
 	switch errCode {
 	case iam.ErrCodeNoSuchEntityException:
 		msg := fmt.Sprintf("The %s with name %s cannot be found.", object, value)
 		errorResp.Error.Message = &msg
 		writeResponse(w, http.StatusNotFound, encodeResponse(errorResp), mimeXML)
+	case iam.ErrCodeServiceFailureException:
+		writeResponse(w, http.StatusInternalServerError, encodeResponse(errorResp), mimeXML)
 	default:
 		writeResponse(w, http.StatusInternalServerError, encodeResponse(errorResp), mimeXML)
-
 	}
 }
 

weed/iamapi/iamapi_management_handlers.go

@@ -11,6 +11,7 @@ import (
 	"math/rand"
 	"net/http"
 	"net/url"
+	"reflect"
 	"strings"
 	"sync"
 	"time"
@@ -19,27 +20,77 @@ import (
 )
 
 const (
-	charsetUpper = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
-	charset      = charsetUpper + "abcdefghijklmnopqrstuvwxyz/"
+	charsetUpper           = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
+	charset                = charsetUpper + "abcdefghijklmnopqrstuvwxyz/"
+	policyDocumentVersion  = "2012-10-17"
+	StatementActionAdmin   = "*"
+	StatementActionWrite   = "Put*"
+	StatementActionRead    = "Get*"
+	StatementActionList    = "List*"
+	StatementActionTagging = "Tagging*"
 )
 
 var (
 	seededRand *rand.Rand = rand.New(
 		rand.NewSource(time.Now().UnixNano()))
 	policyDocuments = map[string]*PolicyDocument{}
+	policyLock      = sync.RWMutex{}
 )
 
+func MapToStatementAction(action string) string {
+	switch action {
+	case StatementActionAdmin:
+		return s3_constants.ACTION_ADMIN
+	case StatementActionWrite:
+		return s3_constants.ACTION_WRITE
+	case StatementActionRead:
+		return s3_constants.ACTION_READ
+	case StatementActionList:
+		return s3_constants.ACTION_LIST
+	case StatementActionTagging:
+		return s3_constants.ACTION_TAGGING
+	default:
+		return ""
+	}
+}
+
+func MapToIdentitiesAction(action string) string {
+	switch action {
+	case s3_constants.ACTION_ADMIN:
+		return StatementActionAdmin
+	case s3_constants.ACTION_WRITE:
+		return StatementActionWrite
+	case s3_constants.ACTION_READ:
+		return StatementActionRead
+	case s3_constants.ACTION_LIST:
+		return StatementActionList
+	case s3_constants.ACTION_TAGGING:
+		return StatementActionTagging
+	default:
+		return ""
+	}
+}
+
 type Statement struct {
 	Effect   string   `json:"Effect"`
 	Action   []string `json:"Action"`
 	Resource []string `json:"Resource"`
 }
 
+type Policies struct {
+	Policies map[string]PolicyDocument `json:"policies"`
+}
+
 type PolicyDocument struct {
 	Version   string       `json:"Version"`
 	Statement []*Statement `json:"Statement"`
 }
 
+func (p PolicyDocument) String() string {
+	b, _ := json.Marshal(p)
+	return string(b)
+}
+
 func Hash(s *string) string {
 	h := sha1.New()
 	h.Write([]byte(*s))
@@ -83,7 +134,7 @@ func (iama *IamApiServer) CreateUser(s3cfg *iam_pb.S3ApiConfiguration, values ur
 func (iama *IamApiServer) DeleteUser(s3cfg *iam_pb.S3ApiConfiguration, userName string) (resp DeleteUserResponse, err error) {
 	for i, ident := range s3cfg.Identities {
 		if userName == ident.Name {
-			ident.Credentials = append(ident.Credentials[:i], ident.Credentials[i+1:]...)
+			s3cfg.Identities = append(s3cfg.Identities[:i], s3cfg.Identities[i+1:]...)
 			return resp, nil
 		}
 	}
@@ -119,7 +170,16 @@ func (iama *IamApiServer) CreatePolicy(s3cfg *iam_pb.S3ApiConfiguration, values
 	resp.CreatePolicyResult.Policy.PolicyName = &policyName
 	resp.CreatePolicyResult.Policy.Arn = &arn
 	resp.CreatePolicyResult.Policy.PolicyId = &policyId
-	policyDocuments[policyName] = &policyDocument
+	policies := Policies{}
+	policyLock.Lock()
+	defer policyLock.Unlock()
+	if err = iama.s3ApiConfig.GetPolicies(&policies); err != nil {
+		return resp, err
+	}
+	policies.Policies[policyName] = policyDocument
+	if err = iama.s3ApiConfig.PutPolicies(&policies); err != nil {
+		return resp, err
+	}
 	return resp, nil
 }
 
@@ -144,19 +204,69 @@ func (iama *IamApiServer) PutUserPolicy(s3cfg *iam_pb.S3ApiConfiguration, values
 	return resp, nil
 }
 
-func MapAction(action string) string {
-	switch action {
-	case "*":
-		return s3_constants.ACTION_ADMIN
-	case "Put*":
-		return s3_constants.ACTION_WRITE
-	case "Get*":
-		return s3_constants.ACTION_READ
-	case "List*":
-		return s3_constants.ACTION_LIST
-	default:
-		return s3_constants.ACTION_TAGGING
+func (iama *IamApiServer) GetUserPolicy(s3cfg *iam_pb.S3ApiConfiguration, values url.Values) (resp GetUserPolicyResponse, err error) {
+	userName := values.Get("UserName")
+	policyName := values.Get("PolicyName")
+	for _, ident := range s3cfg.Identities {
+		if userName != ident.Name {
+			continue
+		}
+
+		resp.GetUserPolicyResult.UserName = userName
+		resp.GetUserPolicyResult.PolicyName = policyName
+		if len(ident.Actions) == 0 {
+			return resp, fmt.Errorf(iam.ErrCodeNoSuchEntityException)
+		}
+
+		policyDocument := PolicyDocument{Version: policyDocumentVersion}
+		statements := make(map[string][]string)
+		for _, action := range ident.Actions {
+			// parse "Read:EXAMPLE-BUCKET"
+			act := strings.Split(action, ":")
+
+			resource := "*"
+			if len(act) == 2 {
+				resource = fmt.Sprintf("arn:aws:s3:::%s/*", act[1])
+			}
+			statements[resource] = append(statements[resource],
+				fmt.Sprintf("s3:%s", MapToIdentitiesAction(act[0])),
+			)
+		}
+		for resource, actions := range statements {
+			isEqAction := false
+			for i, statement := range policyDocument.Statement {
+				if reflect.DeepEqual(statement.Action, actions) {
+					policyDocument.Statement[i].Resource = append(
+						policyDocument.Statement[i].Resource, resource)
+					isEqAction = true
+					break
+				}
+			}
+			if isEqAction {
+				continue
+			}
+			policyDocumentStatement := Statement{
+				Effect: "Allow",
+				Action: actions,
+			}
+			policyDocumentStatement.Resource = append(policyDocumentStatement.Resource, resource)
+			policyDocument.Statement = append(policyDocument.Statement, &policyDocumentStatement)
+		}
+		resp.GetUserPolicyResult.PolicyDocument = policyDocument.String()
+		return resp, nil
 	}
+	return resp, fmt.Errorf(iam.ErrCodeNoSuchEntityException)
+}
+
+func (iama *IamApiServer) DeleteUserPolicy(s3cfg *iam_pb.S3ApiConfiguration, values url.Values) (resp PutUserPolicyResponse, err error) {
+	userName := values.Get("UserName")
+	for i, ident := range s3cfg.Identities {
+		if ident.Name == userName {
+			s3cfg.Identities = append(s3cfg.Identities[:i], s3cfg.Identities[i+1:]...)
+			return resp, nil
+		}
+	}
+	return resp, fmt.Errorf(iam.ErrCodeNoSuchEntityException)
 }
 
 func GetActions(policy *PolicyDocument) (actions []string) {
@@ -178,8 +288,9 @@ func GetActions(policy *PolicyDocument) (actions []string) {
 					glog.Infof("not match action: %s", act)
 					continue
 				}
+				statementAction := MapToStatementAction(act[1])
 				if res[5] == "*" {
-					actions = append(actions, MapAction(act[1]))
+					actions = append(actions, statementAction)
 					continue
 				}
 				// Parse my-bucket/shared/*
@@ -188,7 +299,7 @@ func GetActions(policy *PolicyDocument) (actions []string) {
 					glog.Infof("not match bucket: %s", path)
 					continue
 				}
-				actions = append(actions, fmt.Sprintf("%s:%s", MapAction(act[1]), path[0]))
+				actions = append(actions, fmt.Sprintf("%s:%s", statementAction, path[0]))
 			}
 		}
 	}
@@ -277,14 +388,15 @@ func (iama *IamApiServer) DoActions(w http.ResponseWriter, r *http.Request) {
 		userName := values.Get("UserName")
 		response, err = iama.GetUser(s3cfg, userName)
 		if err != nil {
-			writeIamErrorResponse(w, err, "user", userName)
+			writeIamErrorResponse(w, err, "user", userName, nil)
 			return
 		}
+		changed = false
 	case "DeleteUser":
 		userName := values.Get("UserName")
 		response, err = iama.DeleteUser(s3cfg, userName)
 		if err != nil {
-			writeIamErrorResponse(w, err, "user", userName)
+			writeIamErrorResponse(w, err, "user", userName, nil)
 			return
 		}
 	case "CreateAccessKey":
@@ -305,8 +417,23 @@ func (iama *IamApiServer) DoActions(w http.ResponseWriter, r *http.Request) {
 			writeErrorResponse(w, s3err.ErrInvalidRequest, r.URL)
 			return
 		}
+	case "GetUserPolicy":
+		response, err = iama.GetUserPolicy(s3cfg, values)
+		if err != nil {
+			writeIamErrorResponse(w, err, "user", values.Get("UserName"), nil)
+			return
+		}
+		changed = false
+	case "DeleteUserPolicy":
+		if response, err = iama.DeleteUserPolicy(s3cfg, values); err != nil {
+			writeIamErrorResponse(w, err, "user", values.Get("UserName"), nil)
+		}
 	default:
-		writeErrorResponse(w, s3err.ErrNotImplemented, r.URL)
+		errNotImplemented := s3err.GetAPIError(s3err.ErrNotImplemented)
+		errorResponse := ErrorResponse{}
+		errorResponse.Error.Code = &errNotImplemented.Code
+		errorResponse.Error.Message = &errNotImplemented.Description
+		writeResponse(w, errNotImplemented.HTTPStatusCode, encodeResponse(errorResponse), mimeXML)
 		return
 	}
 	if changed {
@@ -314,7 +441,7 @@ func (iama *IamApiServer) DoActions(w http.ResponseWriter, r *http.Request) {
 		err := iama.s3ApiConfig.PutS3ApiConfiguration(s3cfg)
 		s3cfgLock.Unlock()
 		if err != nil {
-			writeErrorResponse(w, s3err.ErrInternalError, r.URL)
+			writeIamErrorResponse(w, fmt.Errorf(iam.ErrCodeServiceFailureException), "", "", err)
 			return
 		}
 	}

weed/iamapi/iamapi_response.go

@@ -79,6 +79,16 @@ type PutUserPolicyResponse struct {
 	XMLName xml.Name `xml:"https://iam.amazonaws.com/doc/2010-05-08/ PutUserPolicyResponse"`
 }
 
+type GetUserPolicyResponse struct {
+	CommonResponse
+	XMLName             xml.Name `xml:"https://iam.amazonaws.com/doc/2010-05-08/ GetUserPolicyResponse"`
+	GetUserPolicyResult struct {
+		UserName       string `xml:"UserName"`
+		PolicyName     string `xml:"PolicyName"`
+		PolicyDocument string `xml:"PolicyDocument"`
+	} `xml:"GetUserPolicyResult"`
+}
+
 type ErrorResponse struct {
 	CommonResponse
 	XMLName xml.Name `xml:"https://iam.amazonaws.com/doc/2010-05-08/ ErrorResponse"`

weed/iamapi/iamapi_server.go

@@ -4,11 +4,14 @@ package iamapi
 
 import (
 	"bytes"
+	"encoding/json"
 	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/iam_pb"
+	"github.com/chrislusf/seaweedfs/weed/s3api"
+	. "github.com/chrislusf/seaweedfs/weed/s3api/s3_constants"
 	"github.com/chrislusf/seaweedfs/weed/wdclient"
 	"github.com/gorilla/mux"
 	"google.golang.org/grpc"
@@ -19,6 +22,8 @@ import (
 type IamS3ApiConfig interface {
 	GetS3ApiConfiguration(s3cfg *iam_pb.S3ApiConfiguration) (err error)
 	PutS3ApiConfiguration(s3cfg *iam_pb.S3ApiConfiguration) (err error)
+	GetPolicies(policies *Policies) (err error)
+	PutPolicies(policies *Policies) (err error)
 }
 
 type IamS3ApiConfigure struct {
@@ -36,7 +41,7 @@ type IamServerOption struct {
 
 type IamApiServer struct {
 	s3ApiConfig IamS3ApiConfig
-	filerclient *filer_pb.SeaweedFilerClient
+	iam         *s3api.IdentityAccessManagement
 }
 
 var s3ApiConfigure IamS3ApiConfig
@@ -46,9 +51,10 @@ func NewIamApiServer(router *mux.Router, option *IamServerOption) (iamApiServer
 		option:       option,
 		masterClient: wdclient.NewMasterClient(option.GrpcDialOption, pb.AdminShellClient, "", 0, "", strings.Split(option.Masters, ",")),
 	}
-
+	s3Option := s3api.S3ApiServerOption{Filer: option.Filer}
 	iamApiServer = &IamApiServer{
 		s3ApiConfig: s3ApiConfigure,
+		iam:         s3api.NewIdentityAccessManagement(&s3Option),
 	}
 
 	iamApiServer.registerRouter(router)
@@ -62,7 +68,8 @@ func (iama *IamApiServer) registerRouter(router *mux.Router) {
 	// ListBuckets
 
 	// apiRouter.Methods("GET").Path("/").HandlerFunc(track(s3a.iam.Auth(s3a.ListBucketsHandler, ACTION_ADMIN), "LIST"))
-	apiRouter.Path("/").Methods("POST").HandlerFunc(iama.DoActions)
+	apiRouter.Methods("POST").Path("/").HandlerFunc(iama.iam.Auth(iama.DoActions, ACTION_ADMIN))
+	//
 	// NotFound
 	apiRouter.NotFoundHandler = http.HandlerFunc(notFoundHandler)
 }
@@ -102,3 +109,41 @@ func (iam IamS3ApiConfigure) PutS3ApiConfiguration(s3cfg *iam_pb.S3ApiConfigurat
 		},
 	)
 }
+
+func (iam IamS3ApiConfigure) GetPolicies(policies *Policies) (err error) {
+	var buf bytes.Buffer
+	err = pb.WithGrpcFilerClient(iam.option.FilerGrpcAddress, iam.option.GrpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
+		if err = filer.ReadEntry(iam.masterClient, client, filer.IamConfigDirecotry, filer.IamPoliciesFile, &buf); err != nil {
+			return err
+		}
+		return nil
+	})
+	if err != nil {
+		return err
+	}
+	if buf.Len() == 0 {
+		policies.Policies = make(map[string]PolicyDocument)
+		return nil
+	}
+	if err := json.Unmarshal(buf.Bytes(), policies); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (iam IamS3ApiConfigure) PutPolicies(policies *Policies) (err error) {
+	var b []byte
+	if b, err = json.Marshal(policies); err != nil {
+		return err
+	}
+	return pb.WithGrpcFilerClient(
+		iam.option.FilerGrpcAddress,
+		iam.option.GrpcDialOption,
+		func(client filer_pb.SeaweedFilerClient) error {
+			if err := filer.SaveInsideFiler(client, filer.IamConfigDirecotry, filer.IamPoliciesFile, b); err != nil {
+				return err
+			}
+			return nil
+		},
+	)
+}

weed/iamapi/iamapi_test.go

@@ -7,29 +7,43 @@ import (
 	"github.com/aws/aws-sdk-go/service/iam"
 	"github.com/chrislusf/seaweedfs/weed/pb/iam_pb"
 	"github.com/gorilla/mux"
+	"github.com/jinzhu/copier"
 	"github.com/stretchr/testify/assert"
 	"net/http"
 	"net/http/httptest"
 	"testing"
 )
 
-var S3config iam_pb.S3ApiConfiguration
 var GetS3ApiConfiguration func(s3cfg *iam_pb.S3ApiConfiguration) (err error)
 var PutS3ApiConfiguration func(s3cfg *iam_pb.S3ApiConfiguration) (err error)
+var GetPolicies func(policies *Policies) (err error)
+var PutPolicies func(policies *Policies) (err error)
+
+var s3config = iam_pb.S3ApiConfiguration{}
+var policiesFile = Policies{Policies: make(map[string]PolicyDocument)}
+var ias = IamApiServer{s3ApiConfig: iamS3ApiConfigureMock{}}
 
 type iamS3ApiConfigureMock struct{}
 
 func (iam iamS3ApiConfigureMock) GetS3ApiConfiguration(s3cfg *iam_pb.S3ApiConfiguration) (err error) {
-	s3cfg = &S3config
+	_ = copier.Copy(&s3cfg.Identities, &s3config.Identities)
 	return nil
 }
 
 func (iam iamS3ApiConfigureMock) PutS3ApiConfiguration(s3cfg *iam_pb.S3ApiConfiguration) (err error) {
-	S3config = *s3cfg
+	_ = copier.Copy(&s3config.Identities, &s3cfg.Identities)
 	return nil
 }
 
-var a = IamApiServer{}
+func (iam iamS3ApiConfigureMock) GetPolicies(policies *Policies) (err error) {
+	_ = copier.Copy(&policies, &policiesFile)
+	return nil
+}
+
+func (iam iamS3ApiConfigureMock) PutPolicies(policies *Policies) (err error) {
+	_ = copier.Copy(&policiesFile, &policies)
+	return nil
+}
 
 func TestCreateUser(t *testing.T) {
 	userName := aws.String("Test")
@@ -64,17 +78,6 @@ func TestListAccessKeys(t *testing.T) {
 	assert.Equal(t, http.StatusOK, response.Code)
 }
 
-func TestDeleteUser(t *testing.T) {
-	userName := aws.String("Test")
-	params := &iam.DeleteUserInput{UserName: userName}
-	req, _ := iam.New(session.New()).DeleteUserRequest(params)
-	_ = req.Build()
-	out := DeleteUserResponse{}
-	response, err := executeRequest(req.HTTPRequest, out)
-	assert.Equal(t, nil, err)
-	assert.Equal(t, http.StatusNotFound, response.Code)
-}
-
 func TestGetUser(t *testing.T) {
 	userName := aws.String("Test")
 	params := &iam.GetUserInput{UserName: userName}
@@ -83,7 +86,7 @@ func TestGetUser(t *testing.T) {
 	out := GetUserResponse{}
 	response, err := executeRequest(req.HTTPRequest, out)
 	assert.Equal(t, nil, err)
-	assert.Equal(t, http.StatusNotFound, response.Code)
+	assert.Equal(t, http.StatusOK, response.Code)
 }
 
 // Todo flat statement
@@ -147,11 +150,32 @@ func TestPutUserPolicy(t *testing.T) {
 	assert.Equal(t, http.StatusOK, response.Code)
 }
 
+func TestGetUserPolicy(t *testing.T) {
+	userName := aws.String("Test")
+	params := &iam.GetUserPolicyInput{UserName: userName, PolicyName: aws.String("S3-read-only-example-bucket")}
+	req, _ := iam.New(session.New()).GetUserPolicyRequest(params)
+	_ = req.Build()
+	out := GetUserPolicyResponse{}
+	response, err := executeRequest(req.HTTPRequest, out)
+	assert.Equal(t, nil, err)
+	assert.Equal(t, http.StatusOK, response.Code)
+}
+
+func TestDeleteUser(t *testing.T) {
+	userName := aws.String("Test")
+	params := &iam.DeleteUserInput{UserName: userName}
+	req, _ := iam.New(session.New()).DeleteUserRequest(params)
+	_ = req.Build()
+	out := DeleteUserResponse{}
+	response, err := executeRequest(req.HTTPRequest, out)
+	assert.Equal(t, nil, err)
+	assert.Equal(t, http.StatusOK, response.Code)
+}
+
 func executeRequest(req *http.Request, v interface{}) (*httptest.ResponseRecorder, error) {
 	rr := httptest.NewRecorder()
 	apiRouter := mux.NewRouter().SkipClean(true)
-	a.s3ApiConfig = iamS3ApiConfigureMock{}
-	apiRouter.Path("/").Methods("POST").HandlerFunc(a.DoActions)
+	apiRouter.Path("/").Methods("POST").HandlerFunc(ias.DoActions)
 	apiRouter.ServeHTTP(rr, req)
 	return rr, xml.Unmarshal(rr.Body.Bytes(), &v)
 }

weed/operation/assign_file_id.go

@@ -86,6 +86,7 @@ func Assign(masterFn GetMasterFn, grpcDialOption grpc.DialOption, primaryRequest
 			continue
 		}
 
+		break
 	}
 
 	return ret, lastError

weed/operation/upload_content.go

@@ -11,7 +11,6 @@ import (
 	"net/http"
 	"net/textproto"
 	"path/filepath"
-	"runtime/debug"
 	"strings"
 	"time"
 
@@ -235,10 +234,12 @@ func upload_content(uploadUrl string, fillBufferFunction func(w io.Writer) error
 	// print("+")
 	resp, post_err := HttpClient.Do(req)
 	if post_err != nil {
-		if !strings.Contains(post_err.Error(), "connection reset by peer") {
-			glog.Errorf("upload %s %d bytes to %v: %v", filename, originalDataSize, uploadUrl, post_err)
-			debug.PrintStack()
+		if strings.Contains(post_err.Error(), "connection reset by peer") ||
+			strings.Contains(post_err.Error(), "use of closed network connection") {
+			resp, post_err = HttpClient.Do(req)
 		}
+	}
+	if post_err != nil {
 		return nil, fmt.Errorf("upload %s %d bytes to %v: %v", filename, originalDataSize, uploadUrl, post_err)
 	}
 	// print("-")

weed/s3api/auth_signature_v4.go

@@ -24,6 +24,7 @@ import (
 	"crypto/subtle"
 	"encoding/hex"
 	"github.com/chrislusf/seaweedfs/weed/s3api/s3err"
+	"io/ioutil"
 	"net/http"
 	"net/url"
 	"regexp"
@@ -132,6 +133,17 @@ func (iam *IdentityAccessManagement) doesSignatureMatch(hashedPayload string, r
 	// Query string.
 	queryStr := req.URL.Query().Encode()
 
+	// Get hashed Payload
+	if signV4Values.Credential.scope.service != "s3" && hashedPayload == emptySHA256 && r.Body != nil {
+		buf, _ := ioutil.ReadAll(r.Body)
+		r.Body = ioutil.NopCloser(bytes.NewBuffer(buf))
+		b, _ := ioutil.ReadAll(bytes.NewBuffer(buf))
+		if len(b) != 0 {
+			bodyHash := sha256.Sum256(b)
+			hashedPayload = hex.EncodeToString(bodyHash[:])
+		}
+	}
+
 	// Get canonical request.
 	canonicalRequest := getCanonicalRequest(extractedSignedHeaders, hashedPayload, queryStr, req.URL.Path, req.Method)
 
@@ -139,7 +151,10 @@ func (iam *IdentityAccessManagement) doesSignatureMatch(hashedPayload string, r
 	stringToSign := getStringToSign(canonicalRequest, t, signV4Values.Credential.getScope())
 
 	// Get hmac signing key.
-	signingKey := getSigningKey(cred.SecretKey, signV4Values.Credential.scope.date, signV4Values.Credential.scope.region)
+	signingKey := getSigningKey(cred.SecretKey,
+		signV4Values.Credential.scope.date,
+		signV4Values.Credential.scope.region,
+		signV4Values.Credential.scope.service)
 
 	// Calculate signature.
 	newSignature := getSignature(signingKey, stringToSign)
@@ -310,7 +325,7 @@ func (iam *IdentityAccessManagement) doesPolicySignatureV4Match(formValues http.
 	}
 
 	// Get signing key.
-	signingKey := getSigningKey(cred.SecretKey, credHeader.scope.date, credHeader.scope.region)
+	signingKey := getSigningKey(cred.SecretKey, credHeader.scope.date, credHeader.scope.region, credHeader.scope.service)
 
 	// Get signature.
 	newSignature := getSignature(signingKey, formValues.Get("Policy"))
@@ -427,7 +442,10 @@ func (iam *IdentityAccessManagement) doesPresignedSignatureMatch(hashedPayload s
 	presignedStringToSign := getStringToSign(presignedCanonicalReq, t, pSignValues.Credential.getScope())
 
 	// Get hmac presigned signing key.
-	presignedSigningKey := getSigningKey(cred.SecretKey, pSignValues.Credential.scope.date, pSignValues.Credential.scope.region)
+	presignedSigningKey := getSigningKey(cred.SecretKey,
+		pSignValues.Credential.scope.date,
+		pSignValues.Credential.scope.region,
+		pSignValues.Credential.scope.service)
 
 	// Get new signature.
 	newSignature := getSignature(presignedSigningKey, presignedStringToSign)
@@ -655,11 +673,11 @@ func sumHMAC(key []byte, data []byte) []byte {
 }
 
 // getSigningKey hmac seed to calculate final signature.
-func getSigningKey(secretKey string, t time.Time, region string) []byte {
+func getSigningKey(secretKey string, t time.Time, region string, service string) []byte {
 	date := sumHMAC([]byte("AWS4"+secretKey), []byte(t.Format(yyyymmdd)))
 	regionBytes := sumHMAC(date, []byte(region))
-	service := sumHMAC(regionBytes, []byte("s3"))
-	signingKey := sumHMAC(service, []byte("aws4_request"))
+	serviceBytes := sumHMAC(regionBytes, []byte(service))
+	signingKey := sumHMAC(serviceBytes, []byte("aws4_request"))
 	return signingKey
 }
 

weed/s3api/auto_signature_v4_test.go

@@ -370,7 +370,7 @@ func preSignV4(req *http.Request, accessKeyID, secretAccessKey string, expires i
 	queryStr := strings.Replace(query.Encode(), "+", "%20", -1)
 	canonicalRequest := getCanonicalRequest(extractedSignedHeaders, unsignedPayload, queryStr, req.URL.Path, req.Method)
 	stringToSign := getStringToSign(canonicalRequest, date, scope)
-	signingKey := getSigningKey(secretAccessKey, date, region)
+	signingKey := getSigningKey(secretAccessKey, date, region, "s3")
 	signature := getSignature(signingKey, stringToSign)
 
 	req.URL.RawQuery = query.Encode()

weed/s3api/chunked_reader_v4.go

@@ -45,7 +45,7 @@ func getChunkSignature(secretKey string, seedSignature string, region string, da
 		hashedChunk
 
 	// Get hmac signing key.
-	signingKey := getSigningKey(secretKey, date, region)
+	signingKey := getSigningKey(secretKey, date, region, "s3")
 
 	// Calculate signature.
 	newSignature := getSignature(signingKey, stringToSign)
@@ -117,7 +117,7 @@ func (iam *IdentityAccessManagement) calculateSeedSignature(r *http.Request) (cr
 	stringToSign := getStringToSign(canonicalRequest, date, signV4Values.Credential.getScope())
 
 	// Get hmac signing key.
-	signingKey := getSigningKey(cred.SecretKey, signV4Values.Credential.scope.date, region)
+	signingKey := getSigningKey(cred.SecretKey, signV4Values.Credential.scope.date, region, "s3")
 
 	// Calculate signature.
 	newSignature := getSignature(signingKey, stringToSign)

weed/s3api/s3api_object_handlers.go

@@ -311,7 +311,7 @@ func (s3a *S3ApiServer) proxyToFiler(w http.ResponseWriter, r *http.Request, des
 	}
 	defer util.CloseResponse(resp)
 
-	if resp.ContentLength == -1 || resp.StatusCode == 404 {
+	if (resp.ContentLength == -1 || resp.StatusCode == 404) && resp.StatusCode != 304 {
 		if r.Method != "DELETE" {
 			writeErrorResponse(w, s3err.ErrNoSuchKey, r.URL)
 			return

weed/server/filer_server_handlers_read.go

@@ -79,7 +79,7 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request,
 		w.Header().Set("Last-Modified", entry.Attr.Mtime.UTC().Format(http.TimeFormat))
 		if r.Header.Get("If-Modified-Since") != "" {
 			if t, parseError := time.Parse(http.TimeFormat, r.Header.Get("If-Modified-Since")); parseError == nil {
-				if t.After(entry.Attr.Mtime) {
+				if !t.Before(entry.Attr.Mtime) {
 					w.WriteHeader(http.StatusNotModified)
 					return
 				}

weed/server/filer_server_handlers_write_autochunk.go

@@ -142,6 +142,14 @@ func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileNa
 		if fileName != "" {
 			path += fileName
 		}
+	} else {
+		if fileName != "" {
+			if possibleDirEntry, findDirErr := fs.filer.FindEntry(ctx, util.FullPath(path)); findDirErr == nil {
+				if possibleDirEntry.IsDirectory() {
+					path += "/" + fileName
+				}
+			}
+		}
 	}
 
 	var entry *filer.Entry

weed/server/volume_server_handlers_write.go

@@ -13,7 +13,6 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/stats"
 	"github.com/chrislusf/seaweedfs/weed/storage/needle"
 	"github.com/chrislusf/seaweedfs/weed/topology"
-	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
 func (vs *VolumeServer) PostHandler(w http.ResponseWriter, r *http.Request) {
@@ -68,7 +67,7 @@ func (vs *VolumeServer) PostHandler(w http.ResponseWriter, r *http.Request) {
 		ret.Name = string(reqNeedle.Name)
 	}
 	ret.Size = uint32(originalSize)
-	ret.ETag = fmt.Sprintf("%x", util.Base64Md5ToBytes(contentMd5))
+	ret.ETag = reqNeedle.Etag()
 	ret.Mime = string(reqNeedle.Mime)
 	setEtag(w, ret.ETag)
 	w.Header().Set("Content-MD5", contentMd5)

weed/shell/command_fs_meta_load.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"strings"
 
 	"github.com/golang/protobuf/proto"
 
@@ -72,6 +73,7 @@ func (c *commandFsMetaLoad) Do(args []string, commandEnv *CommandEnv, writer io.
 				return err
 			}
 
+			fullEntry.Entry.Name = strings.ReplaceAll(fullEntry.Entry.Name, "/", "x")
 			if err := filer_pb.CreateEntry(client, &filer_pb.CreateEntryRequest{
 				Directory: fullEntry.Dir,
 				Entry:     fullEntry.Entry,

weed/shell/command_fs_meta_save.go

@@ -5,6 +5,8 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"path/filepath"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -46,6 +48,7 @@ func (c *commandFsMetaSave) Do(args []string, commandEnv *CommandEnv, writer io.
 	fsMetaSaveCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	verbose := fsMetaSaveCommand.Bool("v", false, "print out each processed files")
 	outputFileName := fsMetaSaveCommand.String("o", "", "output the meta data to this file")
+	isObfuscate := fsMetaSaveCommand.Bool("obfuscate", false, "obfuscate the file names")
 	// chunksFileName := fsMetaSaveCommand.String("chunks", "", "output all the chunks to this file")
 	if err = fsMetaSaveCommand.Parse(args); err != nil {
 		return nil
@@ -69,6 +72,11 @@ func (c *commandFsMetaSave) Do(args []string, commandEnv *CommandEnv, writer io.
 	}
 	defer dst.Close()
 
+	var cipherKey util.CipherKey
+	if *isObfuscate {
+		cipherKey = util.GenCipherKey()
+	}
+
 	err = doTraverseBfsAndSaving(commandEnv, writer, path, *verbose, func(outputChan chan interface{}) {
 		sizeBuf := make([]byte, 4)
 		for item := range outputChan {
@@ -78,6 +86,13 @@ func (c *commandFsMetaSave) Do(args []string, commandEnv *CommandEnv, writer io.
 			dst.Write(b)
 		}
 	}, func(entry *filer_pb.FullEntry, outputChan chan interface{}) (err error) {
+		if !entry.Entry.IsDirectory {
+			ext := filepath.Ext(entry.Entry.Name)
+			if encrypted, encErr := util.Encrypt([]byte(entry.Entry.Name), cipherKey); encErr == nil {
+				entry.Entry.Name = util.Base64Encode(encrypted)[:len(entry.Entry.Name)] + ext
+				entry.Entry.Name = strings.ReplaceAll(entry.Entry.Name, "/", "x")
+			}
+		}
 		bytes, err := proto.Marshal(entry)
 		if err != nil {
 			fmt.Fprintf(writer, "marshall error: %v\n", err)

weed/shell/command_volume_check_disk.go

@@ -85,6 +85,7 @@ func (c *commandVolumeCheckDisk) Do(args []string, commandEnv *CommandEnv, write
 			}
 			if a.info.ReadOnly || b.info.ReadOnly {
 				fmt.Fprintf(writer, "skipping readonly volume %d on %s and %s\n", a.info.Id, a.location.dataNode.Id, b.location.dataNode.Id)
+				replicas = replicas[1:]
 				continue
 			}
 

weed/shell/command_volume_server_evacuate.go

@@ -176,6 +176,11 @@ func moveAwayOneEcVolume(commandEnv *CommandEnv, ecShardInfo *master_pb.VolumeEc
 
 func moveAwayOneNormalVolume(commandEnv *CommandEnv, volumeReplicas map[uint32][]*VolumeReplica, vol *master_pb.VolumeInformationMessage, thisNode *Node, otherNodes []*Node, applyChange bool) (hasMoved bool, err error) {
 	fn := capacityByFreeVolumeCount(types.ToDiskType(vol.DiskType))
+	for _, n := range otherNodes {
+		n.selectVolumes(func(v *master_pb.VolumeInformationMessage) bool {
+			return v.DiskType == vol.DiskType
+		})
+	}
 	sort.Slice(otherNodes, func(i, j int) bool {
 		return otherNodes[i].localVolumeRatio(fn) > otherNodes[j].localVolumeRatio(fn)
 	})

weed/shell/command_volume_tier_move.go

@@ -133,7 +133,7 @@ func doVolumeTierMove(commandEnv *CommandEnv, writer io.Writer, collection strin
 
 			// remove the remaining replicas
 			for _, loc := range locations {
-				if loc.Url != sourceVolumeServer {
+				if loc.Url != dst.dataNode.Id {
 					if err = deleteVolume(commandEnv.option.GrpcDialOption, vid, loc.Url); err != nil {
 						fmt.Fprintf(writer, "failed to delete volume %d on %s\n", vid, loc.Url)
 					}

weed/shell/shell_liner.go

@@ -2,6 +2,7 @@ package shell
 
 import (
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/util/grace"
 	"io"
 	"os"
 	"path"
@@ -25,6 +26,9 @@ func RunShell(options ShellOptions) {
 
 	line = liner.NewLiner()
 	defer line.Close()
+	grace.OnInterrupt(func() {
+		line.Close()
+	})
 
 	line.SetCtrlCAborts(true)
 

weed/storage/disk_location.go

@@ -131,7 +131,7 @@ func (l *DiskLocation) loadExistingVolume(fileInfo os.FileInfo, needleMapKind Ne
 	l.SetVolume(vid, v)
 
 	size, _, _ := v.FileStat()
-	glog.V(0).Infof("data file %s, replicaPlacement=%s v=%d size=%d ttl=%s",
+	glog.V(0).Infof("data file %s, replication=%s v=%d size=%d ttl=%s",
 		l.Directory+"/"+volumeName+".dat", v.ReplicaPlacement, v.Version(), size, v.Ttl.String())
 	return true
 }

weed/util/chunk_cache/chunk_cache_on_disk.go

@@ -88,15 +88,17 @@ func (v *ChunkCacheVolume) Shutdown() {
 	}
 }
 
-func (v *ChunkCacheVolume) destroy() {
+func (v *ChunkCacheVolume) doReset() {
 	v.Shutdown()
-	os.Remove(v.fileName + ".dat")
-	os.Remove(v.fileName + ".idx")
+	os.Truncate(v.fileName + ".dat", 0)
+	os.Truncate(v.fileName + ".idx", 0)
+	glog.V(4).Infof("cache removeAll %s ...", v.fileName + ".ldb")
 	os.RemoveAll(v.fileName + ".ldb")
+	glog.V(4).Infof("cache removed %s", v.fileName + ".ldb")
 }
 
 func (v *ChunkCacheVolume) Reset() (*ChunkCacheVolume, error) {
-	v.destroy()
+	v.doReset()
 	return LoadOrCreateChunkCacheVolume(v.fileName, v.sizeLimit)
 }
 

weed/util/constants.go

@@ -5,7 +5,7 @@ import (
 )
 
 var (
-	VERSION = fmt.Sprintf("%s %d.%02d", sizeLimit, 2, 39)
+	VERSION = fmt.Sprintf("%s %d.%02d", sizeLimit, 2, 40)
 	COMMIT  = ""
 )