volume: add "-dir.idx" option for separate index storage

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

1 files changed, 18 insertions, 13 deletions

diff --git a/weed/server/volume_grpc_copy.go b/weed/server/volume_grpc_copy.go
index 2aecb140f..cfa3710a8 100644
--- a/weed/server/volume_grpc_copy.go
+++ b/weed/server/volume_grpc_copy.go
@@ -48,7 +48,7 @@ func (vs *VolumeServer) VolumeCopy(ctx context.Context, req *volume_server_pb.Vo
 	//   send .dat file
 	//   confirm size and timestamp
 	var volFileInfoResp *volume_server_pb.ReadVolumeFileStatusResponse
-	var volumeFileName, idxFileName, datFileName string
+	var dataBaseFileName, indexBaseFileName, idxFileName, datFileName string
 	err := operation.WithVolumeServerClient(req.SourceDataNode, vs.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
 		var err error
 		volFileInfoResp, err = client.ReadVolumeFileStatus(context.Background(),
@@ -59,24 +59,25 @@ func (vs *VolumeServer) VolumeCopy(ctx context.Context, req *volume_server_pb.Vo
 			return fmt.Errorf("read volume file status failed, %v", err)
 		}
 
-		volumeFileName = storage.VolumeFileName(location.Directory, volFileInfoResp.Collection, int(req.VolumeId))
+		dataBaseFileName = storage.VolumeFileName(location.Directory, volFileInfoResp.Collection, int(req.VolumeId))
+		indexBaseFileName = storage.VolumeFileName(location.IdxDirectory, volFileInfoResp.Collection, int(req.VolumeId))
 
-		ioutil.WriteFile(volumeFileName+".note", []byte(fmt.Sprintf("copying from %s", req.SourceDataNode)), 0755)
+		ioutil.WriteFile(dataBaseFileName+".note", []byte(fmt.Sprintf("copying from %s", req.SourceDataNode)), 0755)
 
 		// println("source:", volFileInfoResp.String())
-		if err := vs.doCopyFile(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.DatFileSize, volumeFileName, ".dat", false, true); err != nil {
+		if err := vs.doCopyFile(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.DatFileSize, dataBaseFileName, ".dat", false, true); err != nil {
 			return err
 		}
 
-		if err := vs.doCopyFile(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.IdxFileSize, volumeFileName, ".idx", false, false); err != nil {
+		if err := vs.doCopyFile(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.IdxFileSize, indexBaseFileName, ".idx", false, false); err != nil {
 			return err
 		}
 
-		if err := vs.doCopyFile(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.DatFileSize, volumeFileName, ".vif", false, true); err != nil {
+		if err := vs.doCopyFile(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.DatFileSize, dataBaseFileName, ".vif", false, true); err != nil {
 			return err
 		}
 
-		os.Remove(volumeFileName + ".note")
+		os.Remove(dataBaseFileName + ".note")
 
 		return nil
 	})
@@ -84,18 +85,18 @@ func (vs *VolumeServer) VolumeCopy(ctx context.Context, req *volume_server_pb.Vo
 	if err != nil {
 		return nil, err
 	}
-	if volumeFileName == "" {
+	if dataBaseFileName == "" {
 		return nil, fmt.Errorf("not found volume %d file", req.VolumeId)
 	}
 
-	idxFileName = volumeFileName + ".idx"
-	datFileName = volumeFileName + ".dat"
+	idxFileName = indexBaseFileName + ".idx"
+	datFileName = dataBaseFileName + ".dat"
 
 	defer func() {
-		if err != nil && volumeFileName != "" {
+		if err != nil && dataBaseFileName != "" {
 			os.Remove(idxFileName)
 			os.Remove(datFileName)
-			os.Remove(volumeFileName + ".vif")
+			os.Remove(dataBaseFileName + ".vif")
 		}
 	}()
 
@@ -223,7 +224,7 @@ func (vs *VolumeServer) CopyFile(req *volume_server_pb.CopyFileRequest, stream v
 		if uint32(v.CompactionRevision) != req.CompactionRevision && req.CompactionRevision != math.MaxUint32 {
 			return fmt.Errorf("volume %d is compacted", req.VolumeId)
 		}
-		fileName = v.FileName() + req.Ext
+		fileName = v.FileName(req.Ext)
 	} else {
 		baseFileName := erasure_coding.EcShardBaseFileName(req.Collection, int(req.VolumeId)) + req.Ext
 		for _, location := range vs.store.Locations {
@@ -231,6 +232,10 @@ func (vs *VolumeServer) CopyFile(req *volume_server_pb.CopyFileRequest, stream v
 			if util.FileExists(tName) {
 				fileName = tName
 			}
+			tName = util.Join(location.IdxDirectory, baseFileName)
+			if util.FileExists(tName) {
+				fileName = tName
+			}
 		}
 		if fileName == "" {
 			if req.IgnoreSourceFileNotFound {