Merge branch 'fix-pysdk-setxattr' of github.com:juicedata/juicefs int…

…o fix-pysdk-setxattr

.github/scripts/hypo/command.py

@@ -155,7 +155,7 @@ def fsck(self, entry, repair=False, recuisive=False, user='root'):
     @rule(
         entry = Entries.filter(lambda x: x != multiple()),
         parent = Folders.filter(lambda x: x != multiple()),
-        new_entry_name = st_entry_name,
+        new_entry_name = st_file_name,
         user = st_sudo_user,
         preserve = st.booleans()
     )

.github/scripts/hypo/command_op.py

@@ -8,8 +8,6 @@
     __import__('xattr')
 except ImportError:
     subprocess.check_call(["pip", "install", "xattr"])
-from common import is_jfs, get_acl, get_root, get_stat
-from typing import Dict
 try: 
     __import__('psutil')
 except ImportError:

.github/scripts/hypo/command_test.py

@@ -5,7 +5,7 @@ class TestCommand(unittest.TestCase):
     def test_dump(self):
         state = JuicefsCommandMachine()
         folders_0 = state.init_folders()
-        files_0 = state.create_file(content=b'', file_name='aazz', mode='w', parent=folders_0, umask=312, user='root')
+        files_0 = state.create_file(content='', file_name='aazz', mode='w', parent=folders_0, umask=312, user='root')
         value = ''.join([chr(i) for i in range(256)])
         value = value.encode('latin-1')
         value = b'\x2580q\x2589'
@@ -17,14 +17,14 @@ def test_dump(self):
     def skip_test_info(self):
         state = JuicefsCommandMachine()
         folders_0 = state.init_folders()
-        files_2 = state.create_file(content=b'0', file_name='mvvd', mode='a', parent=folders_0, umask=293, user='root')
+        files_2 = state.create_file(content='0', file_name='mvvd', mode='a', parent=folders_0, umask=293, user='root')
         state.info(entry=folders_0, raw=True, recuisive=True, user='user1')
         state.teardown()
 
     def test_clone(self):
         state = JuicefsCommandMachine()
         v1 = state.init_folders()
-        v2 = state.create_file(content=b'\x9bcR\xba', file_name='ygbl', mode='x', parent=v1, umask=466, user='root')
+        v2 = state.create_file(content='\x9bcR\xba', file_name='ygbl', mode='x', parent=v1, umask=466, user='root')
         state.chmod(entry=v1, mode=715, user='root')
         state.clone(entry=v2, new_entry_name='drqj', parent=v1, preserve=False, user='user1')
         state.teardown()

.github/scripts/sync/sync_cluster.sh

@@ -61,7 +61,7 @@ test_sync_without_mount_point(){
 
     sudo -u juicedata meta_url=$META_URL ./juicefs sync -v jfs://meta_url/data/ minio://minioadmin:[email protected]:9000/data1/ \
          --manager-addr 172.20.0.1:8081 --worker [email protected],[email protected] \
-         --list-threads 10 --list-depth 5 \
+         --list-threads 10 --list-depth 5 --check-new \
          2>&1 | tee sync.log
     # diff data/ /jfs/data1/
     check_sync_log $file_count
@@ -86,6 +86,10 @@ test_sync_without_mount_point2(){
          --list-threads 10 --list-depth 5\
          2>&1 | tee sync.log
     check_sync_log $file_count
+    sudo -u juicedata meta_url=$META_URL ./juicefs sync -v  minio://minioadmin:[email protected]:9000/data/ jfs://meta_url/ \
+         --manager-addr 172.20.0.1:8081 --worker [email protected],[email protected] \
+         --list-threads 10 --list-depth 5 --check-all \
+         2>&1 | tee sync.log
     ./juicefs mount -d $META_URL /jfs
     diff data/ /jfs/data/
     ./mc rm -r --force myminio/data

.github/workflows/command2.yml

@@ -120,8 +120,8 @@ jobs:
           grep "<FATAL>:" /var/log/juicefs.log && exit 1 || true
           
       - name: Setup upterm session
-        # if: failure() && (github.event.inputs.debug == 'true' || github.run_attempt != 1)
-        if: failure()
+        if: failure() && (github.event.inputs.debug == 'true' || github.run_attempt != 1)
+        # if: failure()
         timeout-minutes: 60
         uses: lhotari/action-upterm@v1
 

cmd/object.go

@@ -117,6 +117,9 @@ var bufPool = sync.Pool{
 }
 
 func (j *juiceFS) Put(key string, in io.Reader, getters ...object.AttrGetter) (err error) {
+	if vfs.IsSpecialName(key) {
+		return fmt.Errorf("skip special file %s for jfs: %w", key, utils.ErrSkipped)
+	}
 	p := j.path(key)
 	if strings.HasSuffix(p, "/") {
 		eno := j.jfs.MkdirAll(ctx, p, 0777, j.umask)

go.mod

@@ -46,7 +46,6 @@ require (
 	github.com/minio/minio v0.0.0-20210206053228-97fe57bba92c
 	github.com/minio/minio-go/v7 v7.0.11-0.20210302210017-6ae69c73ce78
 	github.com/ncw/swift/v2 v2.0.1
-	github.com/olekukonko/tablewriter v0.0.1
 	github.com/pingcap/log v1.1.1-0.20221015072633-39906604fb81
 	github.com/pkg/errors v0.9.1
 	github.com/pkg/sftp v1.13.5
@@ -66,6 +65,7 @@ require (
 	github.com/urfave/cli/v2 v2.19.3
 	github.com/vbauerster/mpb/v7 v7.0.3
 	github.com/viki-org/dnscache v0.0.0-20130720023526-c70c1f23c5d8
+	github.com/vimeo/go-util v1.4.1
 	github.com/vmware/go-nfs-client v0.0.0-20190605212624-d43b92724c1b
 	github.com/volcengine/ve-tos-golang-sdk/v2 v2.7.0
 	github.com/youmark/pkcs8 v0.0.0-20201027041543-1326539a0a0a

go.sum

@@ -668,7 +668,6 @@ github.com/nrdcg/namesilo v0.2.1/go.mod h1:lwMvfQTyYq+BbjJd30ylEG4GPSS6PII0Tia4r
 github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
 github.com/nxadm/tail v1.4.8/go.mod h1:+ncqLTQzXmGhMZNUePPaPqPvBxHAIsmXswZKocGu+AU=
 github.com/olekukonko/tablewriter v0.0.0-20170122224234-a0225b3f23b5/go.mod h1:vsDQFd/mU46D+Z4whnwzcISnGGzXWMclvtLoiIKAKIo=
-github.com/olekukonko/tablewriter v0.0.1 h1:b3iUnf1v+ppJiOfNX4yxxqfWKMQPZR5yoh8urCTFX88=
 github.com/olekukonko/tablewriter v0.0.1/go.mod h1:vsDQFd/mU46D+Z4whnwzcISnGGzXWMclvtLoiIKAKIo=
 github.com/oliverisaac/shellescape v0.0.0-20220131224704-1b6c6b87b668 h1:WUilXdVrxYH+fFkmstviAOj1o9CfoW5O/Sd0LWPIVUA=
 github.com/oliverisaac/shellescape v0.0.0-20220131224704-1b6c6b87b668/go.mod h1:EDgl+cvbmeOQUMTTH94gjXVtFHr8xDe5BiXhWn7Hf1E=
@@ -903,6 +902,8 @@ github.com/valyala/tcplisten v1.0.0 h1:rBHj/Xf+E1tRGZyWIWwJDiRY0zc1Js+CV5DqwacVS
 github.com/valyala/tcplisten v1.0.0/go.mod h1:T0xQ8SeCZGxckz9qRXTfG43PvQ/mcWh7FwZEA7Ioqkc=
 github.com/viki-org/dnscache v0.0.0-20130720023526-c70c1f23c5d8 h1:EVObHAr8DqpoJCVv6KYTle8FEImKhtkfcZetNqxDoJQ=
 github.com/viki-org/dnscache v0.0.0-20130720023526-c70c1f23c5d8/go.mod h1:dniwbG03GafCjFohMDmz6Zc6oCuiqgH6tGNyXTkHzXE=
+github.com/vimeo/go-util v1.4.1 h1:UbNoaYH1eHv4LqBSH6zIItj+zKqbln0i01oY3iA/QPM=
+github.com/vimeo/go-util v1.4.1/go.mod h1:r+yspV//C48HeMXV8nEvtUeNiIiGfVv3bbEHzOgudwE=
 github.com/volcengine/ve-tos-golang-sdk/v2 v2.7.0 h1:MnTrrKb7gvWoI1W5GxVnjjzdSPmms4++JiR3ioqqoRc=
 github.com/volcengine/ve-tos-golang-sdk/v2 v2.7.0/go.mod h1:IrjK84IJJTuOZOTMv/P18Ydjy/x+ow7fF7q11jAxXLM=
 github.com/vultr/govultr v0.1.4/go.mod h1:9H008Uxr/C4vFNGLqKx232C206GL0PBHzOP0809bGNA=

pkg/meta/redis.go

@@ -908,14 +908,11 @@ func (m *redisMeta) Resolve(ctx Context, parent Ino, path string, inode *Ino, at
 }
 
 func (m *redisMeta) doGetAttr(ctx Context, inode Ino, attr *Attr) syscall.Errno {
-	return errno(m.rdb.Watch(ctx, func(tx *redis.Tx) error {
-		val, err := tx.Get(ctx, m.inodeKey(inode)).Bytes()
-		if err != nil {
-			return err
-		}
-		m.parseAttr(val, attr)
-		return nil
-	}, m.inodeKey(inode)))
+	a, err := m.rdb.Get(ctx, m.inodeKey(inode)).Bytes()
+	if err == nil {
+		m.parseAttr(a, attr)
+	}
+	return errno(err)
 }
 
 type timeoutError interface {
@@ -4552,29 +4549,27 @@ func (m *redisMeta) doSetFacl(ctx Context, ino Ino, aclType uint8, rule *aclAPI.
 }
 
 func (m *redisMeta) doGetFacl(ctx Context, ino Ino, aclType uint8, aclId uint32, rule *aclAPI.Rule) syscall.Errno {
-	return errno(m.rdb.Watch(ctx, func(tx *redis.Tx) error {
-		if aclId == aclAPI.None {
-			val, err := tx.Get(ctx, m.inodeKey(ino)).Bytes()
-			if err != nil {
-				return err
-			}
-			attr := &Attr{}
-			m.parseAttr(val, attr)
-			m.of.Update(ino, attr)
-
-			aclId = getAttrACLId(attr, aclType)
-		}
-
-		a, err := m.getACL(ctx, tx, aclId)
+	if aclId == aclAPI.None {
+		val, err := m.rdb.Get(ctx, m.inodeKey(ino)).Bytes()
 		if err != nil {
-			return err
-		}
-		if a == nil {
-			return ENOATTR
+			return errno(err)
 		}
-		*rule = *a
-		return nil
-	}, m.inodeKey(ino)))
+		attr := &Attr{}
+		m.parseAttr(val, attr)
+		m.of.Update(ino, attr)
+
+		aclId = getAttrACLId(attr, aclType)
+	}
+
+	a, err := m.getACL(ctx, nil, aclId)
+	if err != nil {
+		return errno(err)
+	}
+	if a == nil {
+		return ENOATTR
+	}
+	*rule = *a
+	return 0
 }
 
 func (m *redisMeta) getACL(ctx Context, tx *redis.Tx, id uint32) (*aclAPI.Rule, error) {
@@ -4585,15 +4580,13 @@ func (m *redisMeta) getACL(ctx Context, tx *redis.Tx, id uint32) (*aclAPI.Rule,
 		return cRule, nil
 	}
 
-	cmds, err := tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
-		pipe.HGet(ctx, m.aclKey(), strconv.FormatUint(uint64(id), 10))
-		return nil
-	})
-	if err != nil {
-		return nil, err
+	var val []byte
+	var err error
+	if tx != nil {
+		val, err = tx.HGet(ctx, m.aclKey(), strconv.FormatUint(uint64(id), 10)).Bytes()
+	} else {
+		val, err = m.rdb.HGet(ctx, m.aclKey(), strconv.FormatUint(uint64(id), 10)).Bytes()
 	}
-
-	val, err := cmds[0].(*redis.StringCmd).Bytes()
 	if err != nil {
 		return nil, err
 	}

pkg/sync/cluster.go

@@ -330,7 +330,16 @@ func launchWorker(address string, config *Config, wg *sync.WaitGroup) {
 func marshalObjects(objs []object.Object) ([]byte, error) {
 	var arr []map[string]interface{}
 	for _, o := range objs {
-		arr = append(arr, object.MarshalObject(o))
+		obj := object.MarshalObject(o)
+		switch oo := o.(type) {
+		case *withSize:
+			obj["nsize"] = oo.nsize
+			obj["size"] = oo.Object.Size()
+		case *withFSize:
+			obj["fnsize"] = oo.nsize
+			obj["size"] = oo.File.Size()
+		}
+		arr = append(arr, obj)
 	}
 	return json.MarshalIndent(arr, "", " ")
 }
@@ -343,7 +352,13 @@ func unmarshalObjects(d []byte) ([]object.Object, error) {
 	}
 	var objs []object.Object
 	for _, m := range arr {
-		objs = append(objs, object.UnmarshalObject(m))
+		obj := object.UnmarshalObject(m)
+		if nsize, ok := m["nsize"]; ok {
+			obj = &withSize{obj, int64(nsize.(float64))}
+		} else if fnsize, ok := m["fnsize"]; ok {
+			obj = &withFSize{obj.(object.File), int64(fnsize.(float64))}
+		}
+		objs = append(objs, obj)
 	}
 	return objs, nil
 }

pkg/sync/cluster_test.go

@@ -16,6 +16,7 @@
 package sync
 
 import (
+	"os"
 	"testing"
 	"time"
 
@@ -36,6 +37,9 @@ func (o *obj) Mtime() time.Time     { return o.mtime }
 func (o *obj) IsDir() bool          { return o.isDir }
 func (o *obj) IsSymlink() bool      { return o.isSymlink }
 func (o *obj) StorageClass() string { return "" }
+func (o *obj) Owner() string        { return "" }
+func (o *obj) Group() string        { return "" }
+func (o *obj) Mode() os.FileMode    { return 0 }
 
 func TestCluster(t *testing.T) {
 	// manager
@@ -62,3 +66,28 @@ func TestCluster(t *testing.T) {
 		t.Fatalf("should end")
 	}
 }
+
+func TestMarshal(t *testing.T) {
+	var objs = []object.Object{
+		&obj{key: "test"},
+		&withSize{&obj{key: "test1", size: 100}, -4},
+		&withFSize{&obj{key: "test2", size: 200}, -1},
+	}
+	d, err := marshalObjects(objs)
+	if err != nil {
+		t.Fatal(err)
+	}
+	objs2, e := unmarshalObjects(d)
+	if e != nil {
+		t.Fatal(e)
+	}
+	if objs2[0].Key() != "test" {
+		t.Fatalf("expect test but got %s", objs2[0].Key())
+	}
+	if objs2[1].Key() != "test1" || objs2[1].Size() != -4 || objs2[1].(*withSize).Object.Size() != 100 {
+		t.Fatalf("expect withSize but got %s", objs2[0].Key())
+	}
+	if objs2[2].Key() != "test2" || objs2[2].Size() != -1 || objs2[2].(*withFSize).File.Size() != 200 {
+		t.Fatalf("expect withFSize but got %s", objs2[0].Key())
+	}
+}