collect ec shard from multiple locations

fix https://github.com/seaweedfs/seaweedfs/issues/4365

1 files changed, 10 insertions, 2 deletions

diff --git a/weed/server/volume_grpc_erasure_coding.go b/weed/server/volume_grpc_erasure_coding.go
index c780a6cdf..55659ca0f 100644
--- a/weed/server/volume_grpc_erasure_coding.go
+++ b/weed/server/volume_grpc_erasure_coding.go
@@ -408,7 +408,9 @@ func (vs *VolumeServer) VolumeEcShardsToVolume(ctx context.Context, req *volume_
 
 	glog.V(0).Infof("VolumeEcShardsToVolume: %v", req)
 
-	v, found := vs.store.FindEcVolume(needle.VolumeId(req.VolumeId))
+	// collect .ec00 ~ .ec09 files
+	shardFileNames := make([]string, erasure_coding.DataShardsCount)
+	v, found := vs.store.CollectEcShards(needle.VolumeId(req.VolumeId), shardFileNames)
 	if !found {
 		return nil, fmt.Errorf("ec volume %d not found", req.VolumeId)
 	}
@@ -417,6 +419,12 @@ func (vs *VolumeServer) VolumeEcShardsToVolume(ctx context.Context, req *volume_
 		return nil, fmt.Errorf("existing collection:%v unexpected input: %v", v.Collection, req.Collection)
 	}
 
+	for shardId := 0; shardId < DataShardsCount; shardId++ {
+		if shardFileNames[shardId] == "" {
+			return nil, fmt.Errorf("ec volume %d missing shard %d", req.VolumeId, shardId)
+		}
+	}
+
 	dataBaseFileName, indexBaseFileName := v.DataBaseFileName(), v.IndexBaseFileName()
 	// calculate .dat file size
 	datFileSize, err := erasure_coding.FindDatFileSize(dataBaseFileName, indexBaseFileName)
@@ -425,7 +433,7 @@ func (vs *VolumeServer) VolumeEcShardsToVolume(ctx context.Context, req *volume_
 	}
 
 	// write .dat file from .ec00 ~ .ec09 files
-	if err := erasure_coding.WriteDatFile(dataBaseFileName, datFileSize); err != nil {
+	if err := erasure_coding.WriteDatFile(dataBaseFileName, datFileSize, shardFileNames); err != nil {
 		return nil, fmt.Errorf("WriteDatFile %s: %v", dataBaseFileName, err)
 	}