volume: add a note file to avoid incomplete volume files

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

weed/server/volume_grpc_copy.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"io/ioutil"
 	"math"
 	"os"
 	"time"
@@ -60,6 +61,8 @@ func (vs *VolumeServer) VolumeCopy(ctx context.Context, req *volume_server_pb.Vo
 
 		volumeFileName = storage.VolumeFileName(location.Directory, volFileInfoResp.Collection, int(req.VolumeId))
 
+		ioutil.WriteFile(volumeFileName+".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 {
 			return err
@@ -73,6 +76,8 @@ func (vs *VolumeServer) VolumeCopy(ctx context.Context, req *volume_server_pb.Vo
 			return err
 		}
 
+		os.Remove(volumeFileName+".note")
+
 		return nil
 	})
 

weed/storage/disk_location.go

@@ -13,6 +13,7 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/stats"
 	"github.com/chrislusf/seaweedfs/weed/storage/erasure_coding"
 	"github.com/chrislusf/seaweedfs/weed/storage/needle"
+	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
 type DiskLocation struct {
@@ -60,6 +61,13 @@ func parseCollectionVolumeId(base string) (collection string, vid needle.VolumeI
 func (l *DiskLocation) loadExistingVolume(fileInfo os.FileInfo, needleMapKind NeedleMapType) bool {
 	name := fileInfo.Name()
 	if !fileInfo.IsDir() && strings.HasSuffix(name, ".idx") {
+		noteFile := l.Directory + "/" + name + ".note"
+		if util.FileExists(noteFile) {
+			note, _ := ioutil.ReadFile(noteFile)
+			glog.Warningf("volume %s was not completed: %s", name, string(note))
+			removeVolumeFiles(l.Directory + "/" + name)
+			return false
+		}
 		vid, collection, err := l.volumeIdFromPath(fileInfo)
 		if err != nil {
 			glog.Warningf("get volume id failed, %s, err : %s", name, err)

weed/storage/volume_read_write.go

@@ -56,16 +56,21 @@ func (v *Volume) Destroy() (err error) {
 		}
 	}
 	v.Close()
-	os.Remove(v.FileName() + ".dat")
-	os.Remove(v.FileName() + ".idx")
-	os.Remove(v.FileName() + ".vif")
-	os.Remove(v.FileName() + ".sdx")
-	os.Remove(v.FileName() + ".cpd")
-	os.Remove(v.FileName() + ".cpx")
-	os.RemoveAll(v.FileName() + ".ldb")
+	removeVolumeFiles(v.FileName())
 	return
 }
 
+func removeVolumeFiles(filename string) {
+	os.Remove(filename+ ".dat")
+	os.Remove(filename + ".idx")
+	os.Remove(filename + ".vif")
+	os.Remove(filename + ".sdx")
+	os.Remove(filename + ".cpd")
+	os.Remove(filename + ".cpx")
+	os.RemoveAll(filename + ".ldb")
+	os.Remove(filename + ".note")
+}
+
 func (v *Volume) asyncRequestAppend(request *needle.AsyncRequest) {
 	v.asyncRequestsChan <- request
 }