2.75

fix https://github.com/chrislusf/seaweedfs/issues/2386

k8s/helm_charts2/Chart.yaml

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

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/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/s3api_object_tagging_handlers.go

@@ -90,7 +90,7 @@ func (s3a *S3ApiServer) PutObjectTaggingHandler(w http.ResponseWriter, r *http.R
 		return
 	}
 
-	w.WriteHeader(http.StatusNoContent)
+	w.WriteHeader(http.StatusOK)
 
 }
 

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_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)

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_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_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/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.74)
+	VERSION_NUMBER = fmt.Sprintf("%.02f", 2.75)
 	VERSION        = sizeLimit + " " + VERSION_NUMBER
 	COMMIT         = ""
 )