fix NFS storage according to filesystem test (#4163)

Signed-off-by: xixi <[email protected]>

.github/workflows/unittests.yml

@@ -50,6 +50,7 @@ jobs:
       ETCD_ADDR: 127.0.0.1:3379
       MYSQL_ADDR: (127.0.0.1:3306)/dev
       MYSQL_USER: root
+      NFS_ADDR: 127.0.0.1:/srv/nfs/
     steps:
       - name: Checkout
         uses: actions/checkout@v3
@@ -63,9 +64,18 @@ jobs:
 
       - name: Install Packages
         run: |
-          sudo .github/scripts/apt_install.sh g++-multilib redis-server libacl1-dev attr glusterfs-server libglusterfs-dev
+          sudo .github/scripts/apt_install.sh g++-multilib redis-server libacl1-dev attr glusterfs-server libglusterfs-dev nfs-kernel-server
           sudo mkdir -p /home/travis/.m2/
-
+      - if: matrix.test == 'test.pkg'
+        name: Set up nfs-server
+        run: |
+          sudo mkdir -p /srv/nfs
+          sudo chown nobody:nogroup /srv/nfs
+          sudo chmod 777 /srv/nfs
+          echo "/srv/nfs 127.0.0.1(rw,sync,insecure)" | sudo tee -a /etc/exports
+          sudo systemctl start nfs-kernel-server.service
+          sudo exportfs -arv
+        
       - if: matrix.test == 'test.meta.non-core'
         name: Install redis-cluster
         uses: vishnudxb/[email protected]

pkg/object/filesystem_test.go

@@ -70,6 +70,17 @@ func TestHDFS2(t *testing.T) { //skip mutate
 	testFileSystem(t, dfs)
 }
 
+func TestNFS2(t *testing.T) { //skip mutate
+	if os.Getenv("NFS_ADDR") == "" {
+		t.SkipNow()
+	}
+	b, err := newNFSStore(os.Getenv("NFS_ADDR"), os.Getenv("NFS_ACCESS_KEY"), os.Getenv("NFS_SECRET_KEY"), "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	testFileSystem(t, b)
+}
+
 func testFileSystem(t *testing.T, s ObjectStorage) {
 	keys := []string{
 		"x/",
@@ -140,6 +151,9 @@ func testFileSystem(t *testing.T, s ObjectStorage) {
 				t.Fatalf("list failed: %s mode %o", err, mode)
 			}
 			expectedKeys = []string{"x/", "xy.txt", "xyz/", "xyz/xyz.txt"}
+			if _, ok := ss.(*nfsStore); ok {
+				expectedKeys = []string{"x/", "x/x.txt", "xy.txt", "xyz/", "xyz/xyz.txt"}
+			}
 			if mode == 0422 {
 				if _, ok := ss.(*gluster); ok {
 					expectedKeys = []string{"x/", "x/x.txt", "xy.txt", "xyz/", "xyz/xyz.txt"}

pkg/object/nfs.go

@@ -60,6 +60,13 @@ func (e *nfsEntry) Name() string {
 	return e.name
 }
 
+func (e *nfsEntry) Size() int64 {
+	if e.fi != nil {
+		return e.fi.Size()
+	}
+	return e.EntryPlus.Size()
+}
+
 func (e *nfsEntry) Info() (os.FileInfo, error) {
 	if e.fi != nil {
 		return e.fi, nil
@@ -91,6 +98,14 @@ func (n *nfsStore) Head(key string) (Object, error) {
 	if err != nil {
 		return nil, err
 	}
+	if attr, ok := fi.(*nfs.Fattr); ok && attr.Type == nfs.NF3Lnk {
+		src, err := n.Readlink(p)
+		if err != nil {
+			return nil, err
+		}
+		dir, _ := path.Split(p)
+		return n.Head(path.Join(dir, src))
+	}
 	return n.fileInfo(key, fi), nil
 }
 
@@ -188,7 +203,23 @@ func (n *nfsStore) Put(key string, in io.Reader) error {
 }
 
 func (n *nfsStore) Delete(key string) error {
-	err := n.target.Remove(strings.TrimRight(n.path(key), dirSuffix))
+	path := n.path(key)
+	if path == "./" {
+		return nil
+	}
+	fi, _, err := n.target.Lookup(path)
+	if err != nil {
+		if nfs.IsNotDirError(err) || os.IsNotExist(err) {
+			return nil
+		}
+		return err
+	}
+	p := strings.TrimSuffix(path, "/")
+	if fi.IsDir() {
+		err = n.target.RmDir(p)
+	} else {
+		err = n.target.Remove(p)
+	}
 	if err != nil && os.IsNotExist(err) {
 		err = nil
 	}
@@ -214,21 +245,32 @@ func (n *nfsStore) fileInfo(key string, fi os.FileInfo) Object {
 	return ff
 }
 
-func (n *nfsStore) readDirSorted(dirname string, followLink bool) ([]*nfsEntry, error) {
+func (n *nfsStore) readDirSorted(dir string, followLink bool) ([]*nfsEntry, error) {
+	o, err := n.Head(strings.TrimSuffix(dir, "/"))
+	if err != nil {
+		return nil, err
+	}
+	dirname := o.Key()
 	entries, err := n.target.ReadDirPlus(dirname)
 	if err != nil {
 		return nil, errors.Wrapf(err, "readdir %s", dirname)
 	}
 	nfsEntries := make([]*nfsEntry, len(entries))
-
 	for i, e := range entries {
 		if e.IsDir() {
 			nfsEntries[i] = &nfsEntry{e, e.Name() + dirSuffix, nil, false}
 		} else if e.Attr.Attr.Type == nfs.NF3Lnk && followLink {
 			// follow symlink
-			fi, _, err := n.target.Lookup(path.Join(dirname, e.Name()))
+			nfsEntries[i] = &nfsEntry{e, e.Name(), nil, true}
+			src, err := n.Readlink(path.Join(dirname, e.Name()))
+			if err != nil {
+				logger.Errorf("readlink %s: %s", e.Name(), err)
+				continue
+			}
+			srcPath := path.Clean(path.Join(dirname, src))
+			fi, _, err := n.target.Lookup(srcPath)
 			if err != nil {
-				nfsEntries[i] = &nfsEntry{e, e.Name(), nil, true}
+				logger.Warnf("follow link `%s`: lookup `%s`: %s", path.Join(dirname, e.Name()), srcPath, err)
 				continue
 			}
 			name := e.Name()
@@ -248,9 +290,9 @@ func (n *nfsStore) List(prefix, marker, delimiter string, limit int64, followLin
 	if delimiter != "/" {
 		return nil, notSupported
 	}
-	dir := n.path(prefix)
+	dir := prefix
 	var objs []Object
-	if !strings.HasSuffix(dir, dirSuffix) {
+	if dir != "" && !strings.HasSuffix(dir, dirSuffix) {
 		dir = path.Dir(dir)
 		if !strings.HasSuffix(dir, dirSuffix) {
 			dir += dirSuffix
@@ -267,7 +309,7 @@ func (n *nfsStore) List(prefix, marker, delimiter string, limit int64, followLin
 	}
 	entries, err := n.readDirSorted(dir, followLink)
 	if err != nil {
-		if os.IsPermission(err) {
+		if os.IsPermission(err) || errors.Is(err, nfs.NFS3Error(nfs.NFS3ErrAcces)) {
 			logger.Warnf("skip %s: %s", dir, err)
 			return nil, nil
 		}