2.76

related to https://github.com/chrislusf/seaweedfs/issues/1271

.github/workflows/container_latest.yml

@@ -2,7 +2,8 @@ name: "docker: build latest container"
 
 on:
   push:
-    branches: [ master ]
+    tags:
+      - '*'
   workflow_dispatch: []
 
 jobs:
@@ -23,7 +24,7 @@ jobs:
             chrislusf/seaweedfs
             ghcr.io/chrislusf/seaweedfs
           tags: |
-            type=raw,value=dev
+            type=raw,value=latest
           labels: |
             org.opencontainers.image.title=seaweedfs
             org.opencontainers.image.description=SeaweedFS is a distributed storage system for blobs, objects, files, and data lake, to store and serve billions of files fast!

go.mod

@@ -112,9 +112,9 @@ require (
 	github.com/stretchr/testify v1.7.0
 	github.com/stvp/tempredis v0.0.0-20181119212430-b82af8480203
 	github.com/syndtr/goleveldb v1.0.1-0.20190318030020-c3a204f8e965
-	github.com/tidwall/gjson v1.8.1
-	github.com/tidwall/match v1.0.3
-	github.com/tidwall/pretty v1.1.0 // indirect
+	github.com/tidwall/gjson v1.10.2
+	github.com/tidwall/match v1.1.1
+	github.com/tidwall/pretty v1.2.0 // indirect
 	github.com/tsuna/gohbase v0.0.0-20201125011725-348991136365
 	github.com/tylertreat/BoomFilters v0.0.0-20210315201527-1a82519a3e43
 	github.com/valyala/bytebufferpool v1.0.0

go.sum

@@ -778,11 +778,17 @@ github.com/syndtr/goleveldb v1.0.1-0.20190318030020-c3a204f8e965 h1:1oFLiOyVl+W7
 github.com/syndtr/goleveldb v1.0.1-0.20190318030020-c3a204f8e965/go.mod h1:9OrXJhf154huy1nPWmuSrkgjPUtUNhA+Zmy+6AESzuA=
 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/gjson v1.10.2 h1:APbLGOM0rrEkd8WBw9C24nllro4ajFuJu0Sc9hRz8Bo=
+github.com/tidwall/gjson v1.10.2/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
 github.com/tidwall/match v1.0.3 h1:FQUVvBImDutD8wJLN6c5eMzWtjgONK9MwIBCOrUJKeE=
 github.com/tidwall/match v1.0.3/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
+github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA=
+github.com/tidwall/match v1.1.1/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/tidwall/pretty v1.2.0 h1:RWIZEg2iJ8/g6fDDYzMpobmaoGh5OLl4AXtGUGPcqCs=
+github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
 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/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
 github.com/tsuna/gohbase v0.0.0-20201125011725-348991136365 h1:6iRwZdrFUzbcVYZwa8dXTIILGIxmmhjyUPJEcwzPGaU=

k8s/helm_charts2/Chart.yaml

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

other/java/client/pom.xml

@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.6.8</version>
+    <version>1.6.9</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>

other/java/client/pom.xml.deploy

@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.6.8</version>
+    <version>1.6.9</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>

other/java/client/pom_debug.xml

@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.6.8</version>
+    <version>1.6.9</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>

other/java/client/src/main/java/seaweedfs/client/RemoteUtil.java

@@ -14,8 +14,8 @@ public class RemoteUtil {
         String dir = SeaweedOutputStream.getParentDirectory(fullpath);
         String name = SeaweedOutputStream.getFileName(fullpath);
 
-        final FilerProto.DownloadToLocalResponse downloadToLocalResponse = filerClient.getBlockingStub()
-                .downloadToLocal(FilerProto.DownloadToLocalRequest.newBuilder()
+        final FilerProto.CacheRemoteObjectToLocalClusterResponse downloadToLocalResponse = filerClient.getBlockingStub()
+                .downloadToLocal(FilerProto.CacheRemoteObjectToLocalClusterRequest.newBuilder()
                 .setDirectory(dir).setName(name).build());
 
         return downloadToLocalResponse.getEntry();

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

@@ -69,7 +69,7 @@ service SeaweedFiler {
     rpc KvPut (KvPutRequest) returns (KvPutResponse) {
     }
 
-    rpc DownloadToLocal (DownloadToLocalRequest) returns (DownloadToLocalResponse) {
+    rpc CacheRemoteObjectToLocalCluster (CacheRemoteObjectToLocalClusterRequest) returns (CacheRemoteObjectToLocalClusterResponse) {
     }
 }
 
@@ -403,10 +403,10 @@ message FilerConf {
 /////////////////////////
 // Remote Storage related
 /////////////////////////
-message DownloadToLocalRequest {
+message CacheRemoteObjectToLocalClusterRequest {
     string directory = 1;
     string name = 2;
 }
-message DownloadToLocalResponse {
+message CacheRemoteObjectToLocalClusterResponse {
     Entry entry = 1;
 }

other/java/examples/pom.xml

@@ -11,13 +11,13 @@
         <dependency>
             <groupId>com.github.chrislusf</groupId>
             <artifactId>seaweedfs-client</artifactId>
-            <version>1.6.8</version>
+            <version>1.6.9</version>
             <scope>compile</scope>
         </dependency>
         <dependency>
             <groupId>com.github.chrislusf</groupId>
             <artifactId>seaweedfs-hadoop2-client</artifactId>
-            <version>1.6.8</version>
+            <version>1.6.9</version>
             <scope>compile</scope>
         </dependency>
         <dependency>

other/java/hdfs2/dependency-reduced-pom.xml

@@ -301,7 +301,7 @@
     </snapshotRepository>
   </distributionManagement>
   <properties>
-    <seaweedfs.client.version>1.6.8</seaweedfs.client.version>
+    <seaweedfs.client.version>1.6.9</seaweedfs.client.version>
     <hadoop.version>2.9.2</hadoop.version>
   </properties>
 </project>

other/java/hdfs2/pom.xml

@@ -5,7 +5,7 @@
     <modelVersion>4.0.0</modelVersion>
 
     <properties>
-        <seaweedfs.client.version>1.6.8</seaweedfs.client.version>
+        <seaweedfs.client.version>1.6.9</seaweedfs.client.version>
         <hadoop.version>2.9.2</hadoop.version>
     </properties>
 

other/java/hdfs3/dependency-reduced-pom.xml

@@ -309,7 +309,7 @@
     </snapshotRepository>
   </distributionManagement>
   <properties>
-    <seaweedfs.client.version>1.6.8</seaweedfs.client.version>
+    <seaweedfs.client.version>1.6.9</seaweedfs.client.version>
     <hadoop.version>3.1.1</hadoop.version>
   </properties>
 </project>

other/java/hdfs3/pom.xml

@@ -5,7 +5,7 @@
     <modelVersion>4.0.0</modelVersion>
 
     <properties>
-        <seaweedfs.client.version>1.6.8</seaweedfs.client.version>
+        <seaweedfs.client.version>1.6.9</seaweedfs.client.version>
         <hadoop.version>3.1.1</hadoop.version>
     </properties>
 

weed/command/backup.go

@@ -120,7 +120,7 @@ func runBackup(cmd *Command, args []string) bool {
 	}
 
 	if v.SuperBlock.CompactionRevision < uint16(stats.CompactRevision) {
-		if err = v.Compact2(30*1024*1024*1024, 0); err != nil {
+		if err = v.Compact2(30*1024*1024*1024, 0, nil); err != nil {
 			fmt.Printf("Compact Volume before synchronizing %v\n", err)
 			return true
 		}

weed/command/compact.go

@@ -50,7 +50,7 @@ func runCompact(cmd *Command, args []string) bool {
 			glog.Fatalf("Compact Volume [ERROR] %s\n", err)
 		}
 	} else {
-		if err = v.Compact2(preallocate, 0); err != nil {
+		if err = v.Compact2(preallocate, 0, nil); err != nil {
 			glog.Fatalf("Compact Volume [ERROR] %s\n", err)
 		}
 	}

weed/command/filer_meta_tail.go

@@ -30,6 +30,8 @@ var cmdFilerMetaTail = &Command{
 	weed filer.meta.tail -timeAgo=30h | jq .
 	weed filer.meta.tail -timeAgo=30h | jq .eventNotification.newEntry.name
 
+	weed filer.meta.tail -timeAgo=30h -es=http://<elasticSearchServerHost>:<port> -es.index=seaweedfs
+
   `,
 }
 

weed/filer/filer_delete_entry.go

@@ -80,7 +80,7 @@ func (f *Filer) doBatchDeleteFolderMetaAndData(ctx context.Context, entry *Entry
 			}
 			if lastFileName == "" && !isRecursive && len(entries) > 0 {
 				// only for first iteration in the loop
-				glog.Errorf("deleting a folder %s has children: %+v ...", entry.FullPath, entries[0].Name())
+				glog.V(0).Infof("deleting a folder %s has children: %+v ...", entry.FullPath, entries[0].Name())
 				return nil, nil, fmt.Errorf("%s: %s", MsgFailDelNonEmptyFolder, entry.FullPath)
 			}
 

weed/filer/filerstore_wrapper.go

@@ -296,7 +296,7 @@ func (fsw *FilerStoreWrapper) prefixFilterEntries(ctx context.Context, dirPath u
 		}
 		if count < limit {
 			notPrefixed = notPrefixed[:0]
-			_, err = actualStore.ListDirectoryEntries(ctx, dirPath, lastFileName, false, limit, func(entry *Entry) bool {
+			lastFileName, err = actualStore.ListDirectoryEntries(ctx, dirPath, lastFileName, false, limit, func(entry *Entry) bool {
 				notPrefixed = append(notPrefixed, entry)
 				return true
 			})

weed/filer/read_remote.go

@@ -25,9 +25,9 @@ func MapRemoteStorageLocationPathToFullPath(localMountedDir util.FullPath, remot
 	return localMountedDir.Child(remoteLocationPath[len(remoteMountedLocation.Path):])
 }
 
-func DownloadToLocal(filerClient filer_pb.FilerClient, remoteConf *remote_pb.RemoteConf, remoteLocation *remote_pb.RemoteStorageLocation, parent util.FullPath, entry *filer_pb.Entry) error {
+func CacheRemoteObjectToLocalCluster(filerClient filer_pb.FilerClient, remoteConf *remote_pb.RemoteConf, remoteLocation *remote_pb.RemoteStorageLocation, parent util.FullPath, entry *filer_pb.Entry) error {
 	return filerClient.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
-		_, err := client.DownloadToLocal(context.Background(), &filer_pb.DownloadToLocalRequest{
+		_, err := client.CacheRemoteObjectToLocalCluster(context.Background(), &filer_pb.CacheRemoteObjectToLocalClusterRequest{
 			Directory: string(parent),
 			Name:      entry.Name,
 		})

weed/filer/redis2/universal_redis_store.go

@@ -133,7 +133,10 @@ func (store *UniversalRedis2Store) DeleteFolderChildren(ctx context.Context, ful
 		return nil
 	}
 
-	members, err := store.Client.ZRange(ctx, genDirectoryListKey(string(fullpath)), 0, -1).Result()
+	members, err := store.Client.ZRangeByLex(ctx, genDirectoryListKey(string(fullpath)), &redis.ZRangeBy{
+		Min:    "-",
+		Max:    "+",
+	}).Result()
 	if err != nil {
 		return fmt.Errorf("DeleteFolderChildren %s : %v", fullpath, err)
 	}
@@ -158,14 +161,22 @@ func (store *UniversalRedis2Store) ListDirectoryPrefixedEntries(ctx context.Cont
 func (store *UniversalRedis2Store) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
 
 	dirListKey := genDirectoryListKey(string(dirPath))
-	start := int64(0)
+
+	min := "-"
 	if startFileName != "" {
-		start, _ = store.Client.ZRank(ctx, dirListKey, startFileName).Result()
-		if !includeStartFile {
-			start++
+		if includeStartFile {
+			min = "[" + startFileName
+		} else {
+			min = "(" + startFileName
 		}
 	}
-	members, err := store.Client.ZRange(ctx, dirListKey, start, start+int64(limit)-1).Result()
+
+	members, err := store.Client.ZRangeByLex(ctx, dirListKey, &redis.ZRangeBy{
+		Min:    min,
+		Max:    "+",
+		Offset: 0,
+		Count:  limit,
+	}).Result()
 	if err != nil {
 		return lastFileName, fmt.Errorf("list %s : %v", dirPath, err)
 	}

weed/filesys/file.go

@@ -364,15 +364,15 @@ func (file *File) getEntry() *filer_pb.Entry {
 func (file *File) downloadRemoteEntry(entry *filer_pb.Entry) (*filer_pb.Entry, error) {
 	err := file.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
 
-		request := &filer_pb.DownloadToLocalRequest{
+		request := &filer_pb.CacheRemoteObjectToLocalClusterRequest{
 			Directory: file.dir.FullPath(),
 			Name:      entry.Name,
 		}
 
 		glog.V(4).Infof("download entry: %v", request)
-		resp, err := client.DownloadToLocal(context.Background(), request)
+		resp, err := client.CacheRemoteObjectToLocalCluster(context.Background(), request)
 		if err != nil {
-			glog.Errorf("DownloadToLocal file %s/%s: %v", file.dir.FullPath(), file.Name, err)
+			glog.Errorf("CacheRemoteObjectToLocalCluster file %s/%s: %v", file.dir.FullPath(), file.Name, err)
 			return fuse.EIO
 		}
 

weed/iamapi/iamapi_handlers.go

@@ -8,7 +8,7 @@ import (
 	"net/http"
 )
 
-func writeIamErrorResponse(w http.ResponseWriter, err error, object string, value string, msg error) {
+func writeIamErrorResponse(w http.ResponseWriter, r *http.Request, err error, object string, value string, msg error) {
 	errCode := err.Error()
 	errorResp := ErrorResponse{}
 	errorResp.Error.Type = "Sender"
@@ -22,10 +22,10 @@ func writeIamErrorResponse(w http.ResponseWriter, err error, object string, valu
 	case iam.ErrCodeNoSuchEntityException:
 		msg := fmt.Sprintf("The %s with name %s cannot be found.", object, value)
 		errorResp.Error.Message = &msg
-		s3err.WriteXMLResponse(w, http.StatusNotFound, errorResp)
+		s3err.WriteXMLResponse(w, r, http.StatusNotFound, errorResp)
 	case iam.ErrCodeServiceFailureException:
-		s3err.WriteXMLResponse(w, http.StatusInternalServerError, errorResp)
+		s3err.WriteXMLResponse(w, r, http.StatusInternalServerError, errorResp)
 	default:
-		s3err.WriteXMLResponse(w, http.StatusInternalServerError, errorResp)
+		s3err.WriteXMLResponse(w, r, http.StatusInternalServerError, errorResp)
 	}
 }

weed/iamapi/iamapi_management_handlers.go

@@ -362,7 +362,7 @@ func (iama *IamApiServer) DeleteAccessKey(s3cfg *iam_pb.S3ApiConfiguration, valu
 
 func (iama *IamApiServer) DoActions(w http.ResponseWriter, r *http.Request) {
 	if err := r.ParseForm(); err != nil {
-		s3err.WriteErrorResponse(w, s3err.ErrInvalidRequest, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInvalidRequest)
 		return
 	}
 	values := r.PostForm
@@ -370,7 +370,7 @@ func (iama *IamApiServer) DoActions(w http.ResponseWriter, r *http.Request) {
 	s3cfgLock.RLock()
 	s3cfg := &iam_pb.S3ApiConfiguration{}
 	if err := iama.s3ApiConfig.GetS3ApiConfiguration(s3cfg); err != nil {
-		s3err.WriteErrorResponse(w, s3err.ErrInternalError, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 		return
 	}
 	s3cfgLock.RUnlock()
@@ -392,7 +392,7 @@ func (iama *IamApiServer) DoActions(w http.ResponseWriter, r *http.Request) {
 		userName := values.Get("UserName")
 		response, err = iama.GetUser(s3cfg, userName)
 		if err != nil {
-			writeIamErrorResponse(w, err, "user", userName, nil)
+			writeIamErrorResponse(w, r, err, "user", userName, nil)
 			return
 		}
 		changed = false
@@ -400,7 +400,7 @@ func (iama *IamApiServer) DoActions(w http.ResponseWriter, r *http.Request) {
 		userName := values.Get("UserName")
 		response, err = iama.DeleteUser(s3cfg, userName)
 		if err != nil {
-			writeIamErrorResponse(w, err, "user", userName, nil)
+			writeIamErrorResponse(w, r, err, "user", userName, nil)
 			return
 		}
 	case "CreateAccessKey":
@@ -411,33 +411,33 @@ func (iama *IamApiServer) DoActions(w http.ResponseWriter, r *http.Request) {
 		response, err = iama.CreatePolicy(s3cfg, values)
 		if err != nil {
 			glog.Errorf("CreatePolicy:  %+v", err)
-			s3err.WriteErrorResponse(w, s3err.ErrInvalidRequest, r)
+			s3err.WriteErrorResponse(w, r, s3err.ErrInvalidRequest)
 			return
 		}
 	case "PutUserPolicy":
 		response, err = iama.PutUserPolicy(s3cfg, values)
 		if err != nil {
 			glog.Errorf("PutUserPolicy:  %+v", err)
-			s3err.WriteErrorResponse(w, s3err.ErrInvalidRequest, r)
+			s3err.WriteErrorResponse(w, r, s3err.ErrInvalidRequest)
 			return
 		}
 	case "GetUserPolicy":
 		response, err = iama.GetUserPolicy(s3cfg, values)
 		if err != nil {
-			writeIamErrorResponse(w, err, "user", values.Get("UserName"), nil)
+			writeIamErrorResponse(w, r, err, "user", values.Get("UserName"), nil)
 			return
 		}
 		changed = false
 	case "DeleteUserPolicy":
 		if response, err = iama.DeleteUserPolicy(s3cfg, values); err != nil {
-			writeIamErrorResponse(w, err, "user", values.Get("UserName"), nil)
+			writeIamErrorResponse(w, r, err, "user", values.Get("UserName"), nil)
 		}
 	default:
 		errNotImplemented := s3err.GetAPIError(s3err.ErrNotImplemented)
 		errorResponse := ErrorResponse{}
 		errorResponse.Error.Code = &errNotImplemented.Code
 		errorResponse.Error.Message = &errNotImplemented.Description
-		s3err.WriteXMLResponse(w, errNotImplemented.HTTPStatusCode, errorResponse)
+		s3err.WriteXMLResponse(w, r, errNotImplemented.HTTPStatusCode, errorResponse)
 		return
 	}
 	if changed {
@@ -445,9 +445,9 @@ func (iama *IamApiServer) DoActions(w http.ResponseWriter, r *http.Request) {
 		err := iama.s3ApiConfig.PutS3ApiConfiguration(s3cfg)
 		s3cfgLock.Unlock()
 		if err != nil {
-			writeIamErrorResponse(w, fmt.Errorf(iam.ErrCodeServiceFailureException), "", "", err)
+			writeIamErrorResponse(w, r, fmt.Errorf(iam.ErrCodeServiceFailureException), "", "", err)
 			return
 		}
 	}
-	s3err.WriteXMLResponse(w, http.StatusOK, response)
+	s3err.WriteXMLResponse(w, r, http.StatusOK, response)
 }

weed/pb/filer.proto

@@ -69,7 +69,7 @@ service SeaweedFiler {
     rpc KvPut (KvPutRequest) returns (KvPutResponse) {
     }
 
-    rpc DownloadToLocal (DownloadToLocalRequest) returns (DownloadToLocalResponse) {
+    rpc CacheRemoteObjectToLocalCluster (CacheRemoteObjectToLocalClusterRequest) returns (CacheRemoteObjectToLocalClusterResponse) {
     }
 }
 
@@ -403,10 +403,10 @@ message FilerConf {
 /////////////////////////
 // Remote Storage related
 /////////////////////////
-message DownloadToLocalRequest {
+message CacheRemoteObjectToLocalClusterRequest {
     string directory = 1;
     string name = 2;
 }
-message DownloadToLocalResponse {
+message CacheRemoteObjectToLocalClusterResponse {
     Entry entry = 1;
 }

weed/pb/filer_pb/filer.pb.go

@@ -3334,7 +3334,7 @@ func (x *FilerConf) GetLocations() []*FilerConf_PathConf {
 /////////////////////////
 // Remote Storage related
 /////////////////////////
-type DownloadToLocalRequest struct {
+type CacheRemoteObjectToLocalClusterRequest struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
@@ -3343,8 +3343,8 @@ type DownloadToLocalRequest struct {
 	Name      string `protobuf:"bytes,2,opt,name=name,proto3" json:"name,omitempty"`
 }
 
-func (x *DownloadToLocalRequest) Reset() {
-	*x = DownloadToLocalRequest{}
+func (x *CacheRemoteObjectToLocalClusterRequest) Reset() {
+	*x = CacheRemoteObjectToLocalClusterRequest{}
 	if protoimpl.UnsafeEnabled {
 		mi := &file_filer_proto_msgTypes[51]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
@@ -3352,13 +3352,13 @@ func (x *DownloadToLocalRequest) Reset() {
 	}
 }
 
-func (x *DownloadToLocalRequest) String() string {
+func (x *CacheRemoteObjectToLocalClusterRequest) String() string {
 	return protoimpl.X.MessageStringOf(x)
 }
 
-func (*DownloadToLocalRequest) ProtoMessage() {}
+func (*CacheRemoteObjectToLocalClusterRequest) ProtoMessage() {}
 
-func (x *DownloadToLocalRequest) ProtoReflect() protoreflect.Message {
+func (x *CacheRemoteObjectToLocalClusterRequest) ProtoReflect() protoreflect.Message {
 	mi := &file_filer_proto_msgTypes[51]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
@@ -3370,26 +3370,26 @@ func (x *DownloadToLocalRequest) ProtoReflect() protoreflect.Message {
 	return mi.MessageOf(x)
 }
 
-// Deprecated: Use DownloadToLocalRequest.ProtoReflect.Descriptor instead.
-func (*DownloadToLocalRequest) Descriptor() ([]byte, []int) {
+// Deprecated: Use CacheRemoteObjectToLocalClusterRequest.ProtoReflect.Descriptor instead.
+func (*CacheRemoteObjectToLocalClusterRequest) Descriptor() ([]byte, []int) {
 	return file_filer_proto_rawDescGZIP(), []int{51}
 }
 
-func (x *DownloadToLocalRequest) GetDirectory() string {
+func (x *CacheRemoteObjectToLocalClusterRequest) GetDirectory() string {
 	if x != nil {
 		return x.Directory
 	}
 	return ""
 }
 
-func (x *DownloadToLocalRequest) GetName() string {
+func (x *CacheRemoteObjectToLocalClusterRequest) GetName() string {
 	if x != nil {
 		return x.Name
 	}
 	return ""
 }
 
-type DownloadToLocalResponse struct {
+type CacheRemoteObjectToLocalClusterResponse struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
@@ -3397,8 +3397,8 @@ type DownloadToLocalResponse struct {
 	Entry *Entry `protobuf:"bytes,1,opt,name=entry,proto3" json:"entry,omitempty"`
 }
 
-func (x *DownloadToLocalResponse) Reset() {
-	*x = DownloadToLocalResponse{}
+func (x *CacheRemoteObjectToLocalClusterResponse) Reset() {
+	*x = CacheRemoteObjectToLocalClusterResponse{}
 	if protoimpl.UnsafeEnabled {
 		mi := &file_filer_proto_msgTypes[52]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
@@ -3406,13 +3406,13 @@ func (x *DownloadToLocalResponse) Reset() {
 	}
 }
 
-func (x *DownloadToLocalResponse) String() string {
+func (x *CacheRemoteObjectToLocalClusterResponse) String() string {
 	return protoimpl.X.MessageStringOf(x)
 }
 
-func (*DownloadToLocalResponse) ProtoMessage() {}
+func (*CacheRemoteObjectToLocalClusterResponse) ProtoMessage() {}
 
-func (x *DownloadToLocalResponse) ProtoReflect() protoreflect.Message {
+func (x *CacheRemoteObjectToLocalClusterResponse) ProtoReflect() protoreflect.Message {
 	mi := &file_filer_proto_msgTypes[52]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
@@ -3424,12 +3424,12 @@ func (x *DownloadToLocalResponse) ProtoReflect() protoreflect.Message {
 	return mi.MessageOf(x)
 }
 
-// Deprecated: Use DownloadToLocalResponse.ProtoReflect.Descriptor instead.
-func (*DownloadToLocalResponse) Descriptor() ([]byte, []int) {
+// Deprecated: Use CacheRemoteObjectToLocalClusterResponse.ProtoReflect.Descriptor instead.
+func (*CacheRemoteObjectToLocalClusterResponse) Descriptor() ([]byte, []int) {
 	return file_filer_proto_rawDescGZIP(), []int{52}
 }
 
-func (x *DownloadToLocalResponse) GetEntry() *Entry {
+func (x *CacheRemoteObjectToLocalClusterResponse) GetEntry() *Entry {
 	if x != nil {
 		return x.Entry
 	}
@@ -4042,16 +4042,18 @@ var file_filer_proto_rawDesc = []byte{
 	0x68, 0x5f, 0x63, 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x11, 0x76,
 	0x6f, 0x6c, 0x75, 0x6d, 0x65, 0x47, 0x72, 0x6f, 0x77, 0x74, 0x68, 0x43, 0x6f, 0x75, 0x6e, 0x74,
 	0x12, 0x1b, 0x0a, 0x09, 0x72, 0x65, 0x61, 0x64, 0x5f, 0x6f, 0x6e, 0x6c, 0x79, 0x18, 0x08, 0x20,
-	0x01, 0x28, 0x08, 0x52, 0x08, 0x72, 0x65, 0x61, 0x64, 0x4f, 0x6e, 0x6c, 0x79, 0x22, 0x4a, 0x0a,
-	0x16, 0x44, 0x6f, 0x77, 0x6e, 0x6c, 0x6f, 0x61, 0x64, 0x54, 0x6f, 0x4c, 0x6f, 0x63, 0x61, 0x6c,
+	0x01, 0x28, 0x08, 0x52, 0x08, 0x72, 0x65, 0x61, 0x64, 0x4f, 0x6e, 0x6c, 0x79, 0x22, 0x5a, 0x0a,
+	0x26, 0x43, 0x61, 0x63, 0x68, 0x65, 0x52, 0x65, 0x6d, 0x6f, 0x74, 0x65, 0x4f, 0x62, 0x6a, 0x65,
+	0x63, 0x74, 0x54, 0x6f, 0x4c, 0x6f, 0x63, 0x61, 0x6c, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72,
 	0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x1c, 0x0a, 0x09, 0x64, 0x69, 0x72, 0x65, 0x63,
 	0x74, 0x6f, 0x72, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x64, 0x69, 0x72, 0x65,
 	0x63, 0x74, 0x6f, 0x72, 0x79, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x02, 0x20,
-	0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x22, 0x40, 0x0a, 0x17, 0x44, 0x6f, 0x77,
-	0x6e, 0x6c, 0x6f, 0x61, 0x64, 0x54, 0x6f, 0x4c, 0x6f, 0x63, 0x61, 0x6c, 0x52, 0x65, 0x73, 0x70,
+	0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x22, 0x50, 0x0a, 0x27, 0x43, 0x61, 0x63,
+	0x68, 0x65, 0x52, 0x65, 0x6d, 0x6f, 0x74, 0x65, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x54, 0x6f,
+	0x4c, 0x6f, 0x63, 0x61, 0x6c, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x52, 0x65, 0x73, 0x70,
 	0x6f, 0x6e, 0x73, 0x65, 0x12, 0x25, 0x0a, 0x05, 0x65, 0x6e, 0x74, 0x72, 0x79, 0x18, 0x01, 0x20,
 	0x01, 0x28, 0x0b, 0x32, 0x0f, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x45,
-	0x6e, 0x74, 0x72, 0x79, 0x52, 0x05, 0x65, 0x6e, 0x74, 0x72, 0x79, 0x32, 0x98, 0x0e, 0x0a, 0x0c,
+	0x6e, 0x74, 0x72, 0x79, 0x52, 0x05, 0x65, 0x6e, 0x74, 0x72, 0x79, 0x32, 0xc9, 0x0e, 0x0a, 0x0c,
 	0x53, 0x65, 0x61, 0x77, 0x65, 0x65, 0x64, 0x46, 0x69, 0x6c, 0x65, 0x72, 0x12, 0x67, 0x0a, 0x14,
 	0x4c, 0x6f, 0x6f, 0x6b, 0x75, 0x70, 0x44, 0x69, 0x72, 0x65, 0x63, 0x74, 0x6f, 0x72, 0x79, 0x45,
 	0x6e, 0x74, 0x72, 0x79, 0x12, 0x25, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e,
@@ -4159,18 +4161,21 @@ var file_filer_proto_rawDesc = []byte{
 	0x12, 0x3a, 0x0a, 0x05, 0x4b, 0x76, 0x50, 0x75, 0x74, 0x12, 0x16, 0x2e, 0x66, 0x69, 0x6c, 0x65,
 	0x72, 0x5f, 0x70, 0x62, 0x2e, 0x4b, 0x76, 0x50, 0x75, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
 	0x74, 0x1a, 0x17, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x4b, 0x76, 0x50,
-	0x75, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x58, 0x0a, 0x0f,
-	0x44, 0x6f, 0x77, 0x6e, 0x6c, 0x6f, 0x61, 0x64, 0x54, 0x6f, 0x4c, 0x6f, 0x63, 0x61, 0x6c, 0x12,
-	0x20, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x44, 0x6f, 0x77, 0x6e, 0x6c,
-	0x6f, 0x61, 0x64, 0x54, 0x6f, 0x4c, 0x6f, 0x63, 0x61, 0x6c, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
-	0x74, 0x1a, 0x21, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x44, 0x6f, 0x77,
-	0x6e, 0x6c, 0x6f, 0x61, 0x64, 0x54, 0x6f, 0x4c, 0x6f, 0x63, 0x61, 0x6c, 0x52, 0x65, 0x73, 0x70,
-	0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x42, 0x4f, 0x0a, 0x10, 0x73, 0x65, 0x61, 0x77, 0x65, 0x65,
-	0x64, 0x66, 0x73, 0x2e, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x42, 0x0a, 0x46, 0x69, 0x6c, 0x65,
-	0x72, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x5a, 0x2f, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63,
-	0x6f, 0x6d, 0x2f, 0x63, 0x68, 0x72, 0x69, 0x73, 0x6c, 0x75, 0x73, 0x66, 0x2f, 0x73, 0x65, 0x61,
-	0x77, 0x65, 0x65, 0x64, 0x66, 0x73, 0x2f, 0x77, 0x65, 0x65, 0x64, 0x2f, 0x70, 0x62, 0x2f, 0x66,
-	0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x75, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x88, 0x01, 0x0a,
+	0x1f, 0x43, 0x61, 0x63, 0x68, 0x65, 0x52, 0x65, 0x6d, 0x6f, 0x74, 0x65, 0x4f, 0x62, 0x6a, 0x65,
+	0x63, 0x74, 0x54, 0x6f, 0x4c, 0x6f, 0x63, 0x61, 0x6c, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72,
+	0x12, 0x30, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x43, 0x61, 0x63, 0x68,
+	0x65, 0x52, 0x65, 0x6d, 0x6f, 0x74, 0x65, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x54, 0x6f, 0x4c,
+	0x6f, 0x63, 0x61, 0x6c, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65,
+	0x73, 0x74, 0x1a, 0x31, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x43, 0x61,
+	0x63, 0x68, 0x65, 0x52, 0x65, 0x6d, 0x6f, 0x74, 0x65, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x54,
+	0x6f, 0x4c, 0x6f, 0x63, 0x61, 0x6c, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x52, 0x65, 0x73,
+	0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x42, 0x4f, 0x0a, 0x10, 0x73, 0x65, 0x61, 0x77, 0x65,
+	0x65, 0x64, 0x66, 0x73, 0x2e, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x42, 0x0a, 0x46, 0x69, 0x6c,
+	0x65, 0x72, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x5a, 0x2f, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e,
+	0x63, 0x6f, 0x6d, 0x2f, 0x63, 0x68, 0x72, 0x69, 0x73, 0x6c, 0x75, 0x73, 0x66, 0x2f, 0x73, 0x65,
+	0x61, 0x77, 0x65, 0x65, 0x64, 0x66, 0x73, 0x2f, 0x77, 0x65, 0x65, 0x64, 0x2f, 0x70, 0x62, 0x2f,
+	0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -4187,59 +4192,59 @@ func file_filer_proto_rawDescGZIP() []byte {
 
 var file_filer_proto_msgTypes = make([]protoimpl.MessageInfo, 57)
 var file_filer_proto_goTypes = []interface{}{
-	(*LookupDirectoryEntryRequest)(nil),   // 0: filer_pb.LookupDirectoryEntryRequest
-	(*LookupDirectoryEntryResponse)(nil),  // 1: filer_pb.LookupDirectoryEntryResponse
-	(*ListEntriesRequest)(nil),            // 2: filer_pb.ListEntriesRequest
-	(*ListEntriesResponse)(nil),           // 3: filer_pb.ListEntriesResponse
-	(*RemoteEntry)(nil),                   // 4: filer_pb.RemoteEntry
-	(*Entry)(nil),                         // 5: filer_pb.Entry
-	(*FullEntry)(nil),                     // 6: filer_pb.FullEntry
-	(*EventNotification)(nil),             // 7: filer_pb.EventNotification
-	(*FileChunk)(nil),                     // 8: filer_pb.FileChunk
-	(*FileChunkManifest)(nil),             // 9: filer_pb.FileChunkManifest
-	(*FileId)(nil),                        // 10: filer_pb.FileId
-	(*FuseAttributes)(nil),                // 11: filer_pb.FuseAttributes
-	(*CreateEntryRequest)(nil),            // 12: filer_pb.CreateEntryRequest
-	(*CreateEntryResponse)(nil),           // 13: filer_pb.CreateEntryResponse
-	(*UpdateEntryRequest)(nil),            // 14: filer_pb.UpdateEntryRequest
-	(*UpdateEntryResponse)(nil),           // 15: filer_pb.UpdateEntryResponse
-	(*AppendToEntryRequest)(nil),          // 16: filer_pb.AppendToEntryRequest
-	(*AppendToEntryResponse)(nil),         // 17: filer_pb.AppendToEntryResponse
-	(*DeleteEntryRequest)(nil),            // 18: filer_pb.DeleteEntryRequest
-	(*DeleteEntryResponse)(nil),           // 19: filer_pb.DeleteEntryResponse
-	(*AtomicRenameEntryRequest)(nil),      // 20: filer_pb.AtomicRenameEntryRequest
-	(*AtomicRenameEntryResponse)(nil),     // 21: filer_pb.AtomicRenameEntryResponse
-	(*StreamRenameEntryRequest)(nil),      // 22: filer_pb.StreamRenameEntryRequest
-	(*StreamRenameEntryResponse)(nil),     // 23: filer_pb.StreamRenameEntryResponse
-	(*AssignVolumeRequest)(nil),           // 24: filer_pb.AssignVolumeRequest
-	(*AssignVolumeResponse)(nil),          // 25: filer_pb.AssignVolumeResponse
-	(*LookupVolumeRequest)(nil),           // 26: filer_pb.LookupVolumeRequest
-	(*Locations)(nil),                     // 27: filer_pb.Locations
-	(*Location)(nil),                      // 28: filer_pb.Location
-	(*LookupVolumeResponse)(nil),          // 29: filer_pb.LookupVolumeResponse
-	(*Collection)(nil),                    // 30: filer_pb.Collection
-	(*CollectionListRequest)(nil),         // 31: filer_pb.CollectionListRequest
-	(*CollectionListResponse)(nil),        // 32: filer_pb.CollectionListResponse
-	(*DeleteCollectionRequest)(nil),       // 33: filer_pb.DeleteCollectionRequest
-	(*DeleteCollectionResponse)(nil),      // 34: filer_pb.DeleteCollectionResponse
-	(*StatisticsRequest)(nil),             // 35: filer_pb.StatisticsRequest
-	(*StatisticsResponse)(nil),            // 36: filer_pb.StatisticsResponse
-	(*GetFilerConfigurationRequest)(nil),  // 37: filer_pb.GetFilerConfigurationRequest
-	(*GetFilerConfigurationResponse)(nil), // 38: filer_pb.GetFilerConfigurationResponse
-	(*SubscribeMetadataRequest)(nil),      // 39: filer_pb.SubscribeMetadataRequest
-	(*SubscribeMetadataResponse)(nil),     // 40: filer_pb.SubscribeMetadataResponse
-	(*LogEntry)(nil),                      // 41: filer_pb.LogEntry
-	(*KeepConnectedRequest)(nil),          // 42: filer_pb.KeepConnectedRequest
-	(*KeepConnectedResponse)(nil),         // 43: filer_pb.KeepConnectedResponse
-	(*LocateBrokerRequest)(nil),           // 44: filer_pb.LocateBrokerRequest
-	(*LocateBrokerResponse)(nil),          // 45: filer_pb.LocateBrokerResponse
-	(*KvGetRequest)(nil),                  // 46: filer_pb.KvGetRequest
-	(*KvGetResponse)(nil),                 // 47: filer_pb.KvGetResponse
-	(*KvPutRequest)(nil),                  // 48: filer_pb.KvPutRequest
-	(*KvPutResponse)(nil),                 // 49: filer_pb.KvPutResponse
-	(*FilerConf)(nil),                     // 50: filer_pb.FilerConf
-	(*DownloadToLocalRequest)(nil),        // 51: filer_pb.DownloadToLocalRequest
-	(*DownloadToLocalResponse)(nil),       // 52: filer_pb.DownloadToLocalResponse
+	(*LookupDirectoryEntryRequest)(nil),             // 0: filer_pb.LookupDirectoryEntryRequest
+	(*LookupDirectoryEntryResponse)(nil),            // 1: filer_pb.LookupDirectoryEntryResponse
+	(*ListEntriesRequest)(nil),                      // 2: filer_pb.ListEntriesRequest
+	(*ListEntriesResponse)(nil),                     // 3: filer_pb.ListEntriesResponse
+	(*RemoteEntry)(nil),                             // 4: filer_pb.RemoteEntry
+	(*Entry)(nil),                                   // 5: filer_pb.Entry
+	(*FullEntry)(nil),                               // 6: filer_pb.FullEntry
+	(*EventNotification)(nil),                       // 7: filer_pb.EventNotification
+	(*FileChunk)(nil),                               // 8: filer_pb.FileChunk
+	(*FileChunkManifest)(nil),                       // 9: filer_pb.FileChunkManifest
+	(*FileId)(nil),                                  // 10: filer_pb.FileId
+	(*FuseAttributes)(nil),                          // 11: filer_pb.FuseAttributes
+	(*CreateEntryRequest)(nil),                      // 12: filer_pb.CreateEntryRequest
+	(*CreateEntryResponse)(nil),                     // 13: filer_pb.CreateEntryResponse
+	(*UpdateEntryRequest)(nil),                      // 14: filer_pb.UpdateEntryRequest
+	(*UpdateEntryResponse)(nil),                     // 15: filer_pb.UpdateEntryResponse
+	(*AppendToEntryRequest)(nil),                    // 16: filer_pb.AppendToEntryRequest
+	(*AppendToEntryResponse)(nil),                   // 17: filer_pb.AppendToEntryResponse
+	(*DeleteEntryRequest)(nil),                      // 18: filer_pb.DeleteEntryRequest
+	(*DeleteEntryResponse)(nil),                     // 19: filer_pb.DeleteEntryResponse
+	(*AtomicRenameEntryRequest)(nil),                // 20: filer_pb.AtomicRenameEntryRequest
+	(*AtomicRenameEntryResponse)(nil),               // 21: filer_pb.AtomicRenameEntryResponse
+	(*StreamRenameEntryRequest)(nil),                // 22: filer_pb.StreamRenameEntryRequest
+	(*StreamRenameEntryResponse)(nil),               // 23: filer_pb.StreamRenameEntryResponse
+	(*AssignVolumeRequest)(nil),                     // 24: filer_pb.AssignVolumeRequest
+	(*AssignVolumeResponse)(nil),                    // 25: filer_pb.AssignVolumeResponse
+	(*LookupVolumeRequest)(nil),                     // 26: filer_pb.LookupVolumeRequest
+	(*Locations)(nil),                               // 27: filer_pb.Locations
+	(*Location)(nil),                                // 28: filer_pb.Location
+	(*LookupVolumeResponse)(nil),                    // 29: filer_pb.LookupVolumeResponse
+	(*Collection)(nil),                              // 30: filer_pb.Collection
+	(*CollectionListRequest)(nil),                   // 31: filer_pb.CollectionListRequest
+	(*CollectionListResponse)(nil),                  // 32: filer_pb.CollectionListResponse
+	(*DeleteCollectionRequest)(nil),                 // 33: filer_pb.DeleteCollectionRequest
+	(*DeleteCollectionResponse)(nil),                // 34: filer_pb.DeleteCollectionResponse
+	(*StatisticsRequest)(nil),                       // 35: filer_pb.StatisticsRequest
+	(*StatisticsResponse)(nil),                      // 36: filer_pb.StatisticsResponse
+	(*GetFilerConfigurationRequest)(nil),            // 37: filer_pb.GetFilerConfigurationRequest
+	(*GetFilerConfigurationResponse)(nil),           // 38: filer_pb.GetFilerConfigurationResponse
+	(*SubscribeMetadataRequest)(nil),                // 39: filer_pb.SubscribeMetadataRequest
+	(*SubscribeMetadataResponse)(nil),               // 40: filer_pb.SubscribeMetadataResponse
+	(*LogEntry)(nil),                                // 41: filer_pb.LogEntry
+	(*KeepConnectedRequest)(nil),                    // 42: filer_pb.KeepConnectedRequest
+	(*KeepConnectedResponse)(nil),                   // 43: filer_pb.KeepConnectedResponse
+	(*LocateBrokerRequest)(nil),                     // 44: filer_pb.LocateBrokerRequest
+	(*LocateBrokerResponse)(nil),                    // 45: filer_pb.LocateBrokerResponse
+	(*KvGetRequest)(nil),                            // 46: filer_pb.KvGetRequest
+	(*KvGetResponse)(nil),                           // 47: filer_pb.KvGetResponse
+	(*KvPutRequest)(nil),                            // 48: filer_pb.KvPutRequest
+	(*KvPutResponse)(nil),                           // 49: filer_pb.KvPutResponse
+	(*FilerConf)(nil),                               // 50: filer_pb.FilerConf
+	(*CacheRemoteObjectToLocalClusterRequest)(nil),  // 51: filer_pb.CacheRemoteObjectToLocalClusterRequest
+	(*CacheRemoteObjectToLocalClusterResponse)(nil), // 52: filer_pb.CacheRemoteObjectToLocalClusterResponse
 	nil,                                   // 53: filer_pb.Entry.ExtendedEntry
 	nil,                                   // 54: filer_pb.LookupVolumeResponse.LocationsMapEntry
 	(*LocateBrokerResponse_Resource)(nil), // 55: filer_pb.LocateBrokerResponse.Resource
@@ -4269,7 +4274,7 @@ var file_filer_proto_depIdxs = []int32{
 	7,  // 20: filer_pb.SubscribeMetadataResponse.event_notification:type_name -> filer_pb.EventNotification
 	55, // 21: filer_pb.LocateBrokerResponse.resources:type_name -> filer_pb.LocateBrokerResponse.Resource
 	56, // 22: filer_pb.FilerConf.locations:type_name -> filer_pb.FilerConf.PathConf
-	5,  // 23: filer_pb.DownloadToLocalResponse.entry:type_name -> filer_pb.Entry
+	5,  // 23: filer_pb.CacheRemoteObjectToLocalClusterResponse.entry:type_name -> filer_pb.Entry
 	27, // 24: filer_pb.LookupVolumeResponse.LocationsMapEntry.value:type_name -> filer_pb.Locations
 	0,  // 25: filer_pb.SeaweedFiler.LookupDirectoryEntry:input_type -> filer_pb.LookupDirectoryEntryRequest
 	2,  // 26: filer_pb.SeaweedFiler.ListEntries:input_type -> filer_pb.ListEntriesRequest
@@ -4291,7 +4296,7 @@ var file_filer_proto_depIdxs = []int32{
 	44, // 42: filer_pb.SeaweedFiler.LocateBroker:input_type -> filer_pb.LocateBrokerRequest
 	46, // 43: filer_pb.SeaweedFiler.KvGet:input_type -> filer_pb.KvGetRequest
 	48, // 44: filer_pb.SeaweedFiler.KvPut:input_type -> filer_pb.KvPutRequest
-	51, // 45: filer_pb.SeaweedFiler.DownloadToLocal:input_type -> filer_pb.DownloadToLocalRequest
+	51, // 45: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:input_type -> filer_pb.CacheRemoteObjectToLocalClusterRequest
 	1,  // 46: filer_pb.SeaweedFiler.LookupDirectoryEntry:output_type -> filer_pb.LookupDirectoryEntryResponse
 	3,  // 47: filer_pb.SeaweedFiler.ListEntries:output_type -> filer_pb.ListEntriesResponse
 	13, // 48: filer_pb.SeaweedFiler.CreateEntry:output_type -> filer_pb.CreateEntryResponse
@@ -4312,7 +4317,7 @@ var file_filer_proto_depIdxs = []int32{
 	45, // 63: filer_pb.SeaweedFiler.LocateBroker:output_type -> filer_pb.LocateBrokerResponse
 	47, // 64: filer_pb.SeaweedFiler.KvGet:output_type -> filer_pb.KvGetResponse
 	49, // 65: filer_pb.SeaweedFiler.KvPut:output_type -> filer_pb.KvPutResponse
-	52, // 66: filer_pb.SeaweedFiler.DownloadToLocal:output_type -> filer_pb.DownloadToLocalResponse
+	52, // 66: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:output_type -> filer_pb.CacheRemoteObjectToLocalClusterResponse
 	46, // [46:67] is the sub-list for method output_type
 	25, // [25:46] is the sub-list for method input_type
 	25, // [25:25] is the sub-list for extension type_name
@@ -4939,7 +4944,7 @@ func file_filer_proto_init() {
 			}
 		}
 		file_filer_proto_msgTypes[51].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*DownloadToLocalRequest); i {
+			switch v := v.(*CacheRemoteObjectToLocalClusterRequest); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -4951,7 +4956,7 @@ func file_filer_proto_init() {
 			}
 		}
 		file_filer_proto_msgTypes[52].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*DownloadToLocalResponse); i {
+			switch v := v.(*CacheRemoteObjectToLocalClusterResponse); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -5039,7 +5044,7 @@ type SeaweedFilerClient interface {
 	LocateBroker(ctx context.Context, in *LocateBrokerRequest, opts ...grpc.CallOption) (*LocateBrokerResponse, error)
 	KvGet(ctx context.Context, in *KvGetRequest, opts ...grpc.CallOption) (*KvGetResponse, error)
 	KvPut(ctx context.Context, in *KvPutRequest, opts ...grpc.CallOption) (*KvPutResponse, error)
-	DownloadToLocal(ctx context.Context, in *DownloadToLocalRequest, opts ...grpc.CallOption) (*DownloadToLocalResponse, error)
+	CacheRemoteObjectToLocalCluster(ctx context.Context, in *CacheRemoteObjectToLocalClusterRequest, opts ...grpc.CallOption) (*CacheRemoteObjectToLocalClusterResponse, error)
 }
 
 type seaweedFilerClient struct {
@@ -5344,9 +5349,9 @@ func (c *seaweedFilerClient) KvPut(ctx context.Context, in *KvPutRequest, opts .
 	return out, nil
 }
 
-func (c *seaweedFilerClient) DownloadToLocal(ctx context.Context, in *DownloadToLocalRequest, opts ...grpc.CallOption) (*DownloadToLocalResponse, error) {
-	out := new(DownloadToLocalResponse)
-	err := c.cc.Invoke(ctx, "/filer_pb.SeaweedFiler/DownloadToLocal", in, out, opts...)
+func (c *seaweedFilerClient) CacheRemoteObjectToLocalCluster(ctx context.Context, in *CacheRemoteObjectToLocalClusterRequest, opts ...grpc.CallOption) (*CacheRemoteObjectToLocalClusterResponse, error) {
+	out := new(CacheRemoteObjectToLocalClusterResponse)
+	err := c.cc.Invoke(ctx, "/filer_pb.SeaweedFiler/CacheRemoteObjectToLocalCluster", in, out, opts...)
 	if err != nil {
 		return nil, err
 	}
@@ -5375,7 +5380,7 @@ type SeaweedFilerServer interface {
 	LocateBroker(context.Context, *LocateBrokerRequest) (*LocateBrokerResponse, error)
 	KvGet(context.Context, *KvGetRequest) (*KvGetResponse, error)
 	KvPut(context.Context, *KvPutRequest) (*KvPutResponse, error)
-	DownloadToLocal(context.Context, *DownloadToLocalRequest) (*DownloadToLocalResponse, error)
+	CacheRemoteObjectToLocalCluster(context.Context, *CacheRemoteObjectToLocalClusterRequest) (*CacheRemoteObjectToLocalClusterResponse, error)
 }
 
 // UnimplementedSeaweedFilerServer can be embedded to have forward compatible implementations.
@@ -5442,8 +5447,8 @@ func (*UnimplementedSeaweedFilerServer) KvGet(context.Context, *KvGetRequest) (*
 func (*UnimplementedSeaweedFilerServer) KvPut(context.Context, *KvPutRequest) (*KvPutResponse, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method KvPut not implemented")
 }
-func (*UnimplementedSeaweedFilerServer) DownloadToLocal(context.Context, *DownloadToLocalRequest) (*DownloadToLocalResponse, error) {
-	return nil, status.Errorf(codes.Unimplemented, "method DownloadToLocal not implemented")
+func (*UnimplementedSeaweedFilerServer) CacheRemoteObjectToLocalCluster(context.Context, *CacheRemoteObjectToLocalClusterRequest) (*CacheRemoteObjectToLocalClusterResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method CacheRemoteObjectToLocalCluster not implemented")
 }
 
 func RegisterSeaweedFilerServer(s *grpc.Server, srv SeaweedFilerServer) {
@@ -5830,20 +5835,20 @@ func _SeaweedFiler_KvPut_Handler(srv interface{}, ctx context.Context, dec func(
 	return interceptor(ctx, in, info, handler)
 }
 
-func _SeaweedFiler_DownloadToLocal_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
-	in := new(DownloadToLocalRequest)
+func _SeaweedFiler_CacheRemoteObjectToLocalCluster_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(CacheRemoteObjectToLocalClusterRequest)
 	if err := dec(in); err != nil {
 		return nil, err
 	}
 	if interceptor == nil {
-		return srv.(SeaweedFilerServer).DownloadToLocal(ctx, in)
+		return srv.(SeaweedFilerServer).CacheRemoteObjectToLocalCluster(ctx, in)
 	}
 	info := &grpc.UnaryServerInfo{
 		Server:     srv,
-		FullMethod: "/filer_pb.SeaweedFiler/DownloadToLocal",
+		FullMethod: "/filer_pb.SeaweedFiler/CacheRemoteObjectToLocalCluster",
 	}
 	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
-		return srv.(SeaweedFilerServer).DownloadToLocal(ctx, req.(*DownloadToLocalRequest))
+		return srv.(SeaweedFilerServer).CacheRemoteObjectToLocalCluster(ctx, req.(*CacheRemoteObjectToLocalClusterRequest))
 	}
 	return interceptor(ctx, in, info, handler)
 }
@@ -5913,8 +5918,8 @@ var _SeaweedFiler_serviceDesc = grpc.ServiceDesc{
 			Handler:    _SeaweedFiler_KvPut_Handler,
 		},
 		{
-			MethodName: "DownloadToLocal",
-			Handler:    _SeaweedFiler_DownloadToLocal_Handler,
+			MethodName: "CacheRemoteObjectToLocalCluster",
+			Handler:    _SeaweedFiler_CacheRemoteObjectToLocalCluster_Handler,
 		},
 	},
 	Streams: []grpc.StreamDesc{

weed/pb/volume_server.proto

@@ -14,7 +14,7 @@ service VolumeServer {
 
     rpc VacuumVolumeCheck (VacuumVolumeCheckRequest) returns (VacuumVolumeCheckResponse) {
     }
-    rpc VacuumVolumeCompact (VacuumVolumeCompactRequest) returns (VacuumVolumeCompactResponse) {
+    rpc VacuumVolumeCompact (VacuumVolumeCompactRequest) returns (stream VacuumVolumeCompactResponse) {
     }
     rpc VacuumVolumeCommit (VacuumVolumeCommitRequest) returns (VacuumVolumeCommitResponse) {
     }
@@ -47,7 +47,7 @@ service VolumeServer {
     }
 
     // copy the .idx .dat files, and mount this volume
-    rpc VolumeCopy (VolumeCopyRequest) returns (VolumeCopyResponse) {
+    rpc VolumeCopy (VolumeCopyRequest) returns (stream VolumeCopyResponse) {
     }
     rpc ReadVolumeFileStatus (ReadVolumeFileStatusRequest) returns (ReadVolumeFileStatusResponse) {
     }
@@ -142,6 +142,7 @@ message VacuumVolumeCompactRequest {
     int64 preallocate = 2;
 }
 message VacuumVolumeCompactResponse {
+    int64 processed_bytes = 1;
 }
 
 message VacuumVolumeCommitRequest {
@@ -251,6 +252,7 @@ message VolumeCopyRequest {
 }
 message VolumeCopyResponse {
     uint64 last_append_at_ns = 1;
+    int64 processed_bytes = 2;
 }
 
 message CopyFileRequest {

weed/s3api/auth_credentials.go

@@ -182,7 +182,7 @@ func (iam *IdentityAccessManagement) Auth(f http.HandlerFunc, action Action) htt
 			f(w, r)
 			return
 		}
-		s3err.WriteErrorResponse(w, errCode, r)
+		s3err.WriteErrorResponse(w, r, errCode)
 	}
 }
 

weed/s3api/filer_multipart.go

@@ -68,7 +68,7 @@ func (s3a *S3ApiServer) completeMultipartUpload(input *s3.CompleteMultipartUploa
 
 	uploadDirectory := s3a.genUploadsFolder(*input.Bucket) + "/" + *input.UploadId
 
-	entries, _, err := s3a.list(uploadDirectory, "", "", false, 0)
+	entries, _, err := s3a.list(uploadDirectory, "", "", false, maxPartsList)
 	if err != nil || len(entries) == 0 {
 		glog.Errorf("completeMultipartUpload %s %s error: %v, entries:%d", *input.Bucket, *input.UploadId, err, len(entries))
 		return nil, s3err.ErrNoSuchUpload

weed/s3api/s3api_bucket_handlers.go

@@ -36,7 +36,7 @@ func (s3a *S3ApiServer) ListBucketsHandler(w http.ResponseWriter, r *http.Reques
 	if s3a.iam.isEnabled() {
 		identity, s3Err = s3a.iam.authUser(r)
 		if s3Err != s3err.ErrNone {
-			s3err.WriteErrorResponse(w, s3Err, r)
+			s3err.WriteErrorResponse(w, r, s3Err)
 			return
 		}
 	}
@@ -46,7 +46,7 @@ func (s3a *S3ApiServer) ListBucketsHandler(w http.ResponseWriter, r *http.Reques
 	entries, _, err := s3a.list(s3a.option.BucketsPath, "", "", false, math.MaxInt32)
 
 	if err != nil {
-		s3err.WriteErrorResponse(w, s3err.ErrInternalError, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 		return
 	}
 
@@ -73,7 +73,7 @@ func (s3a *S3ApiServer) ListBucketsHandler(w http.ResponseWriter, r *http.Reques
 		Buckets: buckets,
 	}
 
-	writeSuccessResponseXML(w, response)
+	writeSuccessResponseXML(w, r, response)
 }
 
 func (s3a *S3ApiServer) PutBucketHandler(w http.ResponseWriter, r *http.Request) {
@@ -100,14 +100,14 @@ func (s3a *S3ApiServer) PutBucketHandler(w http.ResponseWriter, r *http.Request)
 		}
 		return nil
 	}); err != nil {
-		s3err.WriteErrorResponse(w, s3err.ErrInternalError, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 		return
 	}
 	if exist, err := s3a.exists(s3a.option.BucketsPath, bucket, true); err == nil && exist {
 		errCode = s3err.ErrBucketAlreadyExists
 	}
 	if errCode != s3err.ErrNone {
-		s3err.WriteErrorResponse(w, errCode, r)
+		s3err.WriteErrorResponse(w, r, errCode)
 		return
 	}
 
@@ -123,11 +123,11 @@ func (s3a *S3ApiServer) PutBucketHandler(w http.ResponseWriter, r *http.Request)
 	// create the folder for bucket, but lazily create actual collection
 	if err := s3a.mkdir(s3a.option.BucketsPath, bucket, fn); err != nil {
 		glog.Errorf("PutBucketHandler mkdir: %v", err)
-		s3err.WriteErrorResponse(w, s3err.ErrInternalError, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 		return
 	}
 
-	writeSuccessResponseEmpty(w)
+	writeSuccessResponseEmpty(w, r)
 }
 
 func (s3a *S3ApiServer) DeleteBucketHandler(w http.ResponseWriter, r *http.Request) {
@@ -136,7 +136,7 @@ func (s3a *S3ApiServer) DeleteBucketHandler(w http.ResponseWriter, r *http.Reque
 	glog.V(3).Infof("DeleteBucketHandler %s", bucket)
 
 	if err := s3a.checkBucket(r, bucket); err != s3err.ErrNone {
-		s3err.WriteErrorResponse(w, err, r)
+		s3err.WriteErrorResponse(w, r, err)
 		return
 	}
 
@@ -158,11 +158,11 @@ func (s3a *S3ApiServer) DeleteBucketHandler(w http.ResponseWriter, r *http.Reque
 	err = s3a.rm(s3a.option.BucketsPath, bucket, false, true)
 
 	if err != nil {
-		s3err.WriteErrorResponse(w, s3err.ErrInternalError, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 		return
 	}
 
-	s3err.WriteEmptyResponse(w, http.StatusNoContent)
+	s3err.WriteEmptyResponse(w, r, http.StatusNoContent)
 }
 
 func (s3a *S3ApiServer) HeadBucketHandler(w http.ResponseWriter, r *http.Request) {
@@ -171,11 +171,11 @@ func (s3a *S3ApiServer) HeadBucketHandler(w http.ResponseWriter, r *http.Request
 	glog.V(3).Infof("HeadBucketHandler %s", bucket)
 
 	if err := s3a.checkBucket(r, bucket); err != s3err.ErrNone {
-		s3err.WriteErrorResponse(w, err, r)
+		s3err.WriteErrorResponse(w, r, err)
 		return
 	}
 
-	writeSuccessResponseEmpty(w)
+	writeSuccessResponseEmpty(w, r)
 }
 
 func (s3a *S3ApiServer) checkBucket(r *http.Request, bucket string) s3err.ErrorCode {
@@ -216,7 +216,7 @@ func (s3a *S3ApiServer) GetBucketAclHandler(w http.ResponseWriter, r *http.Reque
 	glog.V(3).Infof("GetBucketAclHandler %s", bucket)
 
 	if err := s3a.checkBucket(r, bucket); err != s3err.ErrNone {
-		s3err.WriteErrorResponse(w, err, r)
+		s3err.WriteErrorResponse(w, r, err)
 		return
 	}
 
@@ -245,7 +245,7 @@ func (s3a *S3ApiServer) GetBucketAclHandler(w http.ResponseWriter, r *http.Reque
 			})
 		}
 	}
-	writeSuccessResponseXML(w, response)
+	writeSuccessResponseXML(w, r, response)
 }
 
 // GetBucketLifecycleConfigurationHandler Get Bucket Lifecycle configuration
@@ -256,18 +256,19 @@ func (s3a *S3ApiServer) GetBucketLifecycleConfigurationHandler(w http.ResponseWr
 	glog.V(3).Infof("GetBucketAclHandler %s", bucket)
 
 	if err := s3a.checkBucket(r, bucket); err != s3err.ErrNone {
-		s3err.WriteErrorResponse(w, err, r)
+		s3err.WriteErrorResponse(w, r, err)
 		return
 	}
 	fc, err := filer.ReadFilerConf(s3a.option.Filer, s3a.option.GrpcDialOption, nil)
 	if err != nil {
 		glog.Errorf("GetBucketLifecycleConfigurationHandler: %s", err)
-		s3err.WriteErrorResponse(w, s3err.ErrInternalError, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 		return
 	}
 	ttls := fc.GetCollectionTtls(bucket)
 	if len(ttls) == 0 {
-		s3err.WriteErrorResponse(w, s3err.ErrNoSuchLifecycleConfiguration, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchLifecycleConfiguration)
+		return
 	}
 	response := Lifecycle{}
 	for prefix, internalTtl := range ttls {
@@ -284,14 +285,14 @@ func (s3a *S3ApiServer) GetBucketLifecycleConfigurationHandler(w http.ResponseWr
 			Expiration: Expiration{Days: days, set: true},
 		})
 	}
-	writeSuccessResponseXML(w, response)
+	writeSuccessResponseXML(w, r, response)
 }
 
 // PutBucketLifecycleConfigurationHandler Put Bucket Lifecycle configuration
 // https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutBucketLifecycleConfiguration.html
 func (s3a *S3ApiServer) PutBucketLifecycleConfigurationHandler(w http.ResponseWriter, r *http.Request) {
 
-	s3err.WriteErrorResponse(w, s3err.ErrNotImplemented, r)
+	s3err.WriteErrorResponse(w, r, s3err.ErrNotImplemented)
 
 }
 
@@ -299,6 +300,6 @@ func (s3a *S3ApiServer) PutBucketLifecycleConfigurationHandler(w http.ResponseWr
 // https://docs.aws.amazon.com/AmazonS3/latest/API/API_DeleteBucketLifecycle.html
 func (s3a *S3ApiServer) DeleteBucketLifecycleHandler(w http.ResponseWriter, r *http.Request) {
 
-	s3err.WriteErrorResponse(w, s3err.ErrNotImplemented, r)
+	s3err.WriteEmptyResponse(w, r, http.StatusNoContent)
 
 }

weed/s3api/s3api_handlers.go

@@ -26,12 +26,12 @@ func (s3a *S3ApiServer) AdjustedUrl(location *filer_pb.Location) string {
 	return location.Url
 }
 
-func writeSuccessResponseXML(w http.ResponseWriter, response interface{}) {
-	s3err.WriteXMLResponse(w, http.StatusOK, response)
+func writeSuccessResponseXML(w http.ResponseWriter, r *http.Request, response interface{}) {
+	s3err.WriteXMLResponse(w, r, http.StatusOK, response)
 }
 
-func writeSuccessResponseEmpty(w http.ResponseWriter) {
-	s3err.WriteEmptyResponse(w, http.StatusOK)
+func writeSuccessResponseEmpty(w http.ResponseWriter, r *http.Request) {
+	s3err.WriteEmptyResponse(w, r, http.StatusOK)
 }
 
 func validateContentMd5(h http.Header) ([]byte, error) {

weed/s3api/s3api_object_copy_handlers.go

@@ -34,16 +34,16 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
 		dir, name := fullPath.DirAndName()
 		entry, err := s3a.getEntry(dir, name)
 		if err != nil {
-			s3err.WriteErrorResponse(w, s3err.ErrInvalidCopySource, r)
+			s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
 			return
 		}
 		entry.Extended = weed_server.SaveAmzMetaData(r, entry.Extended, isReplace(r))
 		err = s3a.touch(dir, name, entry)
 		if err != nil {
-			s3err.WriteErrorResponse(w, s3err.ErrInvalidCopySource, r)
+			s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
 			return
 		}
-		writeSuccessResponseXML(w, CopyObjectResult{
+		writeSuccessResponseXML(w, r, CopyObjectResult{
 			ETag:         fmt.Sprintf("%x", entry.Attributes.Md5),
 			LastModified: time.Now().UTC(),
 		})
@@ -52,19 +52,19 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
 
 	// If source object is empty or bucket is empty, reply back invalid copy source.
 	if srcObject == "" || srcBucket == "" {
-		s3err.WriteErrorResponse(w, s3err.ErrInvalidCopySource, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
 		return
 	}
 	srcPath := util.FullPath(fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, srcBucket, srcObject))
 	dir, name := srcPath.DirAndName()
 	_, err = s3a.getEntry(dir, name)
 	if err != nil {
-		s3err.WriteErrorResponse(w, s3err.ErrInvalidCopySource, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
 		return
 	}
 
 	if srcBucket == dstBucket && srcObject == dstObject {
-		s3err.WriteErrorResponse(w, s3err.ErrInvalidCopyDest, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopyDest)
 		return
 	}
 
@@ -75,7 +75,7 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
 
 	_, _, resp, err := util.DownloadFile(srcUrl, "")
 	if err != nil {
-		s3err.WriteErrorResponse(w, s3err.ErrInvalidCopySource, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
 		return
 	}
 	defer util.CloseResponse(resp)
@@ -84,7 +84,7 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
 	etag, errCode := s3a.putToFiler(r, dstUrl, resp.Body)
 
 	if errCode != s3err.ErrNone {
-		s3err.WriteErrorResponse(w, errCode, r)
+		s3err.WriteErrorResponse(w, r, errCode)
 		return
 	}
 
@@ -95,7 +95,7 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
 		LastModified: time.Now().UTC(),
 	}
 
-	writeSuccessResponseXML(w, response)
+	writeSuccessResponseXML(w, r, response)
 
 }
 
@@ -128,7 +128,7 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req
 	srcBucket, srcObject := pathToBucketAndObject(cpSrcPath)
 	// If source object is empty or bucket is empty, reply back invalid copy source.
 	if srcObject == "" || srcBucket == "" {
-		s3err.WriteErrorResponse(w, s3err.ErrInvalidCopySource, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
 		return
 	}
 
@@ -137,7 +137,7 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req
 
 	partID, err := strconv.Atoi(partIDString)
 	if err != nil {
-		s3err.WriteErrorResponse(w, s3err.ErrInvalidPart, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInvalidPart)
 		return
 	}
 
@@ -145,7 +145,7 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req
 
 	// check partID with maximum part ID for multipart objects
 	if partID > globalMaxPartID {
-		s3err.WriteErrorResponse(w, s3err.ErrInvalidMaxParts, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInvalidMaxParts)
 		return
 	}
 
@@ -158,7 +158,7 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req
 
 	dataReader, err := util.ReadUrlAsReaderCloser(srcUrl, rangeHeader)
 	if err != nil {
-		s3err.WriteErrorResponse(w, s3err.ErrInvalidCopySource, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
 		return
 	}
 	defer dataReader.Close()
@@ -167,7 +167,7 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req
 	etag, errCode := s3a.putToFiler(r, dstUrl, dataReader)
 
 	if errCode != s3err.ErrNone {
-		s3err.WriteErrorResponse(w, errCode, r)
+		s3err.WriteErrorResponse(w, r, errCode)
 		return
 	}
 
@@ -178,7 +178,7 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req
 		LastModified: time.Now().UTC(),
 	}
 
-	writeSuccessResponseXML(w, response)
+	writeSuccessResponseXML(w, r, response)
 
 }
 

weed/s3api/s3api_object_handlers.go

@@ -39,9 +39,12 @@ func init() {
 
 func mimeDetect(r *http.Request, dataReader io.Reader) io.ReadCloser {
 	mimeBuffer := make([]byte, 512)
-	dataReader.Read(mimeBuffer)
-	r.Header.Set("Content-Type", http.DetectContentType(mimeBuffer))
-	return io.NopCloser(io.MultiReader(bytes.NewReader(mimeBuffer), dataReader))
+	size, _ := dataReader.Read(mimeBuffer)
+	if size > 0 {
+		r.Header.Set("Content-Type", http.DetectContentType(mimeBuffer[:size]))
+		return io.NopCloser(io.MultiReader(bytes.NewReader(mimeBuffer[:size]), dataReader))
+	}
+	return io.NopCloser(dataReader)
 }
 
 func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request) {
@@ -53,20 +56,20 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request)
 
 	_, err := validateContentMd5(r.Header)
 	if err != nil {
-		s3err.WriteErrorResponse(w, s3err.ErrInvalidDigest, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInvalidDigest)
 		return
 	}
 
 	if r.Header.Get("Cache-Control") != "" {
 		if _, err = cacheobject.ParseRequestCacheControl(r.Header.Get("Cache-Control")); err != nil {
-			s3err.WriteErrorResponse(w, s3err.ErrInvalidDigest, r)
+			s3err.WriteErrorResponse(w, r, s3err.ErrInvalidDigest)
 			return
 		}
 	}
 
 	if r.Header.Get("Expires") != "" {
 		if _, err = time.Parse(http.TimeFormat, r.Header.Get("Expires")); err != nil {
-			s3err.WriteErrorResponse(w, s3err.ErrInvalidDigest, r)
+			s3err.WriteErrorResponse(w, r, s3err.ErrInvalidDigest)
 			return
 		}
 	}
@@ -84,12 +87,12 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request)
 			_, s3ErrCode = s3a.iam.reqSignatureV4Verify(r)
 		}
 		if s3ErrCode != s3err.ErrNone {
-			s3err.WriteErrorResponse(w, s3ErrCode, r)
+			s3err.WriteErrorResponse(w, r, s3ErrCode)
 			return
 		}
 	} else {
 		if authTypeStreamingSigned == rAuthType {
-			s3err.WriteErrorResponse(w, s3err.ErrAuthNotSetup, r)
+			s3err.WriteErrorResponse(w, r, s3err.ErrAuthNotSetup)
 			return
 		}
 	}
@@ -97,7 +100,7 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request)
 
 	if strings.HasSuffix(object, "/") {
 		if err := s3a.mkdir(s3a.option.BucketsPath, bucket+object, nil); err != nil {
-			s3err.WriteErrorResponse(w, s3err.ErrInternalError, r)
+			s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 			return
 		}
 	} else {
@@ -110,14 +113,14 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request)
 		etag, errCode := s3a.putToFiler(r, uploadUrl, dataReader)
 
 		if errCode != s3err.ErrNone {
-			s3err.WriteErrorResponse(w, errCode, r)
+			s3err.WriteErrorResponse(w, r, errCode)
 			return
 		}
 
 		setEtag(w, etag)
 	}
 
-	writeSuccessResponseEmpty(w)
+	writeSuccessResponseEmpty(w, r)
 }
 
 func urlPathEscape(object string) string {
@@ -134,7 +137,7 @@ func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request)
 	glog.V(3).Infof("GetObjectHandler %s %s", bucket, object)
 
 	if strings.HasSuffix(r.URL.Path, "/") {
-		s3err.WriteErrorResponse(w, s3err.ErrNotImplemented, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrNotImplemented)
 		return
 	}
 
@@ -212,13 +215,13 @@ func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *h
 
 	deleteXMLBytes, err := io.ReadAll(r.Body)
 	if err != nil {
-		s3err.WriteErrorResponse(w, s3err.ErrInternalError, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 		return
 	}
 
 	deleteObjects := &DeleteObjectsRequest{}
 	if err := xml.Unmarshal(deleteXMLBytes, deleteObjects); err != nil {
-		s3err.WriteErrorResponse(w, s3err.ErrMalformedXML, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrMalformedXML)
 		return
 	}
 
@@ -270,7 +273,7 @@ func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *h
 	}
 	deleteResp.Errors = deleteErrors
 
-	writeSuccessResponseXML(w, deleteResp)
+	writeSuccessResponseXML(w, r, deleteResp)
 
 }
 
@@ -314,7 +317,7 @@ func (s3a *S3ApiServer) proxyToFiler(w http.ResponseWriter, r *http.Request, des
 
 	if err != nil {
 		glog.Errorf("NewRequest %s: %v", destUrl, err)
-		s3err.WriteErrorResponse(w, s3err.ErrInternalError, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 		return
 	}
 
@@ -343,19 +346,19 @@ func (s3a *S3ApiServer) proxyToFiler(w http.ResponseWriter, r *http.Request, des
 
 	if postErr != nil {
 		glog.Errorf("post to filer: %v", postErr)
-		s3err.WriteErrorResponse(w, s3err.ErrInternalError, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 		return
 	}
 	defer util.CloseResponse(resp)
 
 	if resp.StatusCode == http.StatusPreconditionFailed {
-		s3err.WriteErrorResponse(w, s3err.ErrPreconditionFailed, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrPreconditionFailed)
 		return
 	}
 
 	if (resp.ContentLength == -1 || resp.StatusCode == 404) && resp.StatusCode != 304 {
 		if r.Method != "DELETE" {
-			s3err.WriteErrorResponse(w, s3err.ErrNoSuchKey, r)
+			s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
 			return
 		}
 	}

weed/s3api/s3api_object_handlers_postpolicy.go

@@ -29,23 +29,23 @@ func (s3a *S3ApiServer) PostPolicyBucketHandler(w http.ResponseWriter, r *http.R
 
 	reader, err := r.MultipartReader()
 	if err != nil {
-		s3err.WriteErrorResponse(w, s3err.ErrMalformedPOSTRequest, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrMalformedPOSTRequest)
 		return
 	}
 	form, err := reader.ReadForm(int64(5 * humanize.MiByte))
 	if err != nil {
-		s3err.WriteErrorResponse(w, s3err.ErrMalformedPOSTRequest, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrMalformedPOSTRequest)
 		return
 	}
 	defer form.RemoveAll()
 
 	fileBody, fileName, fileSize, formValues, err := extractPostPolicyFormValues(form)
 	if err != nil {
-		s3err.WriteErrorResponse(w, s3err.ErrMalformedPOSTRequest, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrMalformedPOSTRequest)
 		return
 	}
 	if fileBody == nil {
-		s3err.WriteErrorResponse(w, s3err.ErrPOSTFileRequired, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrPOSTFileRequired)
 		return
 	}
 	defer fileBody.Close()
@@ -63,7 +63,7 @@ func (s3a *S3ApiServer) PostPolicyBucketHandler(w http.ResponseWriter, r *http.R
 	if successRedirect != "" {
 		redirectURL, err = url.Parse(successRedirect)
 		if err != nil {
-			s3err.WriteErrorResponse(w, s3err.ErrMalformedPOSTRequest, r)
+			s3err.WriteErrorResponse(w, r, s3err.ErrMalformedPOSTRequest)
 			return
 		}
 	}
@@ -71,13 +71,13 @@ func (s3a *S3ApiServer) PostPolicyBucketHandler(w http.ResponseWriter, r *http.R
 	// Verify policy signature.
 	errCode := s3a.iam.doesPolicySignatureMatch(formValues)
 	if errCode != s3err.ErrNone {
-		s3err.WriteErrorResponse(w, errCode, r)
+		s3err.WriteErrorResponse(w, r, errCode)
 		return
 	}
 
 	policyBytes, err := base64.StdEncoding.DecodeString(formValues.Get("Policy"))
 	if err != nil {
-		s3err.WriteErrorResponse(w, s3err.ErrMalformedPOSTRequest, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrMalformedPOSTRequest)
 		return
 	}
 
@@ -86,7 +86,7 @@ func (s3a *S3ApiServer) PostPolicyBucketHandler(w http.ResponseWriter, r *http.R
 
 		postPolicyForm, err := policy.ParsePostPolicyForm(string(policyBytes))
 		if err != nil {
-			s3err.WriteErrorResponse(w, s3err.ErrPostPolicyConditionInvalidFormat, r)
+			s3err.WriteErrorResponse(w, r, s3err.ErrPostPolicyConditionInvalidFormat)
 			return
 		}
 
@@ -102,12 +102,12 @@ func (s3a *S3ApiServer) PostPolicyBucketHandler(w http.ResponseWriter, r *http.R
 		lengthRange := postPolicyForm.Conditions.ContentLengthRange
 		if lengthRange.Valid {
 			if fileSize < lengthRange.Min {
-				s3err.WriteErrorResponse(w, s3err.ErrEntityTooSmall, r)
+				s3err.WriteErrorResponse(w, r, s3err.ErrEntityTooSmall)
 				return
 			}
 
 			if fileSize > lengthRange.Max {
-				s3err.WriteErrorResponse(w, s3err.ErrEntityTooLarge, r)
+				s3err.WriteErrorResponse(w, r, s3err.ErrEntityTooLarge)
 				return
 			}
 		}
@@ -118,7 +118,7 @@ func (s3a *S3ApiServer) PostPolicyBucketHandler(w http.ResponseWriter, r *http.R
 	etag, errCode := s3a.putToFiler(r, uploadUrl, fileBody)
 
 	if errCode != s3err.ErrNone {
-		s3err.WriteErrorResponse(w, errCode, r)
+		s3err.WriteErrorResponse(w, r, errCode)
 		return
 	}
 
@@ -126,7 +126,7 @@ func (s3a *S3ApiServer) PostPolicyBucketHandler(w http.ResponseWriter, r *http.R
 		// Replace raw query params..
 		redirectURL.RawQuery = getRedirectPostRawQuery(bucket, object, etag)
 		w.Header().Set("Location", redirectURL.String())
-		s3err.WriteEmptyResponse(w, http.StatusSeeOther)
+		s3err.WriteEmptyResponse(w, r, http.StatusSeeOther)
 		return
 	}
 
@@ -141,11 +141,11 @@ func (s3a *S3ApiServer) PostPolicyBucketHandler(w http.ResponseWriter, r *http.R
 			ETag:     `"` + etag + `"`,
 			Location: w.Header().Get("Location"),
 		}
-		s3err.WriteXMLResponse(w, http.StatusCreated, resp)
+		s3err.WriteXMLResponse(w, r, http.StatusCreated, resp)
 	case "200":
-		s3err.WriteEmptyResponse(w, http.StatusOK)
+		s3err.WriteEmptyResponse(w, r, http.StatusOK)
 	default:
-		writeSuccessResponseEmpty(w)
+		writeSuccessResponseEmpty(w, r)
 	}
 
 }

weed/s3api/s3api_object_multipart_handlers.go

@@ -45,11 +45,11 @@ func (s3a *S3ApiServer) NewMultipartUploadHandler(w http.ResponseWriter, r *http
 	glog.V(2).Info("NewMultipartUploadHandler", string(s3err.EncodeXMLResponse(response)), errCode)
 
 	if errCode != s3err.ErrNone {
-		s3err.WriteErrorResponse(w, errCode, r)
+		s3err.WriteErrorResponse(w, r, errCode)
 		return
 	}
 
-	writeSuccessResponseXML(w, response)
+	writeSuccessResponseXML(w, r, response)
 
 }
 
@@ -69,11 +69,11 @@ func (s3a *S3ApiServer) CompleteMultipartUploadHandler(w http.ResponseWriter, r
 	glog.V(2).Info("CompleteMultipartUploadHandler", string(s3err.EncodeXMLResponse(response)), errCode)
 
 	if errCode != s3err.ErrNone {
-		s3err.WriteErrorResponse(w, errCode, r)
+		s3err.WriteErrorResponse(w, r, errCode)
 		return
 	}
 
-	writeSuccessResponseXML(w, response)
+	writeSuccessResponseXML(w, r, response)
 
 }
 
@@ -91,13 +91,13 @@ func (s3a *S3ApiServer) AbortMultipartUploadHandler(w http.ResponseWriter, r *ht
 	})
 
 	if errCode != s3err.ErrNone {
-		s3err.WriteErrorResponse(w, errCode, r)
+		s3err.WriteErrorResponse(w, r, errCode)
 		return
 	}
 
 	glog.V(2).Info("AbortMultipartUploadHandler", string(s3err.EncodeXMLResponse(response)))
 
-	writeSuccessResponseXML(w, response)
+	writeSuccessResponseXML(w, r, response)
 
 }
 
@@ -107,13 +107,13 @@ func (s3a *S3ApiServer) ListMultipartUploadsHandler(w http.ResponseWriter, r *ht
 
 	prefix, keyMarker, uploadIDMarker, delimiter, maxUploads, encodingType := getBucketMultipartResources(r.URL.Query())
 	if maxUploads < 0 {
-		s3err.WriteErrorResponse(w, s3err.ErrInvalidMaxUploads, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInvalidMaxUploads)
 		return
 	}
 	if keyMarker != "" {
 		// Marker not common with prefix is not implemented.
 		if !strings.HasPrefix(keyMarker, prefix) {
-			s3err.WriteErrorResponse(w, s3err.ErrNotImplemented, r)
+			s3err.WriteErrorResponse(w, r, s3err.ErrNotImplemented)
 			return
 		}
 	}
@@ -128,16 +128,16 @@ func (s3a *S3ApiServer) ListMultipartUploadsHandler(w http.ResponseWriter, r *ht
 		UploadIdMarker: aws.String(uploadIDMarker),
 	})
 
-	glog.V(2).Info("ListMultipartUploadsHandler", string(s3err.EncodeXMLResponse(response)), errCode)
+	glog.V(2).Infof("ListMultipartUploadsHandler %s errCode=%d", string(s3err.EncodeXMLResponse(response)), errCode)
 
 	if errCode != s3err.ErrNone {
-		s3err.WriteErrorResponse(w, errCode, r)
+		s3err.WriteErrorResponse(w, r, errCode)
 		return
 	}
 
 	// TODO handle encodingType
 
-	writeSuccessResponseXML(w, response)
+	writeSuccessResponseXML(w, r, response)
 }
 
 // ListObjectPartsHandler - Lists object parts in a multipart upload.
@@ -146,11 +146,11 @@ func (s3a *S3ApiServer) ListObjectPartsHandler(w http.ResponseWriter, r *http.Re
 
 	uploadID, partNumberMarker, maxParts, _ := getObjectResources(r.URL.Query())
 	if partNumberMarker < 0 {
-		s3err.WriteErrorResponse(w, s3err.ErrInvalidPartNumberMarker, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInvalidPartNumberMarker)
 		return
 	}
 	if maxParts < 0 {
-		s3err.WriteErrorResponse(w, s3err.ErrInvalidMaxParts, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInvalidMaxParts)
 		return
 	}
 
@@ -162,14 +162,14 @@ func (s3a *S3ApiServer) ListObjectPartsHandler(w http.ResponseWriter, r *http.Re
 		UploadId:         aws.String(uploadID),
 	})
 
-	glog.V(2).Info("ListObjectPartsHandler", string(s3err.EncodeXMLResponse(response)), errCode)
+	glog.V(2).Infof("ListObjectPartsHandler %s count=%d", string(s3err.EncodeXMLResponse(response)), len(response.Part))
 
 	if errCode != s3err.ErrNone {
-		s3err.WriteErrorResponse(w, errCode, r)
+		s3err.WriteErrorResponse(w, r, errCode)
 		return
 	}
 
-	writeSuccessResponseXML(w, response)
+	writeSuccessResponseXML(w, r, response)
 
 }
 
@@ -180,18 +180,18 @@ func (s3a *S3ApiServer) PutObjectPartHandler(w http.ResponseWriter, r *http.Requ
 	uploadID := r.URL.Query().Get("uploadId")
 	exists, err := s3a.exists(s3a.genUploadsFolder(bucket), uploadID, true)
 	if !exists {
-		s3err.WriteErrorResponse(w, s3err.ErrNoSuchUpload, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchUpload)
 		return
 	}
 
 	partIDString := r.URL.Query().Get("partNumber")
 	partID, err := strconv.Atoi(partIDString)
 	if err != nil {
-		s3err.WriteErrorResponse(w, s3err.ErrInvalidPart, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInvalidPart)
 		return
 	}
 	if partID > globalMaxPartID {
-		s3err.WriteErrorResponse(w, s3err.ErrInvalidMaxParts, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInvalidMaxParts)
 		return
 	}
 
@@ -208,12 +208,14 @@ func (s3a *S3ApiServer) PutObjectPartHandler(w http.ResponseWriter, r *http.Requ
 			_, s3ErrCode = s3a.iam.reqSignatureV4Verify(r)
 		}
 		if s3ErrCode != s3err.ErrNone {
-			s3err.WriteErrorResponse(w, s3ErrCode, r)
+			s3err.WriteErrorResponse(w, r, s3ErrCode)
 			return
 		}
 	}
 	defer dataReader.Close()
 
+	glog.V(2).Infof("PutObjectPartHandler %s %s %04d", bucket, uploadID, partID)
+
 	uploadUrl := fmt.Sprintf("http://%s%s/%s/%04d.part?collection=%s",
 		s3a.option.Filer.ToHttpAddress(), s3a.genUploadsFolder(bucket), uploadID, partID, bucket)
 
@@ -223,13 +225,13 @@ func (s3a *S3ApiServer) PutObjectPartHandler(w http.ResponseWriter, r *http.Requ
 
 	etag, errCode := s3a.putToFiler(r, uploadUrl, dataReader)
 	if errCode != s3err.ErrNone {
-		s3err.WriteErrorResponse(w, errCode, r)
+		s3err.WriteErrorResponse(w, r, errCode)
 		return
 	}
 
 	setEtag(w, etag)
 
-	writeSuccessResponseEmpty(w)
+	writeSuccessResponseEmpty(w, r)
 
 }
 

weed/s3api/s3api_object_tagging_handlers.go

@@ -26,15 +26,15 @@ func (s3a *S3ApiServer) GetObjectTaggingHandler(w http.ResponseWriter, r *http.R
 	if err != nil {
 		if err == filer_pb.ErrNotFound {
 			glog.Errorf("GetObjectTaggingHandler %s: %v", r.URL, err)
-			s3err.WriteErrorResponse(w, s3err.ErrNoSuchKey, r)
+			s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
 		} else {
 			glog.Errorf("GetObjectTaggingHandler %s: %v", r.URL, err)
-			s3err.WriteErrorResponse(w, s3err.ErrInternalError, r)
+			s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 		}
 		return
 	}
 
-	writeSuccessResponseXML(w, FromTags(tags))
+	writeSuccessResponseXML(w, r, FromTags(tags))
 
 }
 
@@ -52,29 +52,29 @@ func (s3a *S3ApiServer) PutObjectTaggingHandler(w http.ResponseWriter, r *http.R
 	input, err := io.ReadAll(io.LimitReader(r.Body, r.ContentLength))
 	if err != nil {
 		glog.Errorf("PutObjectTaggingHandler read input %s: %v", r.URL, err)
-		s3err.WriteErrorResponse(w, s3err.ErrInternalError, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 		return
 	}
 	if err = xml.Unmarshal(input, tagging); err != nil {
 		glog.Errorf("PutObjectTaggingHandler Unmarshal %s: %v", r.URL, err)
-		s3err.WriteErrorResponse(w, s3err.ErrMalformedXML, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrMalformedXML)
 		return
 	}
 	tags := tagging.ToTags()
 	if len(tags) > 10 {
 		glog.Errorf("PutObjectTaggingHandler tags %s: %d tags more than 10", r.URL, len(tags))
-		s3err.WriteErrorResponse(w, s3err.ErrInvalidTag, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInvalidTag)
 		return
 	}
 	for k, v := range tags {
 		if len(k) > 128 {
 			glog.Errorf("PutObjectTaggingHandler tags %s: tag key %s longer than 128", r.URL, k)
-			s3err.WriteErrorResponse(w, s3err.ErrInvalidTag, r)
+			s3err.WriteErrorResponse(w, r, s3err.ErrInvalidTag)
 			return
 		}
 		if len(v) > 256 {
 			glog.Errorf("PutObjectTaggingHandler tags %s: tag value %s longer than 256", r.URL, v)
-			s3err.WriteErrorResponse(w, s3err.ErrInvalidTag, r)
+			s3err.WriteErrorResponse(w, r, s3err.ErrInvalidTag)
 			return
 		}
 	}
@@ -82,15 +82,15 @@ func (s3a *S3ApiServer) PutObjectTaggingHandler(w http.ResponseWriter, r *http.R
 	if err = s3a.setTags(dir, name, tagging.ToTags()); err != nil {
 		if err == filer_pb.ErrNotFound {
 			glog.Errorf("PutObjectTaggingHandler setTags %s: %v", r.URL, err)
-			s3err.WriteErrorResponse(w, s3err.ErrNoSuchKey, r)
+			s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
 		} else {
 			glog.Errorf("PutObjectTaggingHandler setTags %s: %v", r.URL, err)
-			s3err.WriteErrorResponse(w, s3err.ErrInternalError, r)
+			s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 		}
 		return
 	}
 
-	w.WriteHeader(http.StatusNoContent)
+	w.WriteHeader(http.StatusOK)
 
 }
 
@@ -108,10 +108,10 @@ func (s3a *S3ApiServer) DeleteObjectTaggingHandler(w http.ResponseWriter, r *htt
 	if err != nil {
 		if err == filer_pb.ErrNotFound {
 			glog.Errorf("DeleteObjectTaggingHandler %s: %v", r.URL, err)
-			s3err.WriteErrorResponse(w, s3err.ErrNoSuchKey, r)
+			s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
 		} else {
 			glog.Errorf("DeleteObjectTaggingHandler %s: %v", r.URL, err)
-			s3err.WriteErrorResponse(w, s3err.ErrInternalError, r)
+			s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 		}
 		return
 	}

weed/s3api/s3api_objects_list_handlers.go

@@ -45,11 +45,11 @@ func (s3a *S3ApiServer) ListObjectsV2Handler(w http.ResponseWriter, r *http.Requ
 	originalPrefix, continuationToken, startAfter, delimiter, _, maxKeys := getListObjectsV2Args(r.URL.Query())
 
 	if maxKeys < 0 {
-		s3err.WriteErrorResponse(w, s3err.ErrInvalidMaxKeys, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInvalidMaxKeys)
 		return
 	}
 	if delimiter != "" && delimiter != "/" {
-		s3err.WriteErrorResponse(w, s3err.ErrNotImplemented, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrNotImplemented)
 		return
 	}
 
@@ -61,13 +61,13 @@ func (s3a *S3ApiServer) ListObjectsV2Handler(w http.ResponseWriter, r *http.Requ
 	response, err := s3a.listFilerEntries(bucket, originalPrefix, maxKeys, marker, delimiter)
 
 	if err != nil {
-		s3err.WriteErrorResponse(w, s3err.ErrInternalError, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 		return
 	}
 
 	if len(response.Contents) == 0 {
 		if exists, existErr := s3a.exists(s3a.option.BucketsPath, bucket, true); existErr == nil && !exists {
-			s3err.WriteErrorResponse(w, s3err.ErrNoSuchBucket, r)
+			s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchBucket)
 			return
 		}
 	}
@@ -87,7 +87,7 @@ func (s3a *S3ApiServer) ListObjectsV2Handler(w http.ResponseWriter, r *http.Requ
 		StartAfter:            startAfter,
 	}
 
-	writeSuccessResponseXML(w, responseV2)
+	writeSuccessResponseXML(w, r, responseV2)
 }
 
 func (s3a *S3ApiServer) ListObjectsV1Handler(w http.ResponseWriter, r *http.Request) {
@@ -101,29 +101,29 @@ func (s3a *S3ApiServer) ListObjectsV1Handler(w http.ResponseWriter, r *http.Requ
 	originalPrefix, marker, delimiter, maxKeys := getListObjectsV1Args(r.URL.Query())
 
 	if maxKeys < 0 {
-		s3err.WriteErrorResponse(w, s3err.ErrInvalidMaxKeys, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInvalidMaxKeys)
 		return
 	}
 	if delimiter != "" && delimiter != "/" {
-		s3err.WriteErrorResponse(w, s3err.ErrNotImplemented, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrNotImplemented)
 		return
 	}
 
 	response, err := s3a.listFilerEntries(bucket, originalPrefix, maxKeys, marker, delimiter)
 
 	if err != nil {
-		s3err.WriteErrorResponse(w, s3err.ErrInternalError, r)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 		return
 	}
 
 	if len(response.Contents) == 0 {
 		if exists, existErr := s3a.exists(s3a.option.BucketsPath, bucket, true); existErr == nil && !exists {
-			s3err.WriteErrorResponse(w, s3err.ErrNoSuchBucket, r)
+			s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchBucket)
 			return
 		}
 	}
 
-	writeSuccessResponseXML(w, response)
+	writeSuccessResponseXML(w, r, response)
 }
 
 func (s3a *S3ApiServer) listFilerEntries(bucket string, originalPrefix string, maxKeys int, marker string, delimiter string) (response ListBucketResult, err error) {

weed/s3api/s3api_server.go

@@ -103,13 +103,9 @@ func (s3a *S3ApiServer) registerRouter(router *mux.Router) {
 		bucket.Methods("PUT").Path("/{object:.+}").HeadersRegexp("X-Amz-Copy-Source", ".*?(\\/|%2F).*?").HandlerFunc(track(s3a.iam.Auth(s3a.CopyObjectHandler, ACTION_WRITE), "COPY"))
 		// PutObject
 		bucket.Methods("PUT").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.PutObjectHandler, ACTION_WRITE), "PUT"))
-		// PutBucket
-		bucket.Methods("PUT").HandlerFunc(track(s3a.iam.Auth(s3a.PutBucketHandler, ACTION_ADMIN), "PUT"))
 
 		// DeleteObject
 		bucket.Methods("DELETE").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.DeleteObjectHandler, ACTION_WRITE), "DELETE"))
-		// DeleteBucket
-		bucket.Methods("DELETE").HandlerFunc(track(s3a.iam.Auth(s3a.DeleteBucketHandler, ACTION_WRITE), "DELETE"))
 
 		// ListObjectsV2
 		bucket.Methods("GET").HandlerFunc(track(s3a.iam.Auth(s3a.ListObjectsV2Handler, ACTION_LIST), "LIST")).Queries("list-type", "2")
@@ -139,6 +135,13 @@ func (s3a *S3ApiServer) registerRouter(router *mux.Router) {
 
 		// ListObjectsV1 (Legacy)
 		bucket.Methods("GET").HandlerFunc(track(s3a.iam.Auth(s3a.ListObjectsV1Handler, ACTION_LIST), "LIST"))
+
+		// PutBucket
+		bucket.Methods("PUT").HandlerFunc(track(s3a.iam.Auth(s3a.PutBucketHandler, ACTION_ADMIN), "PUT"))
+
+		// DeleteBucket
+		bucket.Methods("DELETE").HandlerFunc(track(s3a.iam.Auth(s3a.DeleteBucketHandler, ACTION_WRITE), "DELETE"))
+
 		/*
 
 			// not implemented

weed/s3api/s3api_status_handlers.go

@@ -4,5 +4,5 @@ import "net/http"
 
 func (s3a *S3ApiServer) StatusHandler(w http.ResponseWriter, r *http.Request) {
 	// write out the response code and content type header
-	writeSuccessResponseEmpty(w)
+	writeSuccessResponseEmpty(w, r)
 }

weed/s3api/s3err/error_handler.go

@@ -19,15 +19,15 @@ const (
 	MimeXML  mimeType = "application/xml"
 )
 
-func WriteXMLResponse(w http.ResponseWriter, statusCode int, response interface{}) {
-	WriteResponse(w, statusCode, EncodeXMLResponse(response), MimeXML)
+func WriteXMLResponse(w http.ResponseWriter, r *http.Request, statusCode int, response interface{}) {
+	WriteResponse(w, r, statusCode, EncodeXMLResponse(response), MimeXML)
 }
 
-func WriteEmptyResponse(w http.ResponseWriter, statusCode int) {
-	WriteResponse(w, statusCode, []byte{}, mimeNone)
+func WriteEmptyResponse(w http.ResponseWriter, r *http.Request, statusCode int) {
+	WriteResponse(w, r, statusCode, []byte{}, mimeNone)
 }
 
-func WriteErrorResponse(w http.ResponseWriter, errorCode ErrorCode, r *http.Request) {
+func WriteErrorResponse(w http.ResponseWriter, r *http.Request, errorCode ErrorCode) {
 	vars := mux.Vars(r)
 	bucket := vars["bucket"]
 	object := vars["object"]
@@ -38,7 +38,7 @@ func WriteErrorResponse(w http.ResponseWriter, errorCode ErrorCode, r *http.Requ
 	apiError := GetAPIError(errorCode)
 	errorResponse := getRESTErrorResponse(apiError, r.URL.Path, bucket, object)
 	encodedErrorResponse := EncodeXMLResponse(errorResponse)
-	WriteResponse(w, apiError.HTTPStatusCode, encodedErrorResponse, MimeXML)
+	WriteResponse(w, r, apiError.HTTPStatusCode, encodedErrorResponse, MimeXML)
 }
 
 func getRESTErrorResponse(err APIError, resource string, bucket, object string) RESTErrorResponse {
@@ -61,13 +61,17 @@ func EncodeXMLResponse(response interface{}) []byte {
 	return bytesBuffer.Bytes()
 }
 
-func setCommonHeaders(w http.ResponseWriter) {
+func setCommonHeaders(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("x-amz-request-id", fmt.Sprintf("%d", time.Now().UnixNano()))
 	w.Header().Set("Accept-Ranges", "bytes")
+	if r.Header.Get("Origin") != "" {
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+		w.Header().Set("Access-Control-Allow-Credentials", "true")
+	}
 }
 
-func WriteResponse(w http.ResponseWriter, statusCode int, response []byte, mType mimeType) {
-	setCommonHeaders(w)
+func WriteResponse(w http.ResponseWriter, r *http.Request, statusCode int, response []byte, mType mimeType) {
+	setCommonHeaders(w, r)
 	if response != nil {
 		w.Header().Set("Content-Length", strconv.Itoa(len(response)))
 	}
@@ -88,5 +92,5 @@ func WriteResponse(w http.ResponseWriter, statusCode int, response []byte, mType
 // If none of the http routes match respond with MethodNotAllowed
 func NotFoundHandler(w http.ResponseWriter, r *http.Request) {
 	glog.V(0).Infof("unsupported %s %s", r.Method, r.RequestURI)
-	WriteErrorResponse(w, ErrMethodNotAllowed, r)
+	WriteErrorResponse(w, r, ErrMethodNotAllowed)
 }

weed/s3api/s3err/s3api_errors.go

@@ -202,7 +202,7 @@ var errorCodeResponse = map[ErrorCode]APIError{
 		HTTPStatusCode: http.StatusBadRequest,
 	},
 	ErrInvalidTag: {
-		Code:           "InvalidArgument",
+		Code:           "InvalidTag",
 		Description:    "The Tag value you have provided is invalid",
 		HTTPStatusCode: http.StatusBadRequest,
 	},

weed/s3api/tags.go

@@ -14,8 +14,9 @@ type TagSet struct {
 }
 
 type Tagging struct {
-	XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ Tagging"`
+	XMLName xml.Name `xml:"Tagging"`
 	TagSet  TagSet   `xml:"TagSet"`
+	Xmlns   string   `xml:"xmlns,attr"`
 }
 
 func (t *Tagging) ToTags() map[string]string {
@@ -27,7 +28,7 @@ func (t *Tagging) ToTags() map[string]string {
 }
 
 func FromTags(tags map[string]string) (t *Tagging) {
-	t = &Tagging{}
+	t = &Tagging{Xmlns: "http://s3.amazonaws.com/doc/2006-03-01/"}
 	for k, v := range tags {
 		t.TagSet.Tag = append(t.TagSet.Tag, Tag{
 			Key:   k,

weed/s3api/tags_test.go

@@ -32,6 +32,7 @@ func TestXMLUnmarshall(t *testing.T) {
 
 func TestXMLMarshall(t *testing.T) {
 	tags := &Tagging{
+		Xmlns: "http://s3.amazonaws.com/doc/2006-03-01/",
 		TagSet: TagSet{
 			[]Tag{
 				{

weed/server/filer_grpc_server_remote.go

@@ -17,7 +17,7 @@ import (
 	"time"
 )
 
-func (fs *FilerServer) DownloadToLocal(ctx context.Context, req *filer_pb.DownloadToLocalRequest) (*filer_pb.DownloadToLocalResponse, error) {
+func (fs *FilerServer) CacheRemoteObjectToLocalCluster(ctx context.Context, req *filer_pb.CacheRemoteObjectToLocalClusterRequest) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error) {
 
 	// load all mappings
 	mappingEntry, err := fs.filer.FindEntry(ctx, util.JoinPath(filer.DirectoryEtcRemote, filer.REMOTE_STORAGE_MOUNT_FILE))
@@ -57,7 +57,7 @@ func (fs *FilerServer) DownloadToLocal(ctx context.Context, req *filer_pb.Downlo
 		return nil, err
 	}
 
-	resp := &filer_pb.DownloadToLocalResponse{}
+	resp := &filer_pb.CacheRemoteObjectToLocalClusterResponse{}
 	if entry.Remote == nil || entry.Remote.RemoteSize == 0 {
 		return resp, nil
 	}

weed/server/filer_server_handlers_read.go

@@ -113,17 +113,15 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
 	w.Header().Set("Access-Control-Expose-Headers", strings.Join(seaweedHeaders, ","))
 
 	//set tag count
-	if r.Method == "GET" {
-		tagCount := 0
-		for k := range entry.Extended {
-			if strings.HasPrefix(k, xhttp.AmzObjectTagging+"-") {
-				tagCount++
-			}
-		}
-		if tagCount > 0 {
-			w.Header().Set(xhttp.AmzTagCount, strconv.Itoa(tagCount))
+	tagCount := 0
+	for k := range entry.Extended {
+		if strings.HasPrefix(k, xhttp.AmzObjectTagging+"-") {
+			tagCount++
 		}
 	}
+	if tagCount > 0 {
+		w.Header().Set(xhttp.AmzTagCount, strconv.Itoa(tagCount))
+	}
 
 	if inm := r.Header.Get("If-None-Match"); inm == "\""+etag+"\"" {
 		w.WriteHeader(http.StatusNotModified)
@@ -171,11 +169,11 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
 		chunks := entry.Chunks
 		if entry.IsInRemoteOnly() {
 			dir, name := entry.FullPath.DirAndName()
-			if resp, err := fs.DownloadToLocal(context.Background(), &filer_pb.DownloadToLocalRequest{
+			if resp, err := fs.CacheRemoteObjectToLocalCluster(context.Background(), &filer_pb.CacheRemoteObjectToLocalClusterRequest{
 				Directory: dir,
 				Name:      name,
 			}); err != nil {
-				glog.Errorf("DownloadToLocal %s: %v", entry.FullPath, err)
+				glog.Errorf("CacheRemoteObjectToLocalCluster %s: %v", entry.FullPath, err)
 				return fmt.Errorf("cache %s: %v", entry.FullPath, err)
 			} else {
 				chunks = resp.Entry.Chunks

weed/server/filer_server_handlers_write_autochunk.go

@@ -327,6 +327,8 @@ func SaveAmzMetaData(r *http.Request, existing map[string][]byte, isReplace bool
 			tag := strings.Split(v, "=")
 			if len(tag) == 2 {
 				metadata[xhttp.AmzObjectTagging+"-"+tag[0]] = []byte(tag[1])
+			} else if len(tag) == 1 {
+				metadata[xhttp.AmzObjectTagging+"-"+tag[0]] = nil
 			}
 		}
 	}

weed/server/volume_grpc_copy.go

@@ -22,7 +22,7 @@ import (
 const BufferSizeLimit = 1024 * 1024 * 2
 
 // VolumeCopy copy the .idx .dat .vif files, and mount the volume
-func (vs *VolumeServer) VolumeCopy(ctx context.Context, req *volume_server_pb.VolumeCopyRequest) (*volume_server_pb.VolumeCopyResponse, error) {
+func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stream volume_server_pb.VolumeServer_VolumeCopyServer) error {
 
 	v := vs.store.GetVolume(needle.VolumeId(req.VolumeId))
 	if v != nil {
@@ -31,7 +31,7 @@ func (vs *VolumeServer) VolumeCopy(ctx context.Context, req *volume_server_pb.Vo
 
 		err := vs.store.DeleteVolume(needle.VolumeId(req.VolumeId))
 		if err != nil {
-			return nil, fmt.Errorf("failed to delete existing volume %d: %v", req.VolumeId, err)
+			return fmt.Errorf("failed to delete existing volume %d: %v", req.VolumeId, err)
 		}
 
 		glog.V(0).Infof("deleted existing volume %d before copying.", req.VolumeId)
@@ -79,22 +79,38 @@ func (vs *VolumeServer) VolumeCopy(ctx context.Context, req *volume_server_pb.Vo
 		}()
 
 		// println("source:", volFileInfoResp.String())
+		copyResponse := &volume_server_pb.VolumeCopyResponse{}
+		reportInterval := int64(1024*1024*128)
+		nextReportTarget := reportInterval
 		var modifiedTsNs int64
-		if modifiedTsNs, err = vs.doCopyFile(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.DatFileSize, dataBaseFileName, ".dat", false, true); err != nil {
+		var sendErr error
+		if modifiedTsNs, err = vs.doCopyFile(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.DatFileSize, dataBaseFileName, ".dat", false, true, func(processed int64) bool {
+			if processed > nextReportTarget {
+				copyResponse.ProcessedBytes = processed
+				if sendErr = stream.Send(copyResponse); sendErr != nil {
+					return false
+				}
+				nextReportTarget = processed + reportInterval
+			}
+			return true
+		}); err != nil {
 			return err
 		}
+		if sendErr != nil {
+			return sendErr
+		}
 		if modifiedTsNs > 0 {
 			os.Chtimes(dataBaseFileName+".dat", time.Unix(0, modifiedTsNs), time.Unix(0, modifiedTsNs))
 		}
 
-		if modifiedTsNs, err = vs.doCopyFile(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.IdxFileSize, indexBaseFileName, ".idx", false, false); err != nil {
+		if modifiedTsNs, err = vs.doCopyFile(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.IdxFileSize, indexBaseFileName, ".idx", false, false, nil); err != nil {
 			return err
 		}
 		if modifiedTsNs > 0 {
 			os.Chtimes(indexBaseFileName+".idx", time.Unix(0, modifiedTsNs), time.Unix(0, modifiedTsNs))
 		}
 
-		if modifiedTsNs, err = vs.doCopyFile(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.DatFileSize, dataBaseFileName, ".vif", false, true); err != nil {
+		if modifiedTsNs, err = vs.doCopyFile(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.DatFileSize, dataBaseFileName, ".vif", false, true, nil); err != nil {
 			return err
 		}
 		if modifiedTsNs > 0 {
@@ -107,10 +123,10 @@ func (vs *VolumeServer) VolumeCopy(ctx context.Context, req *volume_server_pb.Vo
 	})
 
 	if err != nil {
-		return nil, err
+		return err
 	}
 	if dataBaseFileName == "" {
-		return nil, fmt.Errorf("not found volume %d file", req.VolumeId)
+		return fmt.Errorf("not found volume %d file", req.VolumeId)
 	}
 
 	idxFileName = indexBaseFileName + ".idx"
@@ -125,21 +141,25 @@ func (vs *VolumeServer) VolumeCopy(ctx context.Context, req *volume_server_pb.Vo
 	}()
 
 	if err = checkCopyFiles(volFileInfoResp, idxFileName, datFileName); err != nil { // added by panyc16
-		return nil, err
+		return err
 	}
 
 	// mount the volume
 	err = vs.store.MountVolume(needle.VolumeId(req.VolumeId))
 	if err != nil {
-		return nil, fmt.Errorf("failed to mount volume %d: %v", req.VolumeId, err)
+		return fmt.Errorf("failed to mount volume %d: %v", req.VolumeId, err)
 	}
 
-	return &volume_server_pb.VolumeCopyResponse{
+	if err = stream.Send(&volume_server_pb.VolumeCopyResponse{
 		LastAppendAtNs: volFileInfoResp.DatFileTimestampSeconds * uint64(time.Second),
-	}, err
+	}); err != nil {
+		glog.Errorf("send response: %v", err)
+	}
+
+	return err
 }
 
-func (vs *VolumeServer) doCopyFile(client volume_server_pb.VolumeServerClient, isEcVolume bool, collection string, vid, compactRevision uint32, stopOffset uint64, baseFileName, ext string, isAppend, ignoreSourceFileNotFound bool) (modifiedTsNs int64, err error) {
+func (vs *VolumeServer) doCopyFile(client volume_server_pb.VolumeServerClient, isEcVolume bool, collection string, vid, compactRevision uint32, stopOffset uint64, baseFileName, ext string, isAppend, ignoreSourceFileNotFound bool, progressFn storage.ProgressFunc) (modifiedTsNs int64, err error) {
 
 	copyFileClient, err := client.CopyFile(context.Background(), &volume_server_pb.CopyFileRequest{
 		VolumeId:                 vid,
@@ -154,7 +174,7 @@ func (vs *VolumeServer) doCopyFile(client volume_server_pb.VolumeServerClient, i
 		return modifiedTsNs, fmt.Errorf("failed to start copying volume %d %s file: %v", vid, ext, err)
 	}
 
-	modifiedTsNs, err = writeToFile(copyFileClient, baseFileName+ext, util.NewWriteThrottler(vs.compactionBytePerSecond), isAppend)
+	modifiedTsNs, err = writeToFile(copyFileClient, baseFileName+ext, util.NewWriteThrottler(vs.compactionBytePerSecond), isAppend, progressFn)
 	if err != nil {
 		return modifiedTsNs, fmt.Errorf("failed to copy %s file: %v", baseFileName+ext, err)
 	}
@@ -188,7 +208,7 @@ func checkCopyFiles(originFileInf *volume_server_pb.ReadVolumeFileStatusResponse
 	return nil
 }
 
-func writeToFile(client volume_server_pb.VolumeServer_CopyFileClient, fileName string, wt *util.WriteThrottler, isAppend bool) (modifiedTsNs int64, err error) {
+func writeToFile(client volume_server_pb.VolumeServer_CopyFileClient, fileName string, wt *util.WriteThrottler, isAppend bool, progressFn storage.ProgressFunc) (modifiedTsNs int64, err error) {
 	glog.V(4).Infof("writing to %s", fileName)
 	flags := os.O_WRONLY | os.O_CREATE | os.O_TRUNC
 	if isAppend {
@@ -200,6 +220,7 @@ func writeToFile(client volume_server_pb.VolumeServer_CopyFileClient, fileName s
 	}
 	defer dst.Close()
 
+	var progressedBytes int64
 	for {
 		resp, receiveErr := client.Recv()
 		if receiveErr == io.EOF {
@@ -212,6 +233,12 @@ func writeToFile(client volume_server_pb.VolumeServer_CopyFileClient, fileName s
 			return modifiedTsNs, fmt.Errorf("receiving %s: %v", fileName, receiveErr)
 		}
 		dst.Write(resp.FileContent)
+		progressedBytes += int64(len(resp.FileContent))
+		if progressFn != nil {
+			if !progressFn(progressedBytes) {
+				return modifiedTsNs, fmt.Errorf("interrupted copy operation")
+			}
+		}
 		wt.MaybeSlowdown(int64(len(resp.FileContent)))
 	}
 	return modifiedTsNs, nil

weed/server/volume_grpc_erasure_coding.go

@@ -130,7 +130,7 @@ func (vs *VolumeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_serv
 
 		// copy ec data slices
 		for _, shardId := range req.ShardIds {
-			if _, err := vs.doCopyFile(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, dataBaseFileName, erasure_coding.ToExt(int(shardId)), false, false); err != nil {
+			if _, err := vs.doCopyFile(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, dataBaseFileName, erasure_coding.ToExt(int(shardId)), false, false, nil); err != nil {
 				return err
 			}
 		}
@@ -138,7 +138,7 @@ func (vs *VolumeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_serv
 		if req.CopyEcxFile {
 
 			// copy ecx file
-			if _, err := vs.doCopyFile(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, indexBaseFileName, ".ecx", false, false); err != nil {
+			if _, err := vs.doCopyFile(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, indexBaseFileName, ".ecx", false, false, nil); err != nil {
 				return err
 			}
 			return nil
@@ -146,14 +146,14 @@ func (vs *VolumeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_serv
 
 		if req.CopyEcjFile {
 			// copy ecj file
-			if _, err := vs.doCopyFile(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, indexBaseFileName, ".ecj", true, true); err != nil {
+			if _, err := vs.doCopyFile(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, indexBaseFileName, ".ecj", true, true, nil); err != nil {
 				return err
 			}
 		}
 
 		if req.CopyVifFile {
 			// copy vif file
-			if _, err := vs.doCopyFile(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, dataBaseFileName, ".vif", false, true); err != nil {
+			if _, err := vs.doCopyFile(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, dataBaseFileName, ".vif", false, true, nil); err != nil {
 				return err
 			}
 		}

weed/server/volume_grpc_tier_upload.go

@@ -62,13 +62,8 @@ func (vs *VolumeServer) VolumeTierMoveDatToRemote(req *volume_server_pb.VolumeTi
 		})
 	}
 
-	// remember the file original source
-	attributes := make(map[string]string)
-	attributes["volumeId"] = v.Id.String()
-	attributes["collection"] = v.Collection
-	attributes["ext"] = ".dat"
 	// copy the data file
-	key, size, err := backendStorage.CopyFile(diskFile.File, attributes, fn)
+	key, size, err := backendStorage.CopyFile(diskFile.File, fn)
 	if err != nil {
 		return fmt.Errorf("backend %s copy file %s: %v", req.DestinationBackendName, diskFile.Name(), err)
 	}

weed/server/volume_grpc_vacuum.go

@@ -24,19 +24,35 @@ func (vs *VolumeServer) VacuumVolumeCheck(ctx context.Context, req *volume_serve
 
 }
 
-func (vs *VolumeServer) VacuumVolumeCompact(ctx context.Context, req *volume_server_pb.VacuumVolumeCompactRequest) (*volume_server_pb.VacuumVolumeCompactResponse, error) {
+func (vs *VolumeServer) VacuumVolumeCompact(req *volume_server_pb.VacuumVolumeCompactRequest, stream volume_server_pb.VolumeServer_VacuumVolumeCompactServer) error {
 
 	resp := &volume_server_pb.VacuumVolumeCompactResponse{}
+	reportInterval := int64(1024*1024*128)
+	nextReportTarget := reportInterval
 
-	err := vs.store.CompactVolume(needle.VolumeId(req.VolumeId), req.Preallocate, vs.compactionBytePerSecond)
+	var sendErr error
+	err := vs.store.CompactVolume(needle.VolumeId(req.VolumeId), req.Preallocate, vs.compactionBytePerSecond, func(processed int64) bool {
+		if processed > nextReportTarget {
+			resp.ProcessedBytes = processed
+			if sendErr = stream.Send(resp); sendErr != nil {
+				return false
+			}
+			nextReportTarget = processed + reportInterval
+		}
+		return true
+	})
 
 	if err != nil {
 		glog.Errorf("compact volume %d: %v", req.VolumeId, err)
-	} else {
-		glog.V(1).Infof("compact volume %d", req.VolumeId)
+		return err
+	}
+	if sendErr != nil {
+		glog.Errorf("compact volume %d report progress: %v", req.VolumeId, sendErr)
+		return sendErr
 	}
 
-	return resp, err
+	glog.V(1).Infof("compact volume %d", req.VolumeId)
+	return nil
 
 }
 

weed/shell/command_ec_decode.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"flag"
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb"
 	"github.com/chrislusf/seaweedfs/weed/storage/types"
 	"io"
@@ -40,6 +41,7 @@ func (c *commandEcDecode) Do(args []string, commandEnv *CommandEnv, writer io.Wr
 	encodeCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	volumeId := encodeCommand.Int("volumeId", 0, "the volume id")
 	collection := encodeCommand.String("collection", "", "the collection name")
+	forceChanges := encodeCommand.Bool("force", false, "force the encoding even if the cluster has less than recommended 4 nodes")
 	if err = encodeCommand.Parse(args); err != nil {
 		return nil
 	}
@@ -56,6 +58,17 @@ func (c *commandEcDecode) Do(args []string, commandEnv *CommandEnv, writer io.Wr
 		return err
 	}
 
+	if !*forceChanges {
+		var nodeCount int
+		eachDataNode(topologyInfo, func(dc string, rack RackId, dn *master_pb.DataNodeInfo) {
+			nodeCount++
+		})
+		if nodeCount < erasure_coding.ParityShardsCount {
+			glog.V(0).Infof("skip erasure coding with %d nodes, less than recommended %d nodes", nodeCount, erasure_coding.ParityShardsCount)
+			return nil
+		}
+	}
+
 	// volumeId is provided
 	if vid != 0 {
 		return doEcDecode(commandEnv, topologyInfo, *collection, vid)

weed/shell/command_remote_cache.go

@@ -163,10 +163,10 @@ func (c *commandRemoteCache) cacheContentData(commandEnv *CommandEnv, writer io.
 
 			remoteLocation := filer.MapFullPathToRemoteStorageLocation(localMountedDir, remoteMountedLocation, dir.Child(entry.Name))
 
-			if err := filer.DownloadToLocal(commandEnv, remoteConf, remoteLocation, dir, entry); err != nil {
-				fmt.Fprintf(writer, "DownloadToLocal %+v: %v\n", remoteLocation, err)
+			if err := filer.CacheRemoteObjectToLocalCluster(commandEnv, remoteConf, remoteLocation, dir, entry); err != nil {
+				fmt.Fprintf(writer, "CacheRemoteObjectToLocalCluster %+v: %v\n", remoteLocation, err)
 				if executionErr == nil {
-					executionErr = fmt.Errorf("DownloadToLocal %+v: %v\n", remoteLocation, err)
+					executionErr = fmt.Errorf("CacheRemoteObjectToLocalCluster %+v: %v\n", remoteLocation, err)
 				}
 				return
 			}

weed/shell/command_volume_copy.go

@@ -53,6 +53,6 @@ func (c *commandVolumeCopy) Do(args []string, commandEnv *CommandEnv, writer io.
 		return fmt.Errorf("source and target volume servers are the same!")
 	}
 
-	_, err = copyVolume(commandEnv.option.GrpcDialOption, volumeId, sourceVolumeServer, targetVolumeServer, "")
+	_, err = copyVolume(commandEnv.option.GrpcDialOption, writer, volumeId, sourceVolumeServer, targetVolumeServer, "")
 	return
 }

weed/shell/command_volume_fix_replication.go

@@ -255,13 +255,27 @@ func (c *commandVolumeFixReplication) fixOneUnderReplicatedVolume(commandEnv *Co
 			}
 
 			err := operation.WithVolumeServerClient(pb.NewServerAddressFromDataNode(dst.dataNode), commandEnv.option.GrpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
-				_, replicateErr := volumeServerClient.VolumeCopy(context.Background(), &volume_server_pb.VolumeCopyRequest{
+				stream, replicateErr := volumeServerClient.VolumeCopy(context.Background(), &volume_server_pb.VolumeCopyRequest{
 					VolumeId:       replica.info.Id,
 					SourceDataNode: string(pb.NewServerAddressFromDataNode(replica.location.dataNode)),
 				})
 				if replicateErr != nil {
 					return fmt.Errorf("copying from %s => %s : %v", replica.location.dataNode.Id, dst.dataNode.Id, replicateErr)
 				}
+				for {
+					resp, recvErr := stream.Recv()
+					if recvErr != nil {
+						if recvErr == io.EOF {
+							break
+						} else {
+							return recvErr
+						}
+					}
+					if resp.ProcessedBytes > 0 {
+						fmt.Fprintf(writer, "volume %d processed %d bytes\n", replica.info.Id, resp.ProcessedBytes)
+					}
+				}
+
 				return nil
 			})
 

weed/shell/command_volume_move.go

@@ -78,7 +78,7 @@ func (c *commandVolumeMove) Do(args []string, commandEnv *CommandEnv, writer io.
 func LiveMoveVolume(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, sourceVolumeServer, targetVolumeServer pb.ServerAddress, idleTimeout time.Duration, diskType string, skipTailError bool) (err error) {
 
 	log.Printf("copying volume %d from %s to %s", volumeId, sourceVolumeServer, targetVolumeServer)
-	lastAppendAtNs, err := copyVolume(grpcDialOption, volumeId, sourceVolumeServer, targetVolumeServer, diskType)
+	lastAppendAtNs, err := copyVolume(grpcDialOption, writer, volumeId, sourceVolumeServer, targetVolumeServer, diskType)
 	if err != nil {
 		return fmt.Errorf("copy volume %d from %s to %s: %v", volumeId, sourceVolumeServer, targetVolumeServer, err)
 	}
@@ -101,7 +101,7 @@ func LiveMoveVolume(grpcDialOption grpc.DialOption, writer io.Writer, volumeId n
 	return nil
 }
 
-func copyVolume(grpcDialOption grpc.DialOption, volumeId needle.VolumeId, sourceVolumeServer, targetVolumeServer pb.ServerAddress, diskType string) (lastAppendAtNs uint64, err error) {
+func copyVolume(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, sourceVolumeServer, targetVolumeServer pb.ServerAddress, diskType string) (lastAppendAtNs uint64, err error) {
 
 	// check to see if the volume is already read-only and if its not then we need
 	// to mark it as read-only and then before we return we need to undo what we
@@ -141,15 +141,31 @@ func copyVolume(grpcDialOption grpc.DialOption, volumeId needle.VolumeId, source
 	}
 
 	err = operation.WithVolumeServerClient(targetVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
-		resp, replicateErr := volumeServerClient.VolumeCopy(context.Background(), &volume_server_pb.VolumeCopyRequest{
+		stream, replicateErr := volumeServerClient.VolumeCopy(context.Background(), &volume_server_pb.VolumeCopyRequest{
 			VolumeId:       uint32(volumeId),
 			SourceDataNode: string(sourceVolumeServer),
 			DiskType:       diskType,
 		})
-		if replicateErr == nil {
-			lastAppendAtNs = resp.LastAppendAtNs
+		if replicateErr != nil {
+			return replicateErr
 		}
-		return replicateErr
+		for {
+			resp, recvErr := stream.Recv()
+			if recvErr != nil {
+				if recvErr == io.EOF {
+					break
+				} else {
+					return recvErr
+				}
+			}
+			if resp.LastAppendAtNs != 0 {
+				lastAppendAtNs = resp.LastAppendAtNs
+			} else {
+				fmt.Fprintf(writer, "volume %d processed %d bytes\n", volumeId, resp.ProcessedBytes)
+			}
+		}
+
+		return nil
 	})
 
 	return

weed/storage/backend/backend.go

@@ -25,7 +25,7 @@ type BackendStorageFile interface {
 type BackendStorage interface {
 	ToProperties() map[string]string
 	NewStorageFile(key string, tierInfo *volume_server_pb.VolumeInfo) BackendStorageFile
-	CopyFile(f *os.File, attributes map[string]string, fn func(progressed int64, percentage float32) error) (key string, size int64, err error)
+	CopyFile(f *os.File, fn func(progressed int64, percentage float32) error) (key string, size int64, err error)
 	DownloadFile(fileName string, key string, fn func(progressed int64, percentage float32) error) (size int64, err error)
 	DeleteFile(key string) (err error)
 }

weed/storage/backend/s3_backend/s3_backend.go

@@ -79,13 +79,13 @@ func (s *S3BackendStorage) NewStorageFile(key string, tierInfo *volume_server_pb
 	return f
 }
 
-func (s *S3BackendStorage) CopyFile(f *os.File, attributes map[string]string, fn func(progressed int64, percentage float32) error) (key string, size int64, err error) {
+func (s *S3BackendStorage) CopyFile(f *os.File, fn func(progressed int64, percentage float32) error) (key string, size int64, err error) {
 	randomUuid, _ := uuid.NewRandom()
 	key = randomUuid.String()
 
 	glog.V(1).Infof("copying dat file of %s to remote s3.%s as %s", f.Name(), s.id, key)
 
-	size, err = uploadToS3(s.conn, f.Name(), s.bucket, key, attributes, fn)
+	size, err = uploadToS3(s.conn, f.Name(), s.bucket, key, fn)
 
 	return
 }

weed/storage/backend/s3_backend/s3_sessions.go

@@ -2,6 +2,8 @@ package s3_backend
 
 import (
 	"fmt"
+	"github.com/aws/aws-sdk-go/aws/request"
+	"github.com/chrislusf/seaweedfs/weed/util"
 	"sync"
 
 	"github.com/aws/aws-sdk-go/aws"
@@ -47,6 +49,9 @@ func createSession(awsAccessKeyId, awsSecretAccessKey, region, endpoint string)
 	if err != nil {
 		return nil, fmt.Errorf("create aws session in region %s: %v", region, err)
 	}
+	sess.Handlers.Build.PushBack(func(r *request.Request) {
+		r.HTTPRequest.Header.Set("User-Agent", "SeaweedFS/"+util.VERSION_NUMBER)
+	})
 
 	t := s3.New(sess)
 

weed/storage/backend/s3_backend/s3_upload.go

@@ -12,9 +12,7 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/glog"
 )
 
-func uploadToS3(sess s3iface.S3API, filename string, destBucket string, destKey string,
-	attributes map[string]string,
-	fn func(progressed int64, percentage float32) error) (fileSize int64, err error) {
+func uploadToS3(sess s3iface.S3API, filename string, destBucket string, destKey string, fn func(progressed int64, percentage float32) error) (fileSize int64, err error) {
 
 	//open the file
 	f, err := os.Open(filename)
@@ -48,25 +46,13 @@ func uploadToS3(sess s3iface.S3API, filename string, destBucket string, destKey
 		fn:   fn,
 	}
 
-	// process tagging
-	tags := ""
-	for k, v := range attributes {
-		if len(tags) > 0 {
-			tags = tags + "&"
-		}
-		tags = tags + k + "=" + v
-	}
-
 	// Upload the file to S3.
 	var result *s3manager.UploadOutput
 	result, err = uploader.Upload(&s3manager.UploadInput{
 		Bucket:               aws.String(destBucket),
 		Key:                  aws.String(destKey),
 		Body:                 fileReader,
-		ACL:                  aws.String("private"),
-		ServerSideEncryption: aws.String("AES256"),
 		StorageClass:         aws.String("STANDARD_IA"),
-		Tagging:              aws.String(tags),
 	})
 
 	//in case it fails to upload

weed/storage/store_vacuum.go

@@ -15,13 +15,13 @@ func (s *Store) CheckCompactVolume(volumeId needle.VolumeId) (float64, error) {
 	}
 	return 0, fmt.Errorf("volume id %d is not found during check compact", volumeId)
 }
-func (s *Store) CompactVolume(vid needle.VolumeId, preallocate int64, compactionBytePerSecond int64) error {
+func (s *Store) CompactVolume(vid needle.VolumeId, preallocate int64, compactionBytePerSecond int64, progressFn ProgressFunc) error {
 	if v := s.findVolume(vid); v != nil {
 		s := stats.NewDiskStatus(v.dir)
 		if int64(s.Free) < preallocate {
 			return fmt.Errorf("free space: %d bytes, not enough for %d bytes", s.Free, preallocate)
 		}
-		return v.Compact2(preallocate, compactionBytePerSecond)
+		return v.Compact2(preallocate, compactionBytePerSecond, progressFn)
 	}
 	return fmt.Errorf("volume id %d is not found during compact", vid)
 }

weed/storage/volume_vacuum.go

@@ -17,6 +17,8 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
+type ProgressFunc func(processed int64) bool
+
 func (v *Volume) garbageLevel() float64 {
 	if v.ContentSize() == 0 {
 		return 0
@@ -62,7 +64,7 @@ func (v *Volume) Compact(preallocate int64, compactionBytePerSecond int64) error
 }
 
 // compact a volume based on deletions in .idx files
-func (v *Volume) Compact2(preallocate int64, compactionBytePerSecond int64) error {
+func (v *Volume) Compact2(preallocate int64, compactionBytePerSecond int64, progressFn ProgressFunc) error {
 
 	if v.MemoryMapMaxSizeMb != 0 { //it makes no sense to compact in memory
 		return nil
@@ -83,7 +85,7 @@ func (v *Volume) Compact2(preallocate int64, compactionBytePerSecond int64) erro
 	if err := v.nm.Sync(); err != nil {
 		glog.V(0).Infof("compact2 fail to sync volume idx %d: %v", v.Id, err)
 	}
-	return copyDataBasedOnIndexFile(v.FileName(".dat"), v.FileName(".idx"), v.FileName(".cpd"), v.FileName(".cpx"), v.SuperBlock, v.Version(), preallocate, compactionBytePerSecond)
+	return copyDataBasedOnIndexFile(v.FileName(".dat"), v.FileName(".idx"), v.FileName(".cpd"), v.FileName(".cpx"), v.SuperBlock, v.Version(), preallocate, compactionBytePerSecond, progressFn)
 }
 
 func (v *Volume) CommitCompact() error {
@@ -382,7 +384,7 @@ func (v *Volume) copyDataAndGenerateIndexFile(dstName, idxName string, prealloca
 	return
 }
 
-func copyDataBasedOnIndexFile(srcDatName, srcIdxName, dstDatName, datIdxName string, sb super_block.SuperBlock, version needle.Version, preallocate int64, compactionBytePerSecond int64) (err error) {
+func copyDataBasedOnIndexFile(srcDatName, srcIdxName, dstDatName, datIdxName string, sb super_block.SuperBlock, version needle.Version, preallocate, compactionBytePerSecond int64, progressFn ProgressFunc) (err error) {
 	var (
 		srcDatBackend, dstDatBackend backend.BackendStorageFile
 		dataFile                     *os.File
@@ -421,6 +423,12 @@ func copyDataBasedOnIndexFile(srcDatName, srcIdxName, dstDatName, datIdxName str
 			return nil
 		}
 
+		if progressFn != nil {
+			if !progressFn(offset.ToActualOffset()) {
+				return fmt.Errorf("interrupted")
+			}
+		}
+
 		n := new(needle.Needle)
 		err := n.ReadData(srcDatBackend, offset.ToActualOffset(), size, version)
 		if err != nil {

weed/storage/volume_vacuum_test.go

@@ -83,7 +83,7 @@ func TestCompaction(t *testing.T) {
 	}
 
 	startTime := time.Now()
-	v.Compact2(0, 0)
+	v.Compact2(0, 0, nil)
 	speed := float64(v.ContentSize()) / time.Now().Sub(startTime).Seconds()
 	t.Logf("compaction speed: %.2f bytes/s", speed)
 

weed/topology/topology_vacuum.go

@@ -3,6 +3,7 @@ package topology
 import (
 	"context"
 	"github.com/chrislusf/seaweedfs/weed/pb"
+	"io"
 	"sync/atomic"
 	"time"
 
@@ -70,11 +71,26 @@ func (t *Topology) batchVacuumVolumeCompact(grpcDialOption grpc.DialOption, vl *
 		go func(index int, url pb.ServerAddress, vid needle.VolumeId) {
 			glog.V(0).Infoln(index, "Start vacuuming", vid, "on", url)
 			err := operation.WithVolumeServerClient(url, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
-				_, err := volumeServerClient.VacuumVolumeCompact(context.Background(), &volume_server_pb.VacuumVolumeCompactRequest{
+				stream, err := volumeServerClient.VacuumVolumeCompact(context.Background(), &volume_server_pb.VacuumVolumeCompactRequest{
 					VolumeId:    uint32(vid),
 					Preallocate: preallocate,
 				})
-				return err
+				if err != nil {
+					return err
+				}
+
+				for {
+					resp, recvErr := stream.Recv()
+					if recvErr != nil {
+						if recvErr == io.EOF {
+							break
+						} else {
+							return recvErr
+						}
+					}
+					glog.V(0).Infof("%d vacuum %d on %s processed %d bytes", index, vid, url, resp.ProcessedBytes)
+				}
+				return nil
 			})
 			if err != nil {
 				glog.Errorf("Error when vacuuming %d on %s: %v", vid, url, err)

weed/util/constants.go

@@ -5,7 +5,7 @@ import (
 )
 
 var (
-	VERSION_NUMBER = fmt.Sprintf("%.02f", 2.72)
+	VERSION_NUMBER = fmt.Sprintf("%.02f", 2.76)
 	VERSION        = sizeLimit + " " + VERSION_NUMBER
 	COMMIT         = ""
 )