2.60

Fix mysql sql for batch delete

go.mod

@@ -14,7 +14,6 @@ require (
 	github.com/cespare/xxhash v1.1.0
 	github.com/chrislusf/raft v1.0.7
 	github.com/coreos/go-semver v0.3.0 // indirect
-	github.com/dgrijalva/jwt-go v3.2.0+incompatible
 	github.com/disintegration/imaging v1.6.2
 	github.com/dustin/go-humanize v1.0.0
 	github.com/eapache/go-resiliency v1.2.0 // indirect
@@ -30,6 +29,7 @@ require (
 	github.com/go-sql-driver/mysql v1.5.0
 	github.com/gocql/gocql v0.0.0-20190829130954-e163eff7a8c6
 	github.com/gogo/protobuf v1.2.2-0.20190730201129-28a6bbf47e48 // indirect
+	github.com/golang-jwt/jwt v3.2.1+incompatible
 	github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e
 	github.com/golang/protobuf v1.4.3
 	github.com/google/btree v1.0.0
@@ -66,8 +66,8 @@ require (
 	github.com/stretchr/testify v1.6.1
 	github.com/syndtr/goleveldb v1.0.0
 	github.com/tecbot/gorocksdb v0.0.0-20191217155057-f0fad39f321c
-	github.com/tidwall/gjson v1.3.2
-	github.com/tidwall/match v1.0.1
+	github.com/tidwall/gjson v1.8.1
+	github.com/tidwall/match v1.0.3
 	github.com/tsuna/gohbase v0.0.0-20201125011725-348991136365
 	github.com/valyala/bytebufferpool v1.0.0
 	github.com/viant/assertly v0.5.4 // indirect
@@ -76,7 +76,7 @@ require (
 	github.com/willf/bitset v1.1.10 // indirect
 	github.com/willf/bloom v2.0.3+incompatible
 	go.etcd.io/etcd v3.3.15+incompatible
-	go.mongodb.org/mongo-driver v1.3.2
+	go.mongodb.org/mongo-driver v1.7.0
 	gocloud.dev v0.20.0
 	gocloud.dev/pubsub/natspubsub v0.20.0
 	gocloud.dev/pubsub/rabbitpubsub v0.20.0

go.sum

@@ -265,6 +265,8 @@ github.com/gogo/protobuf v1.2.0/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7a
 github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4=
 github.com/gogo/protobuf v1.2.2-0.20190730201129-28a6bbf47e48 h1:X+zN6RZXsvnrSJaAIQhZezPfAfvsqihKKR8oiLHid34=
 github.com/gogo/protobuf v1.2.2-0.20190730201129-28a6bbf47e48/go.mod h1:SlYgWuQ5SjCEi6WLHjHCa1yvBfUnHcTbrrZtXPKa29o=
+github.com/golang-jwt/jwt v3.2.1+incompatible h1:73Z+4BJcrTC+KczS6WvTPvRGOp1WmfEP4Q1lOd9Z/+c=
+github.com/golang-jwt/jwt v3.2.1+incompatible/go.mod h1:8pz2t5EyA70fFQQSrl6XZXzqecmYZeUEB8OUGHkxJ+I=
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
 github.com/golang/groupcache v0.0.0-20160516000752-02826c3e7903/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
 github.com/golang/groupcache v0.0.0-20190129154638-5b532d6fd5ef/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
@@ -307,6 +309,7 @@ github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMyw
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.4.1/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.3/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
@@ -549,8 +552,8 @@ github.com/pact-foundation/pact-go v1.0.4/go.mod h1:uExwJY4kCzNPcHRj+hCR/HBbOOIw
 github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
 github.com/pborman/uuid v1.2.0/go.mod h1:X/NO0urCmaxf9VXbdlT7C2Yzkj2IKimNn4k+gtPdI/k=
 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
-github.com/pelletier/go-toml v1.4.0 h1:u3Z1r+oOXJIkxqw34zVhyPgjBsm6X2wn21NWs/HfSeg=
-github.com/pelletier/go-toml v1.4.0/go.mod h1:PN7xzY2wHTK0K9p34ErDQMlFxa51Fk0OUruD3k1mMwo=
+github.com/pelletier/go-toml v1.7.0 h1:7utD74fnzVc/cpcyy8sjrlFr5vYpypUixARcHIMIGuI=
+github.com/pelletier/go-toml v1.7.0/go.mod h1:vwGMzjaWMwyfHwgIBhI2YUM4fB6nL6lVAvS1LBMMhTE=
 github.com/performancecopilot/speed v3.0.0+incompatible/go.mod h1:/CLtqpZ5gBg1M9iaPbIdPPGyKcA8hKdoy6hAWba7Yac=
 github.com/peterh/liner v1.1.0 h1:f+aAedNJA6uk7+6rXsYBnhdo4Xux7ESLe+kcuVUF5os=
 github.com/peterh/liner v1.1.0/go.mod h1:CRroGNssyjTd/qIG2FyxByd2S8JEAZXBl4qUrZf8GS0=
@@ -615,8 +618,6 @@ github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQD
 github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts=
 github.com/samuel/go-zookeeper v0.0.0-20190923202752-2cc03de413da/go.mod h1:gi+0XIa01GRL2eRQVjQkKGqKF3SF9vZR/HnPullcV2E=
 github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
-github.com/seaweedfs/fuse v1.1.7 h1:T4L5c/Sn+q8lE+0zCmH2MWvIO+B5TttWOSqK5KQPRMQ=
-github.com/seaweedfs/fuse v1.1.7/go.mod h1:+PP6WlkrRUG6KPE+Th2EX5To/PjHaFsvqg/UgQ39aj8=
 github.com/seaweedfs/fuse v1.1.8 h1:YFSDPotG4uhQzV7ooDUvQ8BRVy5rM1XCFPJAmAsZz68=
 github.com/seaweedfs/fuse v1.1.8/go.mod h1:+PP6WlkrRUG6KPE+Th2EX5To/PjHaFsvqg/UgQ39aj8=
 github.com/seaweedfs/goexif v1.0.2 h1:p+rTXYdQ2mgxd+1JaTrQ9N8DvYuw9UH9xgYmJ+Bb29E=
@@ -677,12 +678,13 @@ github.com/syndtr/goleveldb v1.0.0 h1:fBdIW9lB4Iz0n9khmH8w27SJ3QEJ7+IgjPEwGSZiFd
 github.com/syndtr/goleveldb v1.0.0/go.mod h1:ZVVdQEZoIme9iO1Ch2Jdy24qqXrMMOU6lpPAyBWyWuQ=
 github.com/tecbot/gorocksdb v0.0.0-20191217155057-f0fad39f321c h1:g+WoO5jjkqGAzHWCjJB1zZfXPIAaDpzXIEJ0eS6B5Ok=
 github.com/tecbot/gorocksdb v0.0.0-20191217155057-f0fad39f321c/go.mod h1:ahpPrc7HpcfEWDQRZEmnXMzHY03mLDYMCxeDzy46i+8=
-github.com/tidwall/gjson v1.3.2 h1:+7p3qQFaH3fOMXAJSrdZwGKcOO/lYdGS0HqGhPqDdTI=
-github.com/tidwall/gjson v1.3.2/go.mod h1:P256ACg0Mn+j1RXIDXoss50DeIABTYK1PULOJHhxOls=
-github.com/tidwall/match v1.0.1 h1:PnKP62LPNxHKTwvHHZZzdOAOCtsJTjo6dZLCwpKm5xc=
-github.com/tidwall/match v1.0.1/go.mod h1:LujAq0jyVjBy028G1WhWfIzbpQfMO8bBZ6Tyb0+pL9E=
-github.com/tidwall/pretty v1.0.0 h1:HsD+QiTn7sK6flMKIvNmpqz1qrpP3Ps6jOKIKMooyg4=
+github.com/tidwall/gjson v1.8.1 h1:8j5EE9Hrh3l9Od1OIEDAb7IpezNA20UdRngNAj5N0WU=
+github.com/tidwall/gjson v1.8.1/go.mod h1:5/xDoumyyDNerp2U36lyolv46b3uF/9Bu6OfyQ9GImk=
+github.com/tidwall/match v1.0.3 h1:FQUVvBImDutD8wJLN6c5eMzWtjgONK9MwIBCOrUJKeE=
+github.com/tidwall/match v1.0.3/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
 github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
+github.com/tidwall/pretty v1.1.0 h1:K3hMW5epkdAVwibsQEfR/7Zj0Qgt4DxtNumTq/VloO8=
+github.com/tidwall/pretty v1.1.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20170815181823-89b8d40f7ca8/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20190109142713-0ad062ec5ee5 h1:LnC5Kc/wtumK+WB441p7ynQJzVuNRJiqddSIE3IlSEQ=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20190109142713-0ad062ec5ee5/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
@@ -705,14 +707,19 @@ github.com/willf/bloom v2.0.3+incompatible h1:QDacWdqcAUI1MPOwIQZRy9kOR7yxfyEmxX
 github.com/willf/bloom v2.0.3+incompatible/go.mod h1:MmAltL9pDMNTrvUkxdg0k0q5I0suxmuwp3KbyrZLOZ8=
 github.com/wsxiaoys/terminal v0.0.0-20160513160801-0940f3fc43a0 h1:3UeQBvD0TFrlVjOeLOBz+CPAI8dnbqNSVwUwRrkp7vQ=
 github.com/wsxiaoys/terminal v0.0.0-20160513160801-0940f3fc43a0/go.mod h1:IXCdmsXIht47RaVFLEdVnh1t+pgYtTAhQGj73kz+2DM=
-github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c h1:u40Z8hqBAAQyv+vATcGgV0YCnDjqSL7/q/JyPhhJSPk=
+github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c=
+github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
+github.com/xdg-go/scram v1.0.2 h1:akYIkZ28e6A96dkWNJQu3nmCzH3YfwMPQExUYDaRv7w=
+github.com/xdg-go/scram v1.0.2/go.mod h1:1WAq6h33pAW+iRreB34OORO2Nf7qel3VV3fjBj+hCSs=
+github.com/xdg-go/stringprep v1.0.2 h1:6iq84/ryjjeRmMJwxutI51F2GIPlP5BfTvXHeYjyhBc=
+github.com/xdg-go/stringprep v1.0.2/go.mod h1:8F9zXuvzgwmyT5DUm4GUfZGDdT3W+LCvS6+da4O5kxM=
 github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c/go.mod h1:lB8K/P019DLNhemzwFU4jHLhdvlE6uDZjXFejJXr49I=
-github.com/xdg/stringprep v0.0.0-20180714160509-73f8eece6fdc/go.mod h1:Jhud4/sHMO4oL310DaZAKk9ZaJ08SJfe+sJh0HrGL1Y=
-github.com/xdg/stringprep v1.0.0 h1:d9X0esnoa3dFsV0FG35rAT0RIhYFlPq7MiP+DW89La0=
 github.com/xdg/stringprep v1.0.0/go.mod h1:Jhud4/sHMO4oL310DaZAKk9ZaJ08SJfe+sJh0HrGL1Y=
 github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2 h1:eY9dn8+vbi4tKz5Qo6v2eYzo7kUS51QINcR5jNpbZS8=
 github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2/go.mod h1:UETIi67q53MR2AWcXfiuqkDkRtnGDLqkBTpCHuJHxtU=
 github.com/xordataexchange/crypt v0.0.3-0.20170626215501-b2862e3d0a77/go.mod h1:aYKd//L2LvnjZzWKhF00oedf4jCCReLcmhLdhm1A27Q=
+github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d h1:splanxYIlg+5LfHAM6xpdFEAYOk8iySO56hMFq6uLyA=
+github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d/go.mod h1:rHwXgn7JulP+udvsHwJoVG1YGAP6VLg4y9I5dyZdqmA=
 github.com/yuin/goldmark v1.1.25/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
@@ -721,8 +728,8 @@ go.etcd.io/bbolt v1.3.4 h1:hi1bXHMVrlQh6WwxAy+qZCV/SYIlqo+Ushwdpa4tAKg=
 go.etcd.io/bbolt v1.3.4/go.mod h1:G5EMThwa9y8QZGBClrRx5EY+Yw9kAhnjy3bSjsnlVTQ=
 go.etcd.io/etcd v0.5.0-alpha.5.0.20200425165423-262c93980547 h1:s71VGheLtWmCYsnNjf+s7XE8HsrZnd3EYGrLGWVm7nY=
 go.etcd.io/etcd v0.5.0-alpha.5.0.20200425165423-262c93980547/go.mod h1:YoUyTScD3Vcv2RBm3eGVOq7i1ULiz3OuXoQFWOirmAM=
-go.mongodb.org/mongo-driver v1.3.2 h1:IYppNjEV/C+/3VPbhHVxQ4t04eVW0cLp0/pNdW++6Ug=
-go.mongodb.org/mongo-driver v1.3.2/go.mod h1:MSWZXKOynuguX+JSvwP8i+58jYCXxbia8HS3gZBapIE=
+go.mongodb.org/mongo-driver v1.7.0 h1:hHrvOBWlWB2c7+8Gh/Xi5jj82AgidK/t7KVXBZ+IyUA=
+go.mongodb.org/mongo-driver v1.7.0/go.mod h1:Q4oFMbo1+MSNqICAdYMlC/zSTrwCogR4R8NzkI+yfU8=
 go.opencensus.io v0.15.0/go.mod h1:UffZAU+4sDEINUGP/B7UfBBkq4fqLu9zXAX7ke6CHW0=
 go.opencensus.io v0.20.1/go.mod h1:6WKK9ahsWS3RSO+PY9ZHZUfv2irvY6gN279GOPZjmmk=
 go.opencensus.io v0.20.2/go.mod h1:6WKK9ahsWS3RSO+PY9ZHZUfv2irvY6gN279GOPZjmmk=
@@ -768,6 +775,7 @@ golang.org/x/crypto v0.0.0-20190820162420-60c769a6c586/go.mod h1:yigFU9vqHzYiE8U
 golang.org/x/crypto v0.0.0-20191002192127-34f69633bfdc/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20191206172530-e9b2fee46413/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/crypto v0.0.0-20200302210943-78000ba7a073/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200323165209-0ec3e9974c59/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200604202706-70a84ac30bf9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9 h1:psW17arqaxU48Z5kZ0CQnkZWQJsqcURM6tKiBApRjXI=
@@ -921,8 +929,9 @@ golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fq
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
-golang.org/x/text v0.3.3 h1:cokOdA+Jmi5PJGXLlLllQSgYigAEfHXJAERHVMaCc2k=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.5 h1:i6eZZ+zk0SOf0xgBpEpPD18qWcJda6q1sxt3S0kzyUQ=
+golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=

k8s/seaweedfs/Chart.yaml

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

k8s/seaweedfs/values.yaml

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

other/java/client/src/main/proto/filer.proto

@@ -101,6 +101,13 @@ message Entry {
     bytes hard_link_id = 7;
     int32 hard_link_counter = 8; // only exists in hard link meta data
     bytes content = 9; // if not empty, the file content
+
+    message Remote {
+        int64 last_modified_at = 1;
+        int64 size = 2;
+        string e_tag = 3;
+    }
+    Remote remote = 10;
 }
 
 message FullEntry {
@@ -367,3 +374,12 @@ message FilerConf {
     }
     repeated PathConf locations = 2;
 }
+
+message RemoteConf {
+    string type = 1;
+    string name = 2;
+    string s3_access_key = 4;
+    string s3_secret_key = 5;
+    string s3_region = 6;
+    string s3_endpoint = 7;
+}

weed/command/command.go

@@ -8,15 +8,15 @@ import (
 )
 
 var Commands = []*Command{
-	cmdBenchmark,
 	cmdBackup,
+	cmdBenchmark,
 	cmdCompact,
-	cmdCopy,
 	cmdDownload,
 	cmdExport,
 	cmdFiler,
 	cmdFilerBackup,
 	cmdFilerCat,
+	cmdFilerCopy,
 	cmdFilerMetaBackup,
 	cmdFilerMetaTail,
 	cmdFilerReplicate,

weed/command/filer_copy.go

@@ -52,21 +52,21 @@ type CopyOptions struct {
 }
 
 func init() {
-	cmdCopy.Run = runCopy // break init cycle
-	cmdCopy.IsDebug = cmdCopy.Flag.Bool("debug", false, "verbose debug information")
-	copy.include = cmdCopy.Flag.String("include", "", "pattens of files to copy, e.g., *.pdf, *.html, ab?d.txt, works together with -dir")
-	copy.replication = cmdCopy.Flag.String("replication", "", "replication type")
-	copy.collection = cmdCopy.Flag.String("collection", "", "optional collection name")
-	copy.ttl = cmdCopy.Flag.String("ttl", "", "time to live, e.g.: 1m, 1h, 1d, 1M, 1y")
-	copy.diskType = cmdCopy.Flag.String("disk", "", "[hdd|ssd|<tag>] hard drive or solid state drive or any tag")
-	copy.maxMB = cmdCopy.Flag.Int("maxMB", 4, "split files larger than the limit")
-	copy.concurrenctFiles = cmdCopy.Flag.Int("c", 8, "concurrent file copy goroutines")
-	copy.concurrenctChunks = cmdCopy.Flag.Int("concurrentChunks", 8, "concurrent chunk copy goroutines for each file")
-	copy.checkSize = cmdCopy.Flag.Bool("check.size", false, "copy when the target file size is different from the source file")
-	copy.verbose = cmdCopy.Flag.Bool("verbose", false, "print out details during copying")
+	cmdFilerCopy.Run = runCopy // break init cycle
+	cmdFilerCopy.IsDebug = cmdFilerCopy.Flag.Bool("debug", false, "verbose debug information")
+	copy.include = cmdFilerCopy.Flag.String("include", "", "pattens of files to copy, e.g., *.pdf, *.html, ab?d.txt, works together with -dir")
+	copy.replication = cmdFilerCopy.Flag.String("replication", "", "replication type")
+	copy.collection = cmdFilerCopy.Flag.String("collection", "", "optional collection name")
+	copy.ttl = cmdFilerCopy.Flag.String("ttl", "", "time to live, e.g.: 1m, 1h, 1d, 1M, 1y")
+	copy.diskType = cmdFilerCopy.Flag.String("disk", "", "[hdd|ssd|<tag>] hard drive or solid state drive or any tag")
+	copy.maxMB = cmdFilerCopy.Flag.Int("maxMB", 4, "split files larger than the limit")
+	copy.concurrenctFiles = cmdFilerCopy.Flag.Int("c", 8, "concurrent file copy goroutines")
+	copy.concurrenctChunks = cmdFilerCopy.Flag.Int("concurrentChunks", 8, "concurrent chunk copy goroutines for each file")
+	copy.checkSize = cmdFilerCopy.Flag.Bool("check.size", false, "copy when the target file size is different from the source file")
+	copy.verbose = cmdFilerCopy.Flag.Bool("verbose", false, "print out details during copying")
 }
 
-var cmdCopy = &Command{
+var cmdFilerCopy = &Command{
 	UsageLine: "filer.copy file_or_dir1 [file_or_dir2 file_or_dir3] http://localhost:8888/path/to/a/folder/",
 	Short:     "copy one or a list of files to a filer folder",
 	Long: `copy one or a list of files, or batch copy one whole folder recursively, to a filer folder
@@ -154,7 +154,7 @@ func runCopy(cmd *Command, args []string) bool {
 	}
 	copy.ttlSec = int32(ttl.Minutes()) * 60
 
-	if *cmdCopy.IsDebug {
+	if *cmdFilerCopy.IsDebug {
 		grace.SetupProfiling("filer.copy.cpu.pprof", "filer.copy.mem.pprof")
 	}
 
@@ -381,6 +381,9 @@ func (worker *FileCopyWorker) uploadFileAsOne(task FileCopyTask, f *os.File) err
 				if assignResult.Error != "" {
 					return fmt.Errorf("assign volume failure %v: %v", request, assignResult.Error)
 				}
+				if assignResult.Url == "" {
+					return fmt.Errorf("assign volume failure %v: %v", request, assignResult)
+				}
 				return nil
 			})
 		})

weed/command/fuse.go

@@ -62,7 +62,7 @@ func runFuse(cmd *Command, args []string) bool {
 						option.WriteByte(rawArgs[i])
 					}
 
-				// single quote separator read option until next single quote
+					// single quote separator read option until next single quote
 				} else if rawArgs[i] == '\'' {
 					for i++; i < rawArgsLen && rawArgs[i] != '\''; i++ {
 						option.WriteByte(rawArgs[i])
@@ -109,7 +109,7 @@ func runFuse(cmd *Command, args []string) bool {
 		case "filer.path":
 			mountOptions.filerMountRootPath = &parameter.value
 		case "dirAutoCreate":
-			if parsed, err := strconv.ParseBool(parameter.value); err != nil {
+			if parsed, err := strconv.ParseBool(parameter.value); err == nil {
 				mountOptions.dirAutoCreate = &parsed
 			} else {
 				panic(fmt.Errorf("dirAutoCreate: %s", err))
@@ -121,14 +121,14 @@ func runFuse(cmd *Command, args []string) bool {
 		case "disk":
 			mountOptions.diskType = &parameter.value
 		case "ttl":
-			if parsed, err := strconv.ParseInt(parameter.value, 0, 32); err != nil {
+			if parsed, err := strconv.ParseInt(parameter.value, 0, 32); err == nil {
 				intValue := int(parsed)
 				mountOptions.ttlSec = &intValue
 			} else {
 				panic(fmt.Errorf("ttl: %s", err))
 			}
 		case "chunkSizeLimitMB":
-			if parsed, err := strconv.ParseInt(parameter.value, 0, 32); err != nil {
+			if parsed, err := strconv.ParseInt(parameter.value, 0, 32); err == nil {
 				intValue := int(parsed)
 				mountOptions.chunkSizeLimitMB = &intValue
 			} else {
@@ -136,7 +136,7 @@ func runFuse(cmd *Command, args []string) bool {
 			}
 		case "concurrentWriters":
 			i++
-			if parsed, err := strconv.ParseInt(parameter.value, 0, 32); err != nil {
+			if parsed, err := strconv.ParseInt(parameter.value, 0, 32); err == nil {
 				intValue := int(parsed)
 				mountOptions.concurrentWriters = &intValue
 			} else {
@@ -145,7 +145,7 @@ func runFuse(cmd *Command, args []string) bool {
 		case "cacheDir":
 			mountOptions.cacheDir = &parameter.value
 		case "cacheCapacityMB":
-			if parsed, err := strconv.ParseInt(parameter.value, 0, 64); err != nil {
+			if parsed, err := strconv.ParseInt(parameter.value, 0, 64); err == nil {
 				mountOptions.cacheSizeMB = &parsed
 			} else {
 				panic(fmt.Errorf("cacheCapacityMB: %s", err))
@@ -153,7 +153,7 @@ func runFuse(cmd *Command, args []string) bool {
 		case "dataCenter":
 			mountOptions.dataCenter = &parameter.value
 		case "allowOthers":
-			if parsed, err := strconv.ParseBool(parameter.value); err != nil {
+			if parsed, err := strconv.ParseBool(parameter.value); err == nil {
 				mountOptions.allowOthers = &parsed
 			} else {
 				panic(fmt.Errorf("allowOthers: %s", err))
@@ -161,7 +161,7 @@ func runFuse(cmd *Command, args []string) bool {
 		case "umask":
 			mountOptions.umaskString = &parameter.value
 		case "nonempty":
-			if parsed, err := strconv.ParseBool(parameter.value); err != nil {
+			if parsed, err := strconv.ParseBool(parameter.value); err == nil {
 				mountOptions.nonempty = &parsed
 			} else {
 				panic(fmt.Errorf("nonempty: %s", err))
@@ -173,7 +173,7 @@ func runFuse(cmd *Command, args []string) bool {
 		case "map.gid":
 			mountOptions.gidMap = &parameter.value
 		case "readOnly":
-			if parsed, err := strconv.ParseBool(parameter.value); err != nil {
+			if parsed, err := strconv.ParseBool(parameter.value); err == nil {
 				mountOptions.readOnly = &parsed
 			} else {
 				panic(fmt.Errorf("readOnly: %s", err))
@@ -183,7 +183,7 @@ func runFuse(cmd *Command, args []string) bool {
 		case "memprofile":
 			mountMemProfile = &parameter.value
 		case "readRetryTime":
-			if parsed, err := time.ParseDuration(parameter.value); err != nil {
+			if parsed, err := time.ParseDuration(parameter.value); err == nil {
 				mountReadRetryTime = &parsed
 			} else {
 				panic(fmt.Errorf("readRetryTime: %s", err))

weed/command/scaffold.go

@@ -1,6 +1,7 @@
 package command
 
 import (
+	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/command/scaffold"
 	"io/ioutil"
 	"path/filepath"
@@ -56,7 +57,7 @@ func runScaffold(cmd *Command, args []string) bool {
 	if *outputPath != "" {
 		ioutil.WriteFile(filepath.Join(*outputPath, *config+".toml"), []byte(content), 0644)
 	} else {
-		println(content)
+		fmt.Println(content)
 	}
 	return true
 }

weed/command/scaffold/master.toml

@@ -26,6 +26,8 @@ type = "raft"     # Choose [raft|etcd|snowflake] type for storing the file id se
 # when sequencer.type = etcd, set listen client urls of etcd cluster that store file id sequence
 # example : http://127.0.0.1:2379,http://127.0.0.1:2389
 sequencer_etcd_urls = "http://127.0.0.1:2379"
+# when sequencer.type = snowflake, the snowflake id must be different from other masters
+sequencer_snowflake_id = 0     # any number between 1~1023
 
 
 # configurations for tiered cloud storage

weed/command/volume.go

@@ -259,6 +259,7 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v
 
 		// Stop heartbeats
 		if !volumeServer.StopHeartbeat() {
+			volumeServer.SetStopping()
 			glog.V(0).Infof("stop send heartbeat and wait %d seconds until shutdown ...", *v.preStopSeconds)
 			time.Sleep(time.Duration(*v.preStopSeconds) * time.Second)
 		}

weed/filer/abstract_sql/abstract_sql_store.go

@@ -10,7 +10,6 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/util"
 	"strings"
 	"sync"
-	"time"
 )
 
 type SqlGenerator interface {
@@ -262,7 +261,7 @@ func (store *AbstractSqlStore) DeleteEntry(ctx context.Context, fullpath util.Fu
 	return nil
 }
 
-func (store *AbstractSqlStore) DeleteFolderChildren(ctx context.Context, fullpath util.FullPath, limit int64) error {
+func (store *AbstractSqlStore) DeleteFolderChildren(ctx context.Context, fullpath util.FullPath) error {
 
 	db, bucket, shortPath, err := store.getTxOrDB(ctx, fullpath, true)
 	if err != nil {
@@ -280,22 +279,15 @@ func (store *AbstractSqlStore) DeleteFolderChildren(ctx context.Context, fullpat
 		}
 	}
 
-	for {
-		glog.V(4).Infof("delete %s SQL %s %d", string(shortPath), store.GetSqlDeleteFolderChildren(bucket), util.HashStringToLong(string(shortPath)))
-		res, err := db.ExecContext(ctx, store.GetSqlDeleteFolderChildren(bucket), util.HashStringToLong(string(shortPath)), string(shortPath), limit)
-		if err != nil {
-			return fmt.Errorf("deleteFolderChildren %s: %s", fullpath, err)
-		}
+	glog.V(4).Infof("delete %s SQL %s %d", string(shortPath), store.GetSqlDeleteFolderChildren(bucket), util.HashStringToLong(string(shortPath)))
+	res, err := db.ExecContext(ctx, store.GetSqlDeleteFolderChildren(bucket), util.HashStringToLong(string(shortPath)), string(shortPath))
+	if err != nil {
+		return fmt.Errorf("deleteFolderChildren %s: %s", fullpath, err)
+	}
 
-		rowCount, err := res.RowsAffected()
-		if err != nil {
-			return fmt.Errorf("deleteFolderChildren %s but no rows affected: %s", fullpath, err)
-		}
-		if rowCount < limit {
-			break
-		}
-		// to give the Galera Cluster a chance to breath
-		time.Sleep(time.Second)
+	_, err = res.RowsAffected()
+	if err != nil {
+		return fmt.Errorf("deleteFolderChildren %s but no rows affected: %s", fullpath, err)
 	}
 	return nil
 }

weed/filer/cassandra/cassandra_store.go

@@ -124,7 +124,7 @@ func (store *CassandraStore) FindEntry(ctx context.Context, fullpath util.FullPa
 	var data []byte
 	if err := store.session.Query(
 		"SELECT meta FROM filemeta WHERE directory=? AND name=?",
-		dir, name).Consistency(gocql.One).Scan(&data); err != nil {
+		dir, name).Scan(&data); err != nil {
 		if err != gocql.ErrNotFound {
 			return nil, filer_pb.ErrNotFound
 		}
@@ -161,7 +161,7 @@ func (store *CassandraStore) DeleteEntry(ctx context.Context, fullpath util.Full
 	return nil
 }
 
-func (store *CassandraStore) DeleteFolderChildren(ctx context.Context, fullpath util.FullPath, limit int64) error {
+func (store *CassandraStore) DeleteFolderChildren(ctx context.Context, fullpath util.FullPath) error {
 	if _, ok := store.isSuperLargeDirectory(string(fullpath)); ok {
 		return nil // filer.ErrUnsupportedSuperLargeDirectoryListing
 	}

weed/filer/elastic/v7/elastic_store.go

@@ -186,7 +186,7 @@ func (store *ElasticStore) deleteEntry(ctx context.Context, index, id string) (e
 	return fmt.Errorf("delete entry %v.", err)
 }
 
-func (store *ElasticStore) DeleteFolderChildren(ctx context.Context, fullpath weed_util.FullPath, limit int64) (err error) {
+func (store *ElasticStore) DeleteFolderChildren(ctx context.Context, fullpath weed_util.FullPath) (err error) {
 	_, err = store.ListDirectoryEntries(ctx, fullpath, "", false, math.MaxInt32, func(entry *filer.Entry) bool {
 		if err := store.DeleteEntry(ctx, entry.FullPath); err != nil {
 			glog.Errorf("elastic delete %s: %v.", entry.FullPath, err)

weed/filer/entry.go

@@ -42,6 +42,7 @@ type Entry struct {
 	HardLinkId      HardLinkId
 	HardLinkCounter int32
 	Content         []byte
+	Remote          *filer_pb.Entry_Remote
 }
 
 func (entry *Entry) Size() uint64 {
@@ -60,16 +61,34 @@ func (entry *Entry) ToProtoEntry() *filer_pb.Entry {
 	if entry == nil {
 		return nil
 	}
-	return &filer_pb.Entry{
-		Name:            entry.FullPath.Name(),
-		IsDirectory:     entry.IsDirectory(),
-		Attributes:      EntryAttributeToPb(entry),
-		Chunks:          entry.Chunks,
-		Extended:        entry.Extended,
-		HardLinkId:      entry.HardLinkId,
-		HardLinkCounter: entry.HardLinkCounter,
-		Content:         entry.Content,
+	message := &filer_pb.Entry{}
+	message.Name = entry.FullPath.Name()
+	entry.ToExistingProtoEntry(message)
+	return message
+}
+
+func (entry *Entry) ToExistingProtoEntry(message *filer_pb.Entry) {
+	if entry == nil {
+		return
 	}
+	message.IsDirectory = entry.IsDirectory()
+	message.Attributes = EntryAttributeToPb(entry)
+	message.Chunks = entry.Chunks
+	message.Extended = entry.Extended
+	message.HardLinkId = entry.HardLinkId
+	message.HardLinkCounter = entry.HardLinkCounter
+	message.Content = entry.Content
+	message.Remote = entry.Remote
+}
+
+func FromPbEntryToExistingEntry(message *filer_pb.Entry, fsEntry *Entry) {
+	fsEntry.Attr = PbToEntryAttribute(message.Attributes)
+	fsEntry.Chunks = message.Chunks
+	fsEntry.Extended = message.Extended
+	fsEntry.HardLinkId = HardLinkId(message.HardLinkId)
+	fsEntry.HardLinkCounter = message.HardLinkCounter
+	fsEntry.Content = message.Content
+	fsEntry.Remote = message.Remote
 }
 
 func (entry *Entry) ToProtoFullEntry() *filer_pb.FullEntry {
@@ -83,26 +102,11 @@ func (entry *Entry) ToProtoFullEntry() *filer_pb.FullEntry {
 	}
 }
 
-func (entry *Entry) Clone() *Entry {
-	return &Entry{
-		FullPath:        entry.FullPath,
-		Attr:            entry.Attr,
-		Chunks:          entry.Chunks,
-		Extended:        entry.Extended,
-		HardLinkId:      entry.HardLinkId,
-		HardLinkCounter: entry.HardLinkCounter,
-	}
-}
-
 func FromPbEntry(dir string, entry *filer_pb.Entry) *Entry {
-	return &Entry{
-		FullPath:        util.NewFullPath(dir, entry.Name),
-		Attr:            PbToEntryAttribute(entry.Attributes),
-		Chunks:          entry.Chunks,
-		HardLinkId:      HardLinkId(entry.HardLinkId),
-		HardLinkCounter: entry.HardLinkCounter,
-		Content:         entry.Content,
-	}
+	t := &Entry{}
+	t.FullPath = util.NewFullPath(dir, entry.Name)
+	FromPbEntryToExistingEntry(entry, t)
+	return t
 }
 
 func maxUint64(x, y uint64) uint64 {

weed/filer/entry_codec.go

@@ -12,14 +12,8 @@ import (
 )
 
 func (entry *Entry) EncodeAttributesAndChunks() ([]byte, error) {
-	message := &filer_pb.Entry{
-		Attributes:      EntryAttributeToPb(entry),
-		Chunks:          entry.Chunks,
-		Extended:        entry.Extended,
-		HardLinkId:      entry.HardLinkId,
-		HardLinkCounter: entry.HardLinkCounter,
-		Content:         entry.Content,
-	}
+	message := &filer_pb.Entry{}
+	entry.ToExistingProtoEntry(message)
 	return proto.Marshal(message)
 }
 
@@ -31,15 +25,7 @@ func (entry *Entry) DecodeAttributesAndChunks(blob []byte) error {
 		return fmt.Errorf("decoding value blob for %s: %v", entry.FullPath, err)
 	}
 
-	entry.Attr = PbToEntryAttribute(message.Attributes)
-
-	entry.Extended = message.Extended
-
-	entry.Chunks = message.Chunks
-
-	entry.HardLinkId = message.HardLinkId
-	entry.HardLinkCounter = message.HardLinkCounter
-	entry.Content = message.Content
+	FromPbEntryToExistingEntry(message, entry)
 
 	return nil
 }
@@ -129,6 +115,9 @@ func EqualEntry(a, b *Entry) bool {
 	if !bytes.Equal(a.Content, b.Content) {
 		return false
 	}
+	if !proto.Equal(a.Remote, b.Remote) {
+		return false
+	}
 	return true
 }
 

weed/filer/etcd/etcd_store.go

@@ -130,7 +130,7 @@ func (store *EtcdStore) DeleteEntry(ctx context.Context, fullpath weed_util.Full
 	return nil
 }
 
-func (store *EtcdStore) DeleteFolderChildren(ctx context.Context, fullpath weed_util.FullPath, limit int64) (err error) {
+func (store *EtcdStore) DeleteFolderChildren(ctx context.Context, fullpath weed_util.FullPath) (err error) {
 	directoryPrefix := genDirectoryKeyPrefix(fullpath, "")
 
 	if _, err := store.client.Delete(ctx, string(directoryPrefix), clientv3.WithPrefix()); err != nil {

weed/filer/filechunk_manifest.go

@@ -16,7 +16,7 @@ import (
 )
 
 const (
-	ManifestBatch = 1000
+	ManifestBatch = 10000
 )
 
 func HasChunkManifest(chunks []*filer_pb.FileChunk) bool {
@@ -39,9 +39,14 @@ func SeparateManifestChunks(chunks []*filer_pb.FileChunk) (manifestChunks, nonMa
 	return
 }
 
-func ResolveChunkManifest(lookupFileIdFn wdclient.LookupFileIdFunctionType, chunks []*filer_pb.FileChunk) (dataChunks, manifestChunks []*filer_pb.FileChunk, manifestResolveErr error) {
+func ResolveChunkManifest(lookupFileIdFn wdclient.LookupFileIdFunctionType, chunks []*filer_pb.FileChunk, startOffset, stopOffset int64) (dataChunks, manifestChunks []*filer_pb.FileChunk, manifestResolveErr error) {
 	// TODO maybe parallel this
 	for _, chunk := range chunks {
+
+		if max(chunk.Offset, startOffset) >= min(chunk.Offset+int64(chunk.Size), stopOffset) {
+			continue
+		}
+
 		if !chunk.IsChunkManifest {
 			dataChunks = append(dataChunks, chunk)
 			continue
@@ -54,7 +59,7 @@ func ResolveChunkManifest(lookupFileIdFn wdclient.LookupFileIdFunctionType, chun
 
 		manifestChunks = append(manifestChunks, chunk)
 		// recursive
-		dchunks, mchunks, subErr := ResolveChunkManifest(lookupFileIdFn, resolvedChunks)
+		dchunks, mchunks, subErr := ResolveChunkManifest(lookupFileIdFn, resolvedChunks, startOffset, stopOffset)
 		if subErr != nil {
 			return chunks, nil, subErr
 		}

weed/filer/filechunks.go

@@ -53,7 +53,7 @@ func ETagChunks(chunks []*filer_pb.FileChunk) (etag string) {
 
 func CompactFileChunks(lookupFileIdFn wdclient.LookupFileIdFunctionType, chunks []*filer_pb.FileChunk) (compacted, garbage []*filer_pb.FileChunk) {
 
-	visibles, _ := NonOverlappingVisibleIntervals(lookupFileIdFn, chunks)
+	visibles, _ := NonOverlappingVisibleIntervals(lookupFileIdFn, chunks, 0, math.MaxInt64)
 
 	fileIds := make(map[string]bool)
 	for _, interval := range visibles {
@@ -72,11 +72,11 @@ func CompactFileChunks(lookupFileIdFn wdclient.LookupFileIdFunctionType, chunks
 
 func MinusChunks(lookupFileIdFn wdclient.LookupFileIdFunctionType, as, bs []*filer_pb.FileChunk) (delta []*filer_pb.FileChunk, err error) {
 
-	aData, aMeta, aErr := ResolveChunkManifest(lookupFileIdFn, as)
+	aData, aMeta, aErr := ResolveChunkManifest(lookupFileIdFn, as, 0, math.MaxInt64)
 	if aErr != nil {
 		return nil, aErr
 	}
-	bData, bMeta, bErr := ResolveChunkManifest(lookupFileIdFn, bs)
+	bData, bMeta, bErr := ResolveChunkManifest(lookupFileIdFn, bs, 0, math.MaxInt64)
 	if bErr != nil {
 		return nil, bErr
 	}
@@ -117,7 +117,7 @@ func (cv *ChunkView) IsFullChunk() bool {
 
 func ViewFromChunks(lookupFileIdFn wdclient.LookupFileIdFunctionType, chunks []*filer_pb.FileChunk, offset int64, size int64) (views []*ChunkView) {
 
-	visibles, _ := NonOverlappingVisibleIntervals(lookupFileIdFn, chunks)
+	visibles, _ := NonOverlappingVisibleIntervals(lookupFileIdFn, chunks, offset, offset+size)
 
 	return ViewFromVisibleIntervals(visibles, offset, size)
 
@@ -221,9 +221,9 @@ func MergeIntoVisibles(visibles []VisibleInterval, chunk *filer_pb.FileChunk) (n
 
 // NonOverlappingVisibleIntervals translates the file chunk into VisibleInterval in memory
 // If the file chunk content is a chunk manifest
-func NonOverlappingVisibleIntervals(lookupFileIdFn wdclient.LookupFileIdFunctionType, chunks []*filer_pb.FileChunk) (visibles []VisibleInterval, err error) {
+func NonOverlappingVisibleIntervals(lookupFileIdFn wdclient.LookupFileIdFunctionType, chunks []*filer_pb.FileChunk, startOffset int64, stopOffset int64) (visibles []VisibleInterval, err error) {
 
-	chunks, _, err = ResolveChunkManifest(lookupFileIdFn, chunks)
+	chunks, _, err = ResolveChunkManifest(lookupFileIdFn, chunks, startOffset, stopOffset)
 
 	sort.Slice(chunks, func(i, j int) bool {
 		if chunks[i].Mtime == chunks[j].Mtime {

weed/filer/filechunks_test.go

@@ -90,7 +90,7 @@ func TestRandomFileChunksCompact(t *testing.T) {
 		}
 	}
 
-	visibles, _ := NonOverlappingVisibleIntervals(nil, chunks)
+	visibles, _ := NonOverlappingVisibleIntervals(nil, chunks, 0, math.MaxInt64)
 
 	for _, v := range visibles {
 		for x := v.start; x < v.stop; x++ {
@@ -227,7 +227,7 @@ func TestIntervalMerging(t *testing.T) {
 
 	for i, testcase := range testcases {
 		log.Printf("++++++++++ merged test case %d ++++++++++++++++++++", i)
-		intervals, _ := NonOverlappingVisibleIntervals(nil, testcase.Chunks)
+		intervals, _ := NonOverlappingVisibleIntervals(nil, testcase.Chunks, 0, math.MaxInt64)
 		for x, interval := range intervals {
 			log.Printf("test case %d, interval %d, start=%d, stop=%d, fileId=%s",
 				i, x, interval.start, interval.stop, interval.fileId)

weed/filer/filer.go

@@ -19,7 +19,6 @@ import (
 const (
 	LogFlushInterval = time.Minute
 	PaginationSize   = 1024
-	DeleteMaxRows    = 10000
 	FilerStoreId     = "filer.store.id"
 )
 

weed/filer/filer_conf.go

@@ -15,6 +15,7 @@ import (
 const (
 	DirectoryEtcRoot      = "/etc"
 	DirectoryEtcSeaweedFS = "/etc/seaweedfs"
+	DirectoryEtcRemote    = "/etc/remote"
 	FilerConfName         = "filer.conf"
 	IamConfigDirecotry    = "/etc/iam"
 	IamIdentityFile       = "identity.json"

weed/filer/filer_delete_entry.go

@@ -115,7 +115,7 @@ func (f *Filer) doBatchDeleteFolderMetaAndData(ctx context.Context, entry *Entry
 
 	glog.V(3).Infof("deleting directory %v delete %d chunks: %v", entry.FullPath, len(chunks), shouldDeleteChunks)
 
-	if storeDeletionErr := f.Store.DeleteFolderChildren(ctx, entry.FullPath, DeleteMaxRows); storeDeletionErr != nil {
+	if storeDeletionErr := f.Store.DeleteFolderChildren(ctx, entry.FullPath); storeDeletionErr != nil {
 		return nil, nil, fmt.Errorf("filer store delete: %v", storeDeletionErr)
 	}
 

weed/filer/filerstore.go

@@ -25,7 +25,7 @@ type FilerStore interface {
 	// err == filer_pb.ErrNotFound if not found
 	FindEntry(context.Context, util.FullPath) (entry *Entry, err error)
 	DeleteEntry(context.Context, util.FullPath) (err error)
-	DeleteFolderChildren(context.Context, util.FullPath, int64) (err error)
+	DeleteFolderChildren(context.Context, util.FullPath) (err error)
 	ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc ListEachEntryFunc) (lastFileName string, err error)
 	ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc ListEachEntryFunc) (lastFileName string, err error)
 

weed/filer/filerstore_translate_path.go

@@ -100,10 +100,10 @@ func (t *FilerStorePathTranlator) DeleteOneEntry(ctx context.Context, existingEn
 	return t.actualStore.DeleteEntry(ctx, existingEntry.FullPath)
 }
 
-func (t *FilerStorePathTranlator) DeleteFolderChildren(ctx context.Context, fp util.FullPath, limit int64) (err error) {
+func (t *FilerStorePathTranlator) DeleteFolderChildren(ctx context.Context, fp util.FullPath) (err error) {
 	newFullPath := t.translatePath(fp)
 
-	return t.actualStore.DeleteFolderChildren(ctx, newFullPath, limit)
+	return t.actualStore.DeleteFolderChildren(ctx, newFullPath)
 }
 
 func (t *FilerStorePathTranlator) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc ListEachEntryFunc) (string, error) {

weed/filer/filerstore_wrapper.go

@@ -213,7 +213,7 @@ func (fsw *FilerStoreWrapper) DeleteOneEntry(ctx context.Context, existingEntry
 	return actualStore.DeleteEntry(ctx, existingEntry.FullPath)
 }
 
-func (fsw *FilerStoreWrapper) DeleteFolderChildren(ctx context.Context, fp util.FullPath, limit int64) (err error) {
+func (fsw *FilerStoreWrapper) DeleteFolderChildren(ctx context.Context, fp util.FullPath) (err error) {
 	actualStore := fsw.getActualStore(fp + "/")
 	stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "deleteFolderChildren").Inc()
 	start := time.Now()
@@ -222,7 +222,7 @@ func (fsw *FilerStoreWrapper) DeleteFolderChildren(ctx context.Context, fp util.
 	}()
 
 	glog.V(4).Infof("DeleteFolderChildren %s", fp)
-	return actualStore.DeleteFolderChildren(ctx, fp, limit)
+	return actualStore.DeleteFolderChildren(ctx, fp)
 }
 
 func (fsw *FilerStoreWrapper) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc ListEachEntryFunc) (string, error) {

weed/filer/hbase/hbase_store.go

@@ -109,7 +109,7 @@ func (store *HbaseStore) DeleteEntry(ctx context.Context, path util.FullPath) (e
 	return store.doDelete(ctx, store.cfMetaDir, []byte(path))
 }
 
-func (store *HbaseStore) DeleteFolderChildren(ctx context.Context, path util.FullPath, limit int64) (err error) {
+func (store *HbaseStore) DeleteFolderChildren(ctx context.Context, path util.FullPath) (err error) {
 
 	family := map[string][]string{store.cfMetaDir: {COLUMN_NAME}}
 	expectedPrefix := []byte(path.Child(""))

weed/filer/leveldb/leveldb_store.go

@@ -136,7 +136,7 @@ func (store *LevelDBStore) DeleteEntry(ctx context.Context, fullpath weed_util.F
 	return nil
 }
 
-func (store *LevelDBStore) DeleteFolderChildren(ctx context.Context, fullpath weed_util.FullPath, limit int64) (err error) {
+func (store *LevelDBStore) DeleteFolderChildren(ctx context.Context, fullpath weed_util.FullPath) (err error) {
 
 	batch := new(leveldb.Batch)
 

weed/filer/leveldb2/leveldb2_store.go

@@ -5,13 +5,15 @@ import (
 	"context"
 	"crypto/md5"
 	"fmt"
-	"github.com/syndtr/goleveldb/leveldb"
-	leveldb_errors "github.com/syndtr/goleveldb/leveldb/errors"
-	"github.com/syndtr/goleveldb/leveldb/opt"
-	leveldb_util "github.com/syndtr/goleveldb/leveldb/util"
 	"io"
 	"os"
 
+	"github.com/syndtr/goleveldb/leveldb"
+	leveldb_errors "github.com/syndtr/goleveldb/leveldb/errors"
+	"github.com/syndtr/goleveldb/leveldb/filter"
+	"github.com/syndtr/goleveldb/leveldb/opt"
+	leveldb_util "github.com/syndtr/goleveldb/leveldb/util"
+
 	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
@@ -47,6 +49,7 @@ func (store *LevelDB2Store) initialize(dir string, dbCount int) (err error) {
 		BlockCacheCapacity:            32 * 1024 * 1024, // default value is 8MiB
 		WriteBuffer:                   16 * 1024 * 1024, // default value is 4MiB
 		CompactionTableSizeMultiplier: 4,
+		Filter:                        filter.NewBloomFilter(8), // false positive rate 0.02
 	}
 
 	for d := 0; d < dbCount; d++ {
@@ -144,7 +147,7 @@ func (store *LevelDB2Store) DeleteEntry(ctx context.Context, fullpath weed_util.
 	return nil
 }
 
-func (store *LevelDB2Store) DeleteFolderChildren(ctx context.Context, fullpath weed_util.FullPath, limit int64) (err error) {
+func (store *LevelDB2Store) DeleteFolderChildren(ctx context.Context, fullpath weed_util.FullPath) (err error) {
 	directoryPrefix, partitionId := genDirectoryKeyPrefix(fullpath, "", store.dbCount)
 
 	batch := new(leveldb.Batch)

weed/filer/leveldb3/leveldb3_store.go

@@ -5,15 +5,17 @@ import (
 	"context"
 	"crypto/md5"
 	"fmt"
-	"github.com/syndtr/goleveldb/leveldb"
-	leveldb_errors "github.com/syndtr/goleveldb/leveldb/errors"
-	"github.com/syndtr/goleveldb/leveldb/opt"
-	leveldb_util "github.com/syndtr/goleveldb/leveldb/util"
 	"io"
 	"os"
 	"strings"
 	"sync"
 
+	"github.com/syndtr/goleveldb/leveldb"
+	leveldb_errors "github.com/syndtr/goleveldb/leveldb/errors"
+	"github.com/syndtr/goleveldb/leveldb/filter"
+	"github.com/syndtr/goleveldb/leveldb/opt"
+	leveldb_util "github.com/syndtr/goleveldb/leveldb/util"
+
 	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
@@ -62,17 +64,19 @@ func (store *LevelDB3Store) initialize(dir string) (err error) {
 }
 
 func (store *LevelDB3Store) loadDB(name string) (*leveldb.DB, error) {
-
+	bloom := filter.NewBloomFilter(8) // false positive rate 0.02
 	opts := &opt.Options{
 		BlockCacheCapacity:            32 * 1024 * 1024, // default value is 8MiB
 		WriteBuffer:                   16 * 1024 * 1024, // default value is 4MiB
 		CompactionTableSizeMultiplier: 4,
+		Filter:                        bloom,
 	}
 	if name != DEFAULT {
 		opts = &opt.Options{
 			BlockCacheCapacity:            4 * 1024 * 1024, // default value is 8MiB
 			WriteBuffer:                   2 * 1024 * 1024, // default value is 4MiB
 			CompactionTableSizeMultiplier: 4,
+			Filter:                        bloom,
 		}
 	}
 
@@ -245,7 +249,7 @@ func (store *LevelDB3Store) DeleteEntry(ctx context.Context, fullpath weed_util.
 	return nil
 }
 
-func (store *LevelDB3Store) DeleteFolderChildren(ctx context.Context, fullpath weed_util.FullPath, limit int64) (err error) {
+func (store *LevelDB3Store) DeleteFolderChildren(ctx context.Context, fullpath weed_util.FullPath) (err error) {
 
 	db, bucket, shortPath, err := store.findDB(fullpath, true)
 	if err != nil {

weed/filer/mongodb/mongodb_store.go

@@ -167,7 +167,7 @@ func (store *MongodbStore) DeleteEntry(ctx context.Context, fullpath util.FullPa
 	return nil
 }
 
-func (store *MongodbStore) DeleteFolderChildren(ctx context.Context, fullpath util.FullPath, limit int64) error {
+func (store *MongodbStore) DeleteFolderChildren(ctx context.Context, fullpath util.FullPath) error {
 
 	where := bson.M{"directory": fullpath}
 	_, err := store.connect.Database(store.database).Collection(store.collectionName).DeleteMany(ctx, where)

weed/filer/mysql/mysql_sql_gen.go

@@ -38,7 +38,7 @@ func (gen *SqlGenMysql) GetSqlDelete(tableName string) string {
 }
 
 func (gen *SqlGenMysql) GetSqlDeleteFolderChildren(tableName string) string {
-	return fmt.Sprintf("DELETE FROM `%s` WHERE dirhash=? AND directory=? LIMIT ?", tableName)
+	return fmt.Sprintf("DELETE FROM `%s` WHERE dirhash=? AND directory=?", tableName)
 }
 
 func (gen *SqlGenMysql) GetSqlListExclusive(tableName string) string {

weed/filer/redis/universal_redis_store.go

@@ -107,7 +107,7 @@ func (store *UniversalRedisStore) DeleteEntry(ctx context.Context, fullpath util
 	return nil
 }
 
-func (store *UniversalRedisStore) DeleteFolderChildren(ctx context.Context, fullpath util.FullPath, limit int64) (err error) {
+func (store *UniversalRedisStore) DeleteFolderChildren(ctx context.Context, fullpath util.FullPath) (err error) {
 
 	members, err := store.Client.SMembers(ctx, genDirectoryListKey(string(fullpath))).Result()
 	if err != nil {

weed/filer/redis2/universal_redis_store.go

@@ -127,7 +127,7 @@ func (store *UniversalRedis2Store) DeleteEntry(ctx context.Context, fullpath uti
 	return nil
 }
 
-func (store *UniversalRedis2Store) DeleteFolderChildren(ctx context.Context, fullpath util.FullPath, limit int64) (err error) {
+func (store *UniversalRedis2Store) DeleteFolderChildren(ctx context.Context, fullpath util.FullPath) (err error) {
 
 	if store.isSuperLargeDirectory(string(fullpath)) {
 		return nil

weed/filer/rocksdb/rocksdb_store.go

@@ -24,18 +24,21 @@ func init() {
 
 type options struct {
 	opt *gorocksdb.Options
+	bto *gorocksdb.BlockBasedTableOptions
 	ro  *gorocksdb.ReadOptions
 	wo  *gorocksdb.WriteOptions
 }
 
 func (opt *options) init() {
 	opt.opt = gorocksdb.NewDefaultOptions()
+	opt.bto = gorocksdb.NewDefaultBlockBasedTableOptions()
 	opt.ro = gorocksdb.NewDefaultReadOptions()
 	opt.wo = gorocksdb.NewDefaultWriteOptions()
 }
 
 func (opt *options) close() {
 	opt.opt.Destroy()
+	opt.bto.Destroy()
 	opt.ro.Destroy()
 	opt.wo.Destroy()
 }
@@ -69,6 +72,11 @@ func (store *RocksDBStore) initialize(dir string) (err error) {
 	store.opt.SetCompactionFilter(NewTTLFilter())
 	// store.opt.SetMaxBackgroundCompactions(2)
 
+	// https://github.com/tecbot/gorocksdb/issues/132
+	store.bto.SetFilterPolicy(gorocksdb.NewBloomFilterFull(8))
+	store.opt.SetBlockBasedTableFactory(store.bto)
+	// store.opt.EnableStatistics()
+
 	store.db, err = gorocksdb.OpenDb(store.opt, dir)
 
 	return
@@ -148,7 +156,7 @@ func (store *RocksDBStore) DeleteEntry(ctx context.Context, fullpath weed_util.F
 	return nil
 }
 
-func (store *RocksDBStore) DeleteFolderChildren(ctx context.Context, fullpath weed_util.FullPath, limit int64) (err error) {
+func (store *RocksDBStore) DeleteFolderChildren(ctx context.Context, fullpath weed_util.FullPath) (err error) {
 	directoryPrefix := genDirectoryKeyPrefix(fullpath, "")
 
 	batch := gorocksdb.NewWriteBatch()

weed/filesys/dir_rename.go

@@ -63,7 +63,6 @@ func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDirector
 		return fuse.EIO
 	}
 
-
 	return nil
 }
 

weed/filesys/filehandle.go

@@ -130,7 +130,7 @@ func (fh *FileHandle) readFromChunks(buff []byte, offset int64) (int64, error) {
 
 	var chunkResolveErr error
 	if fh.entryViewCache == nil {
-		fh.entryViewCache, chunkResolveErr = filer.NonOverlappingVisibleIntervals(fh.f.wfs.LookupFn(), entry.Chunks)
+		fh.entryViewCache, chunkResolveErr = filer.NonOverlappingVisibleIntervals(fh.f.wfs.LookupFn(), entry.Chunks, 0, math.MaxInt64)
 		if chunkResolveErr != nil {
 			return 0, fmt.Errorf("fail to resolve chunk manifest: %v", chunkResolveErr)
 		}

weed/filesys/meta_cache/meta_cache_init.go

@@ -43,5 +43,5 @@ func EnsureVisited(mc *MetaCache, client filer_pb.FilerClient, dirPath util.Full
 }
 
 func IsHiddenSystemEntry(dir, name string) bool {
-	return dir == "/" && name == "topics"
+	return dir == "/" && (name == "topics" || name == "etc")
 }

weed/images/resizing.go

@@ -11,6 +11,8 @@ import (
 	"github.com/disintegration/imaging"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
+
+	_ "golang.org/x/image/webp"
 )
 
 func Resized(ext string, read io.ReadSeeker, width, height int, mode string) (resized io.ReadSeeker, w int, h int) {
@@ -47,6 +49,9 @@ func Resized(ext string, read io.ReadSeeker, width, height int, mode string) (re
 			jpeg.Encode(&buf, dstImage, nil)
 		case ".gif":
 			gif.Encode(&buf, dstImage, nil)
+		case ".webp":
+			// Webp does not have golang encoder.
+			png.Encode(&buf, dstImage)
 		}
 		return bytes.NewReader(buf.Bytes()), dstImage.Bounds().Dx(), dstImage.Bounds().Dy()
 	} else {

weed/images/resizing_test.go

@@ -0,0 +1,23 @@
+package images
+
+import (
+	"bytes"
+	"io/ioutil"
+	"os"
+	"testing"
+)
+
+func TestResizing(t *testing.T) {
+	fname := "sample2.webp"
+
+	dat, _ := ioutil.ReadFile(fname)
+
+	resized, _, _ := Resized(".webp", bytes.NewReader(dat), 100, 30, "")
+	buf := new(bytes.Buffer)
+	buf.ReadFrom(resized)
+
+	ioutil.WriteFile("resized1.png", buf.Bytes(), 0644)
+
+	os.Remove("resized1.png")
+
+}

weed/operation/assign_file_id.go

@@ -73,6 +73,10 @@ func Assign(masterFn GetMasterFn, grpcDialOption grpc.DialOption, primaryRequest
 			ret.Error = resp.Error
 			ret.Auth = security.EncodedJwt(resp.Auth)
 
+			if resp.Error != "" {
+				return fmt.Errorf("assignRequest: %v", resp.Error)
+			}
+
 			return nil
 
 		})

weed/pb/filer.proto

@@ -101,6 +101,13 @@ message Entry {
     bytes hard_link_id = 7;
     int32 hard_link_counter = 8; // only exists in hard link meta data
     bytes content = 9; // if not empty, the file content
+
+    message Remote {
+        int64 last_modified_at = 1;
+        int64 size = 2;
+        string e_tag = 3;
+    }
+    Remote remote = 10;
 }
 
 message FullEntry {
@@ -367,3 +374,12 @@ message FilerConf {
     }
     repeated PathConf locations = 2;
 }
+
+message RemoteConf {
+    string type = 1;
+    string name = 2;
+    string s3_access_key = 4;
+    string s3_secret_key = 5;
+    string s3_region = 6;
+    string s3_endpoint = 7;
+}

weed/replication/sink/filersink/filer_sink.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/pb"
 	"github.com/chrislusf/seaweedfs/weed/wdclient"
+	"math"
 
 	"google.golang.org/grpc"
 
@@ -133,6 +134,7 @@ func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry, signatures [
 				Attributes:  entry.Attributes,
 				Chunks:      replicatedChunks,
 				Content:     entry.Content,
+				Remote:      entry.Remote,
 			},
 			IsFromOtherCluster: true,
 			Signatures:         signatures,
@@ -227,11 +229,11 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParent
 
 }
 func compareChunks(lookupFileIdFn wdclient.LookupFileIdFunctionType, oldEntry, newEntry *filer_pb.Entry) (deletedChunks, newChunks []*filer_pb.FileChunk, err error) {
-	aData, aMeta, aErr := filer.ResolveChunkManifest(lookupFileIdFn, oldEntry.Chunks)
+	aData, aMeta, aErr := filer.ResolveChunkManifest(lookupFileIdFn, oldEntry.Chunks, 0, math.MaxInt64)
 	if aErr != nil {
 		return nil, nil, aErr
 	}
-	bData, bMeta, bErr := filer.ResolveChunkManifest(lookupFileIdFn, newEntry.Chunks)
+	bData, bMeta, bErr := filer.ResolveChunkManifest(lookupFileIdFn, newEntry.Chunks, 0, math.MaxInt64)
 	if bErr != nil {
 		return nil, nil, bErr
 	}

weed/s3api/filer_multipart.go

@@ -71,7 +71,7 @@ func (s3a *S3ApiServer) completeMultipartUpload(input *s3.CompleteMultipartUploa
 		return nil, s3err.ErrNoSuchUpload
 	}
 
-	pentry, err := s3a.getEntry(s3a.genUploadsFolder(*input.Bucket),*input.UploadId)
+	pentry, err := s3a.getEntry(s3a.genUploadsFolder(*input.Bucket), *input.UploadId)
 	if err != nil {
 		glog.Errorf("completeMultipartUpload %s %s error: %v", *input.Bucket, *input.UploadId, err)
 		return nil, s3err.ErrNoSuchUpload
@@ -112,11 +112,11 @@ func (s3a *S3ApiServer) completeMultipartUpload(input *s3.CompleteMultipartUploa
 		dirName = dirName[:len(dirName)-1]
 	}
 
-	err = s3a.mkFile(dirName, entryName, finalParts,func(entry *filer_pb.Entry) {
+	err = s3a.mkFile(dirName, entryName, finalParts, func(entry *filer_pb.Entry) {
 		if entry.Extended == nil {
 			entry.Extended = make(map[string][]byte)
 		}
-		for k,v := range pentry.Extended{
+		for k, v := range pentry.Extended {
 			if k != "key" {
 				entry.Extended[k] = v
 			}

weed/s3api/filer_util.go

@@ -17,7 +17,7 @@ func (s3a *S3ApiServer) mkdir(parentDirectoryPath string, dirName string, fn fun
 
 func (s3a *S3ApiServer) mkFile(parentDirectoryPath string, fileName string, chunks []*filer_pb.FileChunk, fn func(entry *filer_pb.Entry)) error {
 
-	return filer_pb.MkFile(s3a, parentDirectoryPath, fileName, chunks,fn)
+	return filer_pb.MkFile(s3a, parentDirectoryPath, fileName, chunks, fn)
 
 }
 

weed/s3api/s3api_object_multipart_handlers.go

@@ -38,7 +38,7 @@ func (s3a *S3ApiServer) NewMultipartUploadHandler(w http.ResponseWriter, r *http
 
 	response, errCode := s3a.createMultipartUpload(createMultipartUploadInput)
 
-	glog.V(2).Info("NewMultipartUploadHandler", s3err.EncodeXMLResponse(response), errCode)
+	glog.V(2).Info("NewMultipartUploadHandler", string(s3err.EncodeXMLResponse(response)), errCode)
 
 	if errCode != s3err.ErrNone {
 		s3err.WriteErrorResponse(w, errCode, r)
@@ -62,7 +62,7 @@ func (s3a *S3ApiServer) CompleteMultipartUploadHandler(w http.ResponseWriter, r
 		UploadId: aws.String(uploadID),
 	})
 
-	glog.V(2).Info("CompleteMultipartUploadHandler", s3err.EncodeXMLResponse(response), errCode)
+	glog.V(2).Info("CompleteMultipartUploadHandler", string(s3err.EncodeXMLResponse(response)), errCode)
 
 	if errCode != s3err.ErrNone {
 		s3err.WriteErrorResponse(w, errCode, r)
@@ -91,7 +91,7 @@ func (s3a *S3ApiServer) AbortMultipartUploadHandler(w http.ResponseWriter, r *ht
 		return
 	}
 
-	glog.V(2).Info("AbortMultipartUploadHandler", s3err.EncodeXMLResponse(response))
+	glog.V(2).Info("AbortMultipartUploadHandler", string(s3err.EncodeXMLResponse(response)))
 
 	writeSuccessResponseXML(w, response)
 
@@ -124,7 +124,7 @@ func (s3a *S3ApiServer) ListMultipartUploadsHandler(w http.ResponseWriter, r *ht
 		UploadIdMarker: aws.String(uploadIDMarker),
 	})
 
-	glog.V(2).Info("ListMultipartUploadsHandler", s3err.EncodeXMLResponse(response), errCode)
+	glog.V(2).Info("ListMultipartUploadsHandler", string(s3err.EncodeXMLResponse(response)), errCode)
 
 	if errCode != s3err.ErrNone {
 		s3err.WriteErrorResponse(w, errCode, r)
@@ -158,7 +158,7 @@ func (s3a *S3ApiServer) ListObjectPartsHandler(w http.ResponseWriter, r *http.Re
 		UploadId:         aws.String(uploadID),
 	})
 
-	glog.V(2).Info("ListObjectPartsHandler", s3err.EncodeXMLResponse(response), errCode)
+	glog.V(2).Info("ListObjectPartsHandler", string(s3err.EncodeXMLResponse(response)), errCode)
 
 	if errCode != s3err.ErrNone {
 		s3err.WriteErrorResponse(w, errCode, r)

weed/security/jwt.go

@@ -7,7 +7,7 @@ import (
 	"time"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
-	jwt "github.com/dgrijalva/jwt-go"
+	"github.com/golang-jwt/jwt"
 )
 
 type EncodedJwt string

weed/sequence/snowflake_sequencer.go

@@ -13,8 +13,11 @@ type SnowflakeSequencer struct {
 	node *snowflake.Node
 }
 
-func NewSnowflakeSequencer(nodeid string) (*SnowflakeSequencer, error) {
+func NewSnowflakeSequencer(nodeid string, snowflakeId int) (*SnowflakeSequencer, error) {
 	nodeid_hash := hash(nodeid) & 0x3ff
+	if snowflakeId != 0 {
+		nodeid_hash = uint32(snowflakeId)
+	}
 	glog.V(0).Infof("use snowflake seq id generator, nodeid:%s hex_of_nodeid: %x", nodeid, nodeid_hash)
 	node, err := snowflake.NewNode(int64(nodeid_hash))
 	if err != nil {

weed/server/filer_grpc_server.go

@@ -31,16 +31,7 @@ func (fs *FilerServer) LookupDirectoryEntry(ctx context.Context, req *filer_pb.L
 	}
 
 	return &filer_pb.LookupDirectoryEntryResponse{
-		Entry: &filer_pb.Entry{
-			Name:            req.Name,
-			IsDirectory:     entry.IsDirectory(),
-			Attributes:      filer.EntryAttributeToPb(entry),
-			Chunks:          entry.Chunks,
-			Extended:        entry.Extended,
-			HardLinkId:      entry.HardLinkId,
-			HardLinkCounter: entry.HardLinkCounter,
-			Content:         entry.Content,
-		},
+		Entry: entry.ToProtoEntry(),
 	}, nil
 }
 
@@ -66,16 +57,7 @@ func (fs *FilerServer) ListEntries(req *filer_pb.ListEntriesRequest, stream file
 		lastFileName, listErr = fs.filer.StreamListDirectoryEntries(stream.Context(), util.FullPath(req.Directory), lastFileName, includeLastFile, int64(paginationLimit), req.Prefix, "", "", func(entry *filer.Entry) bool {
 			hasEntries = true
 			if err = stream.Send(&filer_pb.ListEntriesResponse{
-				Entry: &filer_pb.Entry{
-					Name:            entry.Name(),
-					IsDirectory:     entry.IsDirectory(),
-					Chunks:          entry.Chunks,
-					Attributes:      filer.EntryAttributeToPb(entry),
-					Extended:        entry.Extended,
-					HardLinkId:      entry.HardLinkId,
-					HardLinkCounter: entry.HardLinkCounter,
-					Content:         entry.Content,
-				},
+				Entry: entry.ToProtoEntry(),
 			}); err != nil {
 				return false
 			}
@@ -161,15 +143,10 @@ func (fs *FilerServer) CreateEntry(ctx context.Context, req *filer_pb.CreateEntr
 		return &filer_pb.CreateEntryResponse{}, fmt.Errorf("CreateEntry cleanupChunks %s %s: %v", req.Directory, req.Entry.Name, err2)
 	}
 
-	createErr := fs.filer.CreateEntry(ctx, &filer.Entry{
-		FullPath:        util.JoinPath(req.Directory, req.Entry.Name),
-		Attr:            filer.PbToEntryAttribute(req.Entry.Attributes),
-		Chunks:          chunks,
-		Extended:        req.Entry.Extended,
-		HardLinkId:      filer.HardLinkId(req.Entry.HardLinkId),
-		HardLinkCounter: req.Entry.HardLinkCounter,
-		Content:         req.Entry.Content,
-	}, req.OExcl, req.IsFromOtherCluster, req.Signatures)
+	newEntry := filer.FromPbEntry(req.Directory, req.Entry)
+	newEntry.Chunks = chunks
+
+	createErr := fs.filer.CreateEntry(ctx, newEntry, req.OExcl, req.IsFromOtherCluster, req.Signatures)
 
 	if createErr == nil {
 		fs.filer.DeleteChunks(garbage)
@@ -196,35 +173,8 @@ func (fs *FilerServer) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntr
 		return &filer_pb.UpdateEntryResponse{}, fmt.Errorf("UpdateEntry cleanupChunks %s: %v", fullpath, err2)
 	}
 
-	newEntry := &filer.Entry{
-		FullPath:        util.JoinPath(req.Directory, req.Entry.Name),
-		Attr:            entry.Attr,
-		Extended:        req.Entry.Extended,
-		Chunks:          chunks,
-		HardLinkId:      filer.HardLinkId(req.Entry.HardLinkId),
-		HardLinkCounter: req.Entry.HardLinkCounter,
-		Content:         req.Entry.Content,
-	}
-
-	glog.V(3).Infof("updating %s: %+v, chunks %d: %v => %+v, chunks %d: %v, extended: %v => %v",
-		fullpath, entry.Attr, len(entry.Chunks), entry.Chunks,
-		req.Entry.Attributes, len(req.Entry.Chunks), req.Entry.Chunks,
-		entry.Extended, req.Entry.Extended)
-
-	if req.Entry.Attributes != nil {
-		if req.Entry.Attributes.Mtime != 0 {
-			newEntry.Attr.Mtime = time.Unix(req.Entry.Attributes.Mtime, 0)
-		}
-		if req.Entry.Attributes.FileMode != 0 {
-			newEntry.Attr.Mode = os.FileMode(req.Entry.Attributes.FileMode)
-		}
-		newEntry.Attr.Uid = req.Entry.Attributes.Uid
-		newEntry.Attr.Gid = req.Entry.Attributes.Gid
-		newEntry.Attr.Mime = req.Entry.Attributes.Mime
-		newEntry.Attr.UserName = req.Entry.Attributes.UserName
-		newEntry.Attr.GroupNames = req.Entry.Attributes.GroupName
-
-	}
+	newEntry := filer.FromPbEntry(req.Directory, req.Entry)
+	newEntry.Chunks = chunks
 
 	if filer.EqualEntry(entry, newEntry) {
 		return &filer_pb.UpdateEntryResponse{}, err

weed/server/filer_server_handlers.go

@@ -35,11 +35,11 @@ func (fs *FilerServer) filerHandler(w http.ResponseWriter, r *http.Request) {
 	switch r.Method {
 	case "GET":
 		stats.FilerRequestCounter.WithLabelValues("get").Inc()
-		fs.GetOrHeadHandler(w, r, true)
+		fs.GetOrHeadHandler(w, r)
 		stats.FilerRequestHistogram.WithLabelValues("get").Observe(time.Since(start).Seconds())
 	case "HEAD":
 		stats.FilerRequestCounter.WithLabelValues("head").Inc()
-		fs.GetOrHeadHandler(w, r, false)
+		fs.GetOrHeadHandler(w, r)
 		stats.FilerRequestHistogram.WithLabelValues("head").Observe(time.Since(start).Seconds())
 	case "DELETE":
 		stats.FilerRequestCounter.WithLabelValues("delete").Inc()
@@ -95,11 +95,11 @@ func (fs *FilerServer) readonlyFilerHandler(w http.ResponseWriter, r *http.Reque
 	switch r.Method {
 	case "GET":
 		stats.FilerRequestCounter.WithLabelValues("get").Inc()
-		fs.GetOrHeadHandler(w, r, true)
+		fs.GetOrHeadHandler(w, r)
 		stats.FilerRequestHistogram.WithLabelValues("get").Observe(time.Since(start).Seconds())
 	case "HEAD":
 		stats.FilerRequestCounter.WithLabelValues("head").Inc()
-		fs.GetOrHeadHandler(w, r, false)
+		fs.GetOrHeadHandler(w, r)
 		stats.FilerRequestHistogram.WithLabelValues("head").Observe(time.Since(start).Seconds())
 	case "OPTIONS":
 		stats.FilerRequestCounter.WithLabelValues("options").Inc()

weed/server/filer_server_handlers_read.go

@@ -21,7 +21,7 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
-func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request, isGetMethod bool) {
+func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) {
 
 	path := r.URL.Path
 	isForDirectory := strings.HasSuffix(path, "/")

weed/server/filer_server_handlers_write_upload.go

@@ -28,19 +28,14 @@ var bufPool = sync.Pool{
 	},
 }
 
-func (fs *FilerServer) uploadReaderToChunks(w http.ResponseWriter, r *http.Request, reader io.Reader, chunkSize int32, fileName, contentType string, contentLength int64, so *operation.StorageOption) ([]*filer_pb.FileChunk, hash.Hash, int64, error, []byte) {
+func (fs *FilerServer) uploadReaderToChunks(w http.ResponseWriter, r *http.Request, reader io.Reader, chunkSize int32, fileName, contentType string, contentLength int64, so *operation.StorageOption) (fileChunks []*filer_pb.FileChunk, md5Hash hash.Hash, chunkOffset int64, uploadErr error, smallContent []byte) {
 
-	md5Hash := md5.New()
+	md5Hash = md5.New()
 	var partReader = ioutil.NopCloser(io.TeeReader(reader, md5Hash))
 
-	chunkOffset := int64(0)
-	var smallContent []byte
-	var uploadErr error
-
 	var wg sync.WaitGroup
 	var bytesBufferCounter int64
 	bytesBufferLimitCond := sync.NewCond(new(sync.Mutex))
-	var fileChunks []*filer_pb.FileChunk
 	var fileChunksLock sync.Mutex
 	for {
 
@@ -67,7 +62,7 @@ func (fs *FilerServer) uploadReaderToChunks(w http.ResponseWriter, r *http.Reque
 			bufPool.Put(bytesBuffer)
 			atomic.AddInt64(&bytesBufferCounter, -1)
 			bytesBufferLimitCond.Signal()
-			return nil, md5Hash, 0, err, nil
+			break
 		}
 		if chunkOffset == 0 && !isAppend(r) {
 			if dataSize < fs.option.SaveToFilerLimit || strings.HasPrefix(r.URL.Path, filer.DirectoryEtcRoot) {

weed/server/master_server.go

@@ -28,6 +28,7 @@ import (
 const (
 	SequencerType     = "master.sequencer.type"
 	SequencerEtcdUrls = "master.sequencer.sequencer_etcd_urls"
+	SequencerSnowflakeId = "master.sequencer.sequencer_snowflake_id"
 )
 
 type MasterOption struct {
@@ -293,7 +294,8 @@ func (ms *MasterServer) createSequencer(option *MasterOption) sequence.Sequencer
 		}
 	case "snowflake":
 		var err error
-		seq, err = sequence.NewSnowflakeSequencer(fmt.Sprintf("%s:%d", option.Host, option.Port))
+		snowflakeId := v.GetInt(SequencerSnowflakeId)
+		seq, err = sequence.NewSnowflakeSequencer(fmt.Sprintf("%s:%d", option.Host, option.Port), snowflakeId)
 		if err != nil {
 			glog.Error(err)
 			seq = nil

weed/server/volume_server.go

@@ -112,6 +112,11 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
 	return vs
 }
 
+func (vs *VolumeServer) SetStopping() {
+	glog.V(0).Infoln("Stopping volume server...")
+	vs.store.SetStopping()
+}
+
 func (vs *VolumeServer) Shutdown() {
 	glog.V(0).Infoln("Shutting down volume server...")
 	vs.store.Close()

weed/server/volume_server_handlers_read.go

@@ -264,7 +264,7 @@ func conditionallyResizeImages(originalDataReaderSeeker io.ReadSeeker, ext strin
 }
 
 func shouldResizeImages(ext string, r *http.Request) (width, height int, mode string, shouldResize bool) {
-	if ext == ".png" || ext == ".jpg" || ext == ".jpeg" || ext == ".gif" {
+	if ext == ".png" || ext == ".jpg" || ext == ".jpeg" || ext == ".gif" || ext == ".webp" {
 		if r.FormValue("width") != "" {
 			width, _ = strconv.Atoi(r.FormValue("width"))
 		}

weed/server/webdav_server.go

@@ -532,7 +532,7 @@ func (f *WebDavFile) Read(p []byte) (readSize int, err error) {
 		return 0, io.EOF
 	}
 	if f.entryViewCache == nil {
-		f.entryViewCache, _ = filer.NonOverlappingVisibleIntervals(filer.LookupFn(f.fs), f.entry.Chunks)
+		f.entryViewCache, _ = filer.NonOverlappingVisibleIntervals(filer.LookupFn(f.fs), f.entry.Chunks, 0, math.MaxInt64)
 		f.reader = nil
 	}
 	if f.reader == nil {

weed/shell/command_fs_configure.go

@@ -62,20 +62,11 @@ func (c *commandFsConfigure) Do(args []string, commandEnv *CommandEnv, writer io
 		return nil
 	}
 
-	var buf bytes.Buffer
-	if err = commandEnv.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
-		return filer.ReadEntry(commandEnv.MasterClient, client, filer.DirectoryEtcSeaweedFS, filer.FilerConfName, &buf)
-	}); err != nil && err != filer_pb.ErrNotFound {
+	fc, err := readFilerConf(commandEnv)
+	if err != nil {
 		return err
 	}
 
-	fc := filer.NewFilerConf()
-	if buf.Len() > 0 {
-		if err = fc.LoadFromBytes(buf.Bytes()); err != nil {
-			return err
-		}
-	}
-
 	if *locationPrefix != "" {
 		locConf := &filer_pb.FilerConf_PathConf{
 			LocationPrefix:    *locationPrefix,
@@ -112,16 +103,16 @@ func (c *commandFsConfigure) Do(args []string, commandEnv *CommandEnv, writer io
 		}
 	}
 
-	buf.Reset()
-	fc.ToText(&buf)
+	var buf2 bytes.Buffer
+	fc.ToText(&buf2)
 
-	fmt.Fprintf(writer, string(buf.Bytes()))
+	fmt.Fprintf(writer, string(buf2.Bytes()))
 	fmt.Fprintln(writer)
 
 	if *apply {
 
 		if err = commandEnv.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
-			return filer.SaveInsideFiler(client, filer.DirectoryEtcSeaweedFS, filer.FilerConfName, buf.Bytes())
+			return filer.SaveInsideFiler(client, filer.DirectoryEtcSeaweedFS, filer.FilerConfName, buf2.Bytes())
 		}); err != nil && err != filer_pb.ErrNotFound {
 			return err
 		}
@@ -131,3 +122,20 @@ func (c *commandFsConfigure) Do(args []string, commandEnv *CommandEnv, writer io
 	return nil
 
 }
+
+func readFilerConf(commandEnv *CommandEnv) (*filer.FilerConf, error) {
+	var buf bytes.Buffer
+	if err := commandEnv.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+		return filer.ReadEntry(commandEnv.MasterClient, client, filer.DirectoryEtcSeaweedFS, filer.FilerConfName, &buf)
+	}); err != nil && err != filer_pb.ErrNotFound {
+		return nil, fmt.Errorf("read %s/%s: %v", filer.DirectoryEtcSeaweedFS, filer.FilerConfName, err)
+	}
+
+	fc := filer.NewFilerConf()
+	if buf.Len() > 0 {
+		if err := fc.LoadFromBytes(buf.Bytes()); err != nil {
+			return nil, fmt.Errorf("parse %s/%s: %v", filer.DirectoryEtcSeaweedFS, filer.FilerConfName, err)
+		}
+	}
+	return fc, nil
+}

weed/shell/command_fs_mkdir.go

@@ -0,0 +1,54 @@
+package shell
+
+import (
+	"context"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"io"
+	"os"
+)
+
+func init() {
+	Commands = append(Commands, &commandFsMkdir{})
+}
+
+type commandFsMkdir struct {
+}
+
+func (c *commandFsMkdir) Name() string {
+	return "fs.mkdir"
+}
+
+func (c *commandFsMkdir) Help() string {
+	return `create a directory
+
+	fs.mkdir path/to/dir
+`
+}
+
+func (c *commandFsMkdir) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
+
+	path, err := commandEnv.parseUrl(findInputDirectory(args))
+	if err != nil {
+		return err
+	}
+
+	dir, name := util.FullPath(path).DirAndName()
+
+	err = commandEnv.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+
+		_, createErr := client.CreateEntry(context.Background(), &filer_pb.CreateEntryRequest{
+			Directory: dir,
+			Entry: &filer_pb.Entry{
+				Name:        name,
+				IsDirectory: true,
+				Attributes: &filer_pb.FuseAttributes{
+					FileMode: uint32(0777 | os.ModeDir),
+				},
+			},
+		})
+		return createErr
+	})
+
+	return
+}

weed/shell/command_fs_rm.go

@@ -0,0 +1,100 @@
+package shell
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"strings"
+
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/chrislusf/seaweedfs/weed/util"
+)
+
+func init() {
+	Commands = append(Commands, &commandFsRm{})
+}
+
+type commandFsRm struct {
+}
+
+func (c *commandFsRm) Name() string {
+	return "fs.rm"
+}
+
+func (c *commandFsRm) Help() string {
+	return `remove file and directory entries
+
+	fs.rm [-rf] <entry1> <entry2> ...
+
+	fs.rm /dir/file_name1 dir/file_name2
+	fs.rm /dir
+
+	The option "-r" can be recursive.
+	The option "-f" can be ignored by recursive error.
+`
+}
+
+func (c *commandFsRm) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
+	isRecursive := false
+	ignoreRecursiveError := false
+	var entiries []string
+	for _, arg := range args {
+		if !strings.HasPrefix(arg, "-") {
+			entiries = append(entiries, arg)
+			continue
+		}
+		for _, t := range arg {
+			switch t {
+			case 'r':
+				isRecursive = true
+			case 'f':
+				ignoreRecursiveError = true
+			}
+		}
+	}
+	if len(entiries) < 1 {
+		return fmt.Errorf("need to have arguments")
+	}
+
+	commandEnv.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+		for _, entry := range entiries {
+			targetPath, err := commandEnv.parseUrl(entry)
+			if err != nil {
+				fmt.Fprintf(writer, "rm: %s: %v\n", targetPath, err)
+				continue
+			}
+
+			targetDir, targetName := util.FullPath(targetPath).DirAndName()
+
+			lookupRequest := &filer_pb.LookupDirectoryEntryRequest{
+				Directory: targetDir,
+				Name:      targetName,
+			}
+			_, err = filer_pb.LookupEntry(client, lookupRequest)
+			if err != nil {
+				fmt.Fprintf(writer, "rm: %s: %v\n", targetPath, err)
+				continue
+			}
+
+			request := &filer_pb.DeleteEntryRequest{
+				Directory:            targetDir,
+				Name:                 targetName,
+				IgnoreRecursiveError: ignoreRecursiveError,
+				IsDeleteData:         true,
+				IsRecursive:          isRecursive,
+				IsFromOtherCluster:   false,
+				Signatures:           nil,
+			}
+			if resp, err := client.DeleteEntry(context.Background(), request); err != nil {
+				fmt.Fprintf(writer, "rm: %s: %v\n", targetPath, err)
+			} else {
+				if resp.Error != "" {
+					fmt.Fprintf(writer, "rm: %s: %v\n", targetPath, resp.Error)
+				}
+			}
+		}
+		return nil
+	})
+
+	return
+}

weed/shell/command_remote_configure.go

@@ -0,0 +1,142 @@
+package shell
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/filer"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"github.com/golang/protobuf/proto"
+	"io"
+	"regexp"
+)
+
+func init() {
+	Commands = append(Commands, &commandRemoteConfigure{})
+}
+
+type commandRemoteConfigure struct {
+}
+
+func (c *commandRemoteConfigure) Name() string {
+	return "remote.configure"
+}
+
+func (c *commandRemoteConfigure) Help() string {
+	return `remote storage configuration
+
+	# see the current configurations
+	remote.configure
+
+	# set or update a configuration
+	remote.configure -name=cloud1 -type=s3 -access_key=xxx -secret_key=yyy
+
+	# delete one configuration
+	remote.configure -delete -name=cloud1
+
+`
+}
+
+var (
+	isAlpha = regexp.MustCompile(`^[A-Za-z][A-Za-z0-9]*$`).MatchString
+)
+
+func (c *commandRemoteConfigure) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
+
+	conf := &filer_pb.RemoteConf{}
+
+	remoteConfigureCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	isDelete := remoteConfigureCommand.Bool("delete", false, "delete one remote storage by its name")
+
+	remoteConfigureCommand.StringVar(&conf.Name, "name", "", "a short name to identify the remote storage")
+	remoteConfigureCommand.StringVar(&conf.Type, "type", "s3", "storage type, currently only support s3")
+
+	remoteConfigureCommand.StringVar(&conf.S3AccessKey, "s3.access_key", "", "s3 access key")
+	remoteConfigureCommand.StringVar(&conf.S3SecretKey, "s3.secret_key", "", "s3 secret key")
+	remoteConfigureCommand.StringVar(&conf.S3Region, "s3.region", "us-east-2", "s3 region")
+	remoteConfigureCommand.StringVar(&conf.S3Endpoint, "s3.endpoint", "", "endpoint for s3-compatible local object store")
+
+	if err = remoteConfigureCommand.Parse(args); err != nil {
+		return nil
+	}
+
+	if conf.Name == "" {
+		return c.listExistingRemoteStorages(commandEnv, writer)
+	}
+
+	if !isAlpha(conf.Name) {
+		return fmt.Errorf("only letters and numbers allowed in name: %v", conf.Name)
+	}
+
+	if *isDelete {
+		return c.deleteRemoteStorage(commandEnv, writer, conf.Name)
+	}
+
+	return c.saveRemoteStorage(commandEnv, writer, conf)
+
+}
+
+func (c *commandRemoteConfigure) listExistingRemoteStorages(commandEnv *CommandEnv, writer io.Writer) error {
+
+	return filer_pb.ReadDirAllEntries(commandEnv, util.FullPath(filer.DirectoryEtcRemote), "", func(entry *filer_pb.Entry, isLast bool) error {
+		if len(entry.Content) == 0 {
+			fmt.Fprintf(writer, "skipping %s\n", entry.Name)
+			return nil
+		}
+		conf := &filer_pb.RemoteConf{}
+
+		if err := proto.Unmarshal(entry.Content, conf); err != nil {
+			return fmt.Errorf("unmarshal %s/%s: %v", filer.DirectoryEtcRemote, entry.Name, err)
+		}
+
+		conf.S3SecretKey = ""
+
+		fmt.Fprintf(writer, "%+v\n", conf)
+
+		return nil
+	})
+
+}
+
+func (c *commandRemoteConfigure) deleteRemoteStorage(commandEnv *CommandEnv, writer io.Writer, storageName string) error {
+
+	return commandEnv.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+
+		request := &filer_pb.DeleteEntryRequest{
+			Directory:            filer.DirectoryEtcRemote,
+			Name:                 storageName,
+			IgnoreRecursiveError: false,
+			IsDeleteData:         true,
+			IsRecursive:          true,
+			IsFromOtherCluster:   false,
+			Signatures:           nil,
+		}
+		_, err := client.DeleteEntry(context.Background(), request)
+
+		if err == nil {
+			fmt.Fprintf(writer, "removed: %s\n", storageName)
+		}
+
+		return err
+
+	})
+
+}
+
+func (c *commandRemoteConfigure) saveRemoteStorage(commandEnv *CommandEnv, writer io.Writer, conf *filer_pb.RemoteConf) error {
+
+	data, err := proto.Marshal(conf)
+	if err != nil {
+		return err
+	}
+
+	if err = commandEnv.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+		return filer.SaveInsideFiler(client, filer.DirectoryEtcRemote, conf.Name, data)
+	}); err != nil && err != filer_pb.ErrNotFound {
+		return err
+	}
+
+	return nil
+
+}

weed/shell/command_s3_bucket_delete.go

@@ -1,8 +1,10 @@
 package shell
 
 import (
+	"context"
 	"flag"
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
 	"io"
 
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
@@ -49,6 +51,17 @@ func (c *commandS3BucketDelete) Do(args []string, commandEnv *CommandEnv, writer
 		return fmt.Errorf("read buckets: %v", err)
 	}
 
+	// delete the collection directly first
+	err = commandEnv.MasterClient.WithClient(func(client master_pb.SeaweedClient) error {
+		_, err = client.CollectionDelete(context.Background(), &master_pb.CollectionDeleteRequest{
+			Name: *bucketName,
+		})
+		return err
+	})
+	if err != nil {
+		return
+	}
+
 	return filer_pb.Remove(commandEnv, filerBucketsPath, *bucketName, false, true, true, false, nil)
 
 }

weed/shell/command_volume_check_disk.go

@@ -89,25 +89,8 @@ func (c *commandVolumeCheckDisk) Do(args []string, commandEnv *CommandEnv, write
 				continue
 			}
 
-			// reset index db
-			aDB.Close()
-			bDB.Close()
-			aDB, bDB = needle_map.NewMemDb(), needle_map.NewMemDb()
-
-			// read index db
-			if err := c.readIndexDatabase(aDB, a.info.Collection, a.info.Id, a.location.dataNode.Id, *verbose, writer); err != nil {
-				return err
-			}
-			if err := c.readIndexDatabase(bDB, b.info.Collection, b.info.Id, b.location.dataNode.Id, *verbose, writer); err != nil {
-				return err
-			}
-
-			// find and make up the differnces
-			if err := c.doVolumeCheckDisk(aDB, bDB, a, b, *verbose, writer, *applyChanges, *nonRepairThreshold); err != nil {
-				return err
-			}
-			if err := c.doVolumeCheckDisk(bDB, aDB, b, a, *verbose, writer, *applyChanges, *nonRepairThreshold); err != nil {
-				return err
+			if err := c.syncTwoReplicas(aDB, bDB, a, verbose, writer, b, err, applyChanges, nonRepairThreshold); err != nil {
+				fmt.Fprintf(writer, "snyc volume %d on %s and %s: %v\n", a.info.Id, a.location.dataNode.Id, b.location.dataNode.Id, err)
 			}
 			replicas = replicas[1:]
 		}
@@ -116,7 +99,34 @@ func (c *commandVolumeCheckDisk) Do(args []string, commandEnv *CommandEnv, write
 	return nil
 }
 
-func (c *commandVolumeCheckDisk) doVolumeCheckDisk(subtrahend, minuend *needle_map.MemDb, source, target *VolumeReplica, verbose bool, writer io.Writer, applyChanges bool, nonRepairThreshold float64) error {
+func (c *commandVolumeCheckDisk) syncTwoReplicas(aDB *needle_map.MemDb, bDB *needle_map.MemDb, a *VolumeReplica, verbose *bool, writer io.Writer, b *VolumeReplica, err error, applyChanges *bool, nonRepairThreshold *float64) error {
+	aHasChanges, bHasChanges := true, true
+	for aHasChanges || bHasChanges {
+		// reset index db
+		aDB.Close()
+		bDB.Close()
+		aDB, bDB = needle_map.NewMemDb(), needle_map.NewMemDb()
+
+		// read index db
+		if err := c.readIndexDatabase(aDB, a.info.Collection, a.info.Id, a.location.dataNode.Id, *verbose, writer); err != nil {
+			return err
+		}
+		if err := c.readIndexDatabase(bDB, b.info.Collection, b.info.Id, b.location.dataNode.Id, *verbose, writer); err != nil {
+			return err
+		}
+
+		// find and make up the differences
+		if aHasChanges, err = c.doVolumeCheckDisk(aDB, bDB, a, b, *verbose, writer, *applyChanges, *nonRepairThreshold); err != nil {
+			return err
+		}
+		if bHasChanges, err = c.doVolumeCheckDisk(bDB, aDB, b, a, *verbose, writer, *applyChanges, *nonRepairThreshold); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (c *commandVolumeCheckDisk) doVolumeCheckDisk(subtrahend, minuend *needle_map.MemDb, source, target *VolumeReplica, verbose bool, writer io.Writer, applyChanges bool, nonRepairThreshold float64) (hasChanges bool, err error) {
 
 	// find missing keys
 	// hash join, can be more efficient
@@ -133,12 +143,12 @@ func (c *commandVolumeCheckDisk) doVolumeCheckDisk(subtrahend, minuend *needle_m
 	fmt.Fprintf(writer, "volume %d %s has %d entries, %s missed %d entries\n", source.info.Id, source.location.dataNode.Id, counter, target.location.dataNode.Id, len(missingNeedles))
 
 	if counter == 0 || len(missingNeedles) == 0 {
-		return nil
+		return false, nil
 	}
 
 	missingNeedlesFraction := float64(len(missingNeedles)) / float64(counter)
 	if missingNeedlesFraction > nonRepairThreshold {
-		return fmt.Errorf(
+		return false, fmt.Errorf(
 			"failed to start repair volume %d, percentage of missing keys is greater than the threshold: %.2f > %.2f",
 			source.info.Id, missingNeedlesFraction, nonRepairThreshold)
 	}
@@ -147,7 +157,7 @@ func (c *commandVolumeCheckDisk) doVolumeCheckDisk(subtrahend, minuend *needle_m
 
 		needleBlob, err := c.readSourceNeedleBlob(source.location.dataNode.Id, source.info.Id, needleValue)
 		if err != nil {
-			return err
+			return hasChanges, err
 		}
 
 		if !applyChanges {
@@ -158,13 +168,15 @@ func (c *commandVolumeCheckDisk) doVolumeCheckDisk(subtrahend, minuend *needle_m
 			fmt.Fprintf(writer, "read %d,%x %s => %s \n", source.info.Id, needleValue.Key, source.location.dataNode.Id, target.location.dataNode.Id)
 		}
 
-		if err := c.writeNeedleBlobToTarget(target.location.dataNode.Id, source.info.Id, needleValue, needleBlob); err != nil {
-			return err
+		hasChanges = true
+
+		if err = c.writeNeedleBlobToTarget(target.location.dataNode.Id, source.info.Id, needleValue, needleBlob); err != nil {
+			return hasChanges, err
 		}
 
 	}
 
-	return nil
+	return
 }
 
 func (c *commandVolumeCheckDisk) readSourceNeedleBlob(sourceVolumeServer string, volumeId uint32, needleValue needle_map.NeedleValue) (needleBlob []byte, err error) {

weed/shell/command_volume_fix_replication.go

@@ -58,6 +58,7 @@ func (c *commandVolumeFixReplication) Do(args []string, commandEnv *CommandEnv,
 	volFixReplicationCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	c.collectionPattern = volFixReplicationCommand.String("collectionPattern", "", "match with wildcard characters '*' and '?'")
 	skipChange := volFixReplicationCommand.Bool("n", false, "skip the changes")
+	retryCount := volFixReplicationCommand.Int("retry", 0, "how many times to retry")
 	if err = volFixReplicationCommand.Parse(args); err != nil {
 		return nil
 	}
@@ -100,7 +101,7 @@ func (c *commandVolumeFixReplication) Do(args []string, commandEnv *CommandEnv,
 	}
 
 	// find the most under populated data nodes
-	return c.fixUnderReplicatedVolumes(commandEnv, writer, takeAction, underReplicatedVolumeIds, volumeReplicas, allLocations)
+	return c.fixUnderReplicatedVolumes(commandEnv, writer, takeAction, underReplicatedVolumeIds, volumeReplicas, allLocations, *retryCount)
 
 }
 
@@ -154,64 +155,72 @@ func (c *commandVolumeFixReplication) fixOverReplicatedVolumes(commandEnv *Comma
 	return nil
 }
 
-func (c *commandVolumeFixReplication) fixUnderReplicatedVolumes(commandEnv *CommandEnv, writer io.Writer, takeAction bool, underReplicatedVolumeIds []uint32, volumeReplicas map[uint32][]*VolumeReplica, allLocations []location) error {
+func (c *commandVolumeFixReplication) fixUnderReplicatedVolumes(commandEnv *CommandEnv, writer io.Writer, takeAction bool, underReplicatedVolumeIds []uint32, volumeReplicas map[uint32][]*VolumeReplica, allLocations []location, retryCount int) (err error) {
 
 	for _, vid := range underReplicatedVolumeIds {
-		replicas := volumeReplicas[vid]
-		replica := pickOneReplicaToCopyFrom(replicas)
-		replicaPlacement, _ := super_block.NewReplicaPlacementFromByte(byte(replica.info.ReplicaPlacement))
-		foundNewLocation := false
-		hasSkippedCollection := false
-		keepDataNodesSorted(allLocations, types.ToDiskType(replica.info.DiskType))
-		fn := capacityByFreeVolumeCount(types.ToDiskType(replica.info.DiskType))
-		for _, dst := range allLocations {
-			// check whether data nodes satisfy the constraints
-			if fn(dst.dataNode) > 0 && satisfyReplicaPlacement(replicaPlacement, replicas, dst) {
-				// check collection name pattern
-				if *c.collectionPattern != "" {
-					matched, err := filepath.Match(*c.collectionPattern, replica.info.Collection)
-					if err != nil {
-						return fmt.Errorf("match pattern %s with collection %s: %v", *c.collectionPattern, replica.info.Collection, err)
-					}
-					if !matched {
-						hasSkippedCollection = true
-						break
-					}
-				}
-
-				// ask the volume server to replicate the volume
-				foundNewLocation = true
-				fmt.Fprintf(writer, "replicating volume %d %s from %s to dataNode %s ...\n", replica.info.Id, replicaPlacement, replica.location.dataNode.Id, dst.dataNode.Id)
-
-				if !takeAction {
-					break
-				}
-
-				err := operation.WithVolumeServerClient(dst.dataNode.Id, commandEnv.option.GrpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
-					_, replicateErr := volumeServerClient.VolumeCopy(context.Background(), &volume_server_pb.VolumeCopyRequest{
-						VolumeId:       replica.info.Id,
-						SourceDataNode: replica.location.dataNode.Id,
-					})
-					if replicateErr != nil {
-						return fmt.Errorf("copying from %s => %s : %v", replica.location.dataNode.Id, dst.dataNode.Id, replicateErr)
-					}
-					return nil
-				})
-
-				if err != nil {
-					return err
-				}
-
-				// adjust free volume count
-				dst.dataNode.DiskInfos[replica.info.DiskType].FreeVolumeCount--
-				break
+		for i := 0; i < retryCount+1; i++ {
+			if err = c.fixOneUnderReplicatedVolume(commandEnv, writer, takeAction, volumeReplicas, vid, allLocations); err == nil {
+				continue
 			}
 		}
+	}
+	return
+}
 
-		if !foundNewLocation && !hasSkippedCollection {
-			fmt.Fprintf(writer, "failed to place volume %d replica as %s, existing:%+v\n", replica.info.Id, replicaPlacement, len(replicas))
+func (c *commandVolumeFixReplication) fixOneUnderReplicatedVolume(commandEnv *CommandEnv, writer io.Writer, takeAction bool, volumeReplicas map[uint32][]*VolumeReplica, vid uint32, allLocations []location) error {
+	replicas := volumeReplicas[vid]
+	replica := pickOneReplicaToCopyFrom(replicas)
+	replicaPlacement, _ := super_block.NewReplicaPlacementFromByte(byte(replica.info.ReplicaPlacement))
+	foundNewLocation := false
+	hasSkippedCollection := false
+	keepDataNodesSorted(allLocations, types.ToDiskType(replica.info.DiskType))
+	fn := capacityByFreeVolumeCount(types.ToDiskType(replica.info.DiskType))
+	for _, dst := range allLocations {
+		// check whether data nodes satisfy the constraints
+		if fn(dst.dataNode) > 0 && satisfyReplicaPlacement(replicaPlacement, replicas, dst) {
+			// check collection name pattern
+			if *c.collectionPattern != "" {
+				matched, err := filepath.Match(*c.collectionPattern, replica.info.Collection)
+				if err != nil {
+					return fmt.Errorf("match pattern %s with collection %s: %v", *c.collectionPattern, replica.info.Collection, err)
+				}
+				if !matched {
+					hasSkippedCollection = true
+					break
+				}
+			}
+
+			// ask the volume server to replicate the volume
+			foundNewLocation = true
+			fmt.Fprintf(writer, "replicating volume %d %s from %s to dataNode %s ...\n", replica.info.Id, replicaPlacement, replica.location.dataNode.Id, dst.dataNode.Id)
+
+			if !takeAction {
+				break
+			}
+
+			err := operation.WithVolumeServerClient(dst.dataNode.Id, commandEnv.option.GrpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
+				_, replicateErr := volumeServerClient.VolumeCopy(context.Background(), &volume_server_pb.VolumeCopyRequest{
+					VolumeId:       replica.info.Id,
+					SourceDataNode: replica.location.dataNode.Id,
+				})
+				if replicateErr != nil {
+					return fmt.Errorf("copying from %s => %s : %v", replica.location.dataNode.Id, dst.dataNode.Id, replicateErr)
+				}
+				return nil
+			})
+
+			if err != nil {
+				return err
+			}
+
+			// adjust free volume count
+			dst.dataNode.DiskInfos[replica.info.DiskType].FreeVolumeCount--
+			break
 		}
+	}
 
+	if !foundNewLocation && !hasSkippedCollection {
+		fmt.Fprintf(writer, "failed to place volume %d replica as %s, existing:%+v\n", replica.info.Id, replicaPlacement, len(replicas))
 	}
 	return nil
 }

weed/shell/command_volume_fsck.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"flag"
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/storage/needle"
 	"io"
 	"io/ioutil"
 	"math"
@@ -163,7 +164,7 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(volumeIdToServer map[uint
 		if verbose && entry.Entry.IsDirectory {
 			fmt.Fprintf(writer, "checking directory %s\n", util.NewFullPath(entry.Dir, entry.Entry.Name))
 		}
-		dChunks, mChunks, resolveErr := filer.ResolveChunkManifest(filer.LookupFn(c.env), entry.Entry.Chunks)
+		dChunks, mChunks, resolveErr := filer.ResolveChunkManifest(filer.LookupFn(c.env), entry.Entry.Chunks, 0, math.MaxInt64)
 		if resolveErr != nil {
 			return nil
 		}
@@ -214,8 +215,14 @@ func (c *commandVolumeFsck) findExtraChunksInVolumeServers(volumeIdToVInfo map[u
 			if vinfo.isEcVolume {
 				fmt.Fprintf(writer, "Skip purging for Erasure Coded volumes.\n")
 			}
-			if err := c.purgeFileIdsForOneVolume(volumeId, orphanFileIds, writer); err != nil {
-				return fmt.Errorf("purge for volume %d: %v\n", volumeId, err)
+			if inUseCount == 0 {
+				if err := deleteVolume(c.env.option.GrpcDialOption, needle.VolumeId(volumeId), vinfo.server); err != nil {
+					return fmt.Errorf("delete volume %d: %v\n", volumeId, err)
+				}
+			} else {
+				if err := c.purgeFileIdsForOneVolume(volumeId, orphanFileIds, writer); err != nil {
+					return fmt.Errorf("purge for volume %d: %v\n", volumeId, err)
+				}
 			}
 		}
 	}
@@ -304,7 +311,7 @@ func (c *commandVolumeFsck) collectFilerFileIds(tempFolder string, volumeIdToSer
 			files[i.vid].Write(buffer)
 		}
 	}, func(entry *filer_pb.FullEntry, outputChan chan interface{}) (err error) {
-		dChunks, mChunks, resolveErr := filer.ResolveChunkManifest(filer.LookupFn(c.env), entry.Entry.Chunks)
+		dChunks, mChunks, resolveErr := filer.ResolveChunkManifest(filer.LookupFn(c.env), entry.Entry.Chunks, 0, math.MaxInt64)
 		if resolveErr != nil {
 			return nil
 		}

weed/shell/command_volume_server_evacuate.go

@@ -52,6 +52,7 @@ func (c *commandVolumeServerEvacuate) Do(args []string, commandEnv *CommandEnv,
 	volumeServer := vsEvacuateCommand.String("node", "", "<host>:<port> of the volume server")
 	skipNonMoveable := vsEvacuateCommand.Bool("skipNonMoveable", false, "skip volumes that can not be moved")
 	applyChange := vsEvacuateCommand.Bool("force", false, "actually apply the changes")
+	retryCount := vsEvacuateCommand.Int("retry", 0, "how many times to retry")
 	if err = vsEvacuateCommand.Parse(args); err != nil {
 		return nil
 	}
@@ -60,7 +61,13 @@ func (c *commandVolumeServerEvacuate) Do(args []string, commandEnv *CommandEnv,
 		return fmt.Errorf("need to specify volume server by -node=<host>:<port>")
 	}
 
-	return volumeServerEvacuate(commandEnv, *volumeServer, *skipNonMoveable, *applyChange, writer)
+	for i := 0; i < *retryCount+1; i++ {
+		if err = volumeServerEvacuate(commandEnv, *volumeServer, *skipNonMoveable, *applyChange, writer); err == nil {
+			return nil
+		}
+	}
+
+	return
 
 }
 

weed/storage/store.go

@@ -46,6 +46,7 @@ type Store struct {
 	DeletedVolumesChan  chan master_pb.VolumeShortInformationMessage
 	NewEcShardsChan     chan master_pb.VolumeEcShardInformationMessage
 	DeletedEcShardsChan chan master_pb.VolumeEcShardInformationMessage
+	isStopping          bool
 }
 
 func (s *Store) String() (str string) {
@@ -321,6 +322,10 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 
 }
 
+func (s *Store) SetStopping() {
+	s.isStopping = true
+}
+
 func (s *Store) Close() {
 	for _, location := range s.Locations {
 		location.Close()
@@ -333,7 +338,7 @@ func (s *Store) WriteVolumeNeedle(i needle.VolumeId, n *needle.Needle, fsync boo
 			err = fmt.Errorf("volume %d is read only", i)
 			return
 		}
-		_, _, isUnchanged, err = v.writeNeedle2(n, fsync)
+		_, _, isUnchanged, err = v.writeNeedle2(n, fsync && s.isStopping)
 		return
 	}
 	glog.V(0).Infoln("volume", i, "not found!")

weed/util/constants.go

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

weed/util/log_buffer/log_buffer.go

@@ -170,8 +170,8 @@ func (m *LogBuffer) copyToFlush() *dataToFlush {
 			m.lastFlushTime = m.stopTime
 		}
 		m.buf = m.prevBuffers.SealBuffer(m.startTime, m.stopTime, m.buf, m.pos)
-		m.startTime = time.Unix(0,0)
-		m.stopTime = time.Unix(0,0)
+		m.startTime = time.Unix(0, 0)
+		m.stopTime = time.Unix(0, 0)
 		m.pos = 0
 		m.idx = m.idx[:0]
 		return d