adds VolumeEcGenerateSlices, VolumeEcCopy

weed/server/volume_grpc_copy.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"math"
 	"os"
 	"time"
 
@@ -49,37 +50,13 @@ func (vs *VolumeServer) VolumeCopy(ctx context.Context, req *volume_server_pb.Vo
 		volumeFileName = storage.VolumeFileName(volFileInfoResp.Collection, location.Directory, int(req.VolumeId))
 
 		// println("source:", volFileInfoResp.String())
-
-		copyFileClient, err := client.CopyFile(ctx, &volume_server_pb.CopyFileRequest{
-			VolumeId:           req.VolumeId,
-			Ext:                ".idx",
-			CompactionRevision: volFileInfoResp.CompactionRevision,
-			StopOffset:         volFileInfoResp.IdxFileSize,
-		})
-		if err != nil {
-			return fmt.Errorf("failed to start copying volume %d idx file: %v", req.VolumeId, err)
+		// copy ecx file
+		if err:=vs.doCopyFile(ctx, client, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.IdxFileSize, volumeFileName, ".idx"); err!=nil{
+			return err
 		}
 
-		idxFileName = volumeFileName + ".idx"
-		err = writeToFile(copyFileClient, idxFileName, util.NewWriteThrottler(vs.compactionBytePerSecond))
-		if err != nil {
-			return fmt.Errorf("failed to copy volume %d idx file: %v", req.VolumeId, err)
-		}
-
-		copyFileClient, err = client.CopyFile(ctx, &volume_server_pb.CopyFileRequest{
-			VolumeId:           req.VolumeId,
-			Ext:                ".dat",
-			CompactionRevision: volFileInfoResp.CompactionRevision,
-			StopOffset:         volFileInfoResp.DatFileSize,
-		})
-		if err != nil {
-			return fmt.Errorf("failed to start copying volume %d dat file: %v", req.VolumeId, err)
-		}
-
-		datFileName = volumeFileName + ".dat"
-		err = writeToFile(copyFileClient, datFileName, util.NewWriteThrottler(vs.compactionBytePerSecond))
-		if err != nil {
-			return fmt.Errorf("failed to copy volume %d dat file: %v", req.VolumeId, err)
+		if err:=vs.doCopyFile(ctx, client, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.DatFileSize, volumeFileName, ".dat"); err!=nil{
+			return err
 		}
 
 		return nil
@@ -109,6 +86,28 @@ func (vs *VolumeServer) VolumeCopy(ctx context.Context, req *volume_server_pb.Vo
 	}, err
 }
 
+func (vs *VolumeServer) doCopyFile(ctx context.Context, client volume_server_pb.VolumeServerClient, vid uint32,
+	compactRevision uint32, stopOffset uint64, baseFileName, ext string) error {
+
+	copyFileClient, err := client.CopyFile(ctx, &volume_server_pb.CopyFileRequest{
+		VolumeId:           vid,
+		Ext:                ext,
+		CompactionRevision: compactRevision,
+		StopOffset:         stopOffset,
+	})
+	if err != nil {
+		return fmt.Errorf("failed to start copying volume %d %s file: %v", vid, ext, err)
+	}
+
+	err = writeToFile(copyFileClient, baseFileName+ext, util.NewWriteThrottler(vs.compactionBytePerSecond))
+	if err != nil {
+		return fmt.Errorf("failed to copy volume %d %s file: %v", vid, ext, err)
+	}
+
+	return nil
+
+}
+
 /**
 only check the the differ of the file size
 todo: maybe should check the received count and deleted count of the volume
@@ -175,6 +174,9 @@ func (vs *VolumeServer) ReadVolumeFileStatus(ctx context.Context, req *volume_se
 	return resp, nil
 }
 
+// CopyFile client pulls the volume related file from the source server.
+// if req.CompactionRevision != math.MaxUint32, it ensures the compact revision is as expected
+// The copying still stop at req.StopOffset, but you can set it to math.MaxUint64 in order to read all data.
 func (vs *VolumeServer) CopyFile(req *volume_server_pb.CopyFileRequest, stream volume_server_pb.VolumeServer_CopyFileServer) error {
 
 	v := vs.store.GetVolume(needle.VolumeId(req.VolumeId))
@@ -182,7 +184,7 @@ func (vs *VolumeServer) CopyFile(req *volume_server_pb.CopyFileRequest, stream v
 		return fmt.Errorf("not found volume id %d", req.VolumeId)
 	}
 
-	if uint32(v.CompactionRevision) != req.CompactionRevision {
+	if uint32(v.CompactionRevision) != req.CompactionRevision && req.CompactionRevision != math.MaxUint32 {
 		return fmt.Errorf("volume %d is compacted", req.VolumeId)
 	}