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

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

1 files changed, 35 insertions, 30 deletions

diff --git a/weed/storage/volume_loading.go b/weed/storage/volume_loading.go
index 05684cbdb..fe4980e31 100644
--- a/weed/storage/volume_loading.go
+++ b/weed/storage/volume_loading.go
@@ -23,7 +23,6 @@ func loadVolumeWithoutIndex(dirname string, collection string, id needle.VolumeI
 }
 
 func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind NeedleMapType, preallocate int64) (err error) {
-	fileName := v.FileName()
 	alreadyHasSuperBlock := false
 
 	hasVolumeInfoFile := v.maybeLoadVolumeInfo() && v.volumeInfo.Version != 0
@@ -34,17 +33,17 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind
 		glog.V(0).Infof("loading volume %d from remote %v", v.Id, v.volumeInfo.Files)
 		v.LoadRemoteFile()
 		alreadyHasSuperBlock = true
-	} else if exists, canRead, canWrite, modifiedTime, fileSize := util.CheckFile(fileName + ".dat"); exists {
+	} else if exists, canRead, canWrite, modifiedTime, fileSize := util.CheckFile(v.FileName(".dat")); exists {
 		// open dat file
 		if !canRead {
-			return fmt.Errorf("cannot read Volume Data file %s.dat", fileName)
+			return fmt.Errorf("cannot read Volume Data file %s", v.FileName(".dat"))
 		}
 		var dataFile *os.File
 		if canWrite {
-			dataFile, err = os.OpenFile(fileName+".dat", os.O_RDWR|os.O_CREATE, 0644)
+			dataFile, err = os.OpenFile(v.FileName(".dat"), os.O_RDWR|os.O_CREATE, 0644)
 		} else {
-			glog.V(0).Infoln("opening " + fileName + ".dat in READONLY mode")
-			dataFile, err = os.Open(fileName + ".dat")
+			glog.V(0).Infof("opening %s in READONLY mode", v.FileName(".dat"))
+			dataFile, err = os.Open(v.FileName(".dat"))
 			v.noWriteOrDelete = true
 		}
 		v.lastModifiedTsSeconds = uint64(modifiedTime.Unix())
@@ -54,17 +53,17 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind
 		v.DataBackend = backend.NewDiskFile(dataFile)
 	} else {
 		if createDatIfMissing {
-			v.DataBackend, err = backend.CreateVolumeFile(fileName+".dat", preallocate, v.MemoryMapMaxSizeMb)
+			v.DataBackend, err = backend.CreateVolumeFile(v.FileName(".dat"), preallocate, v.MemoryMapMaxSizeMb)
 		} else {
-			return fmt.Errorf("Volume Data file %s.dat does not exist.", fileName)
+			return fmt.Errorf("volume data file %s does not exist", v.FileName(".dat"))
 		}
 	}
 
 	if err != nil {
 		if !os.IsPermission(err) {
-			return fmt.Errorf("cannot load Volume Data %s.dat: %v", fileName, err)
+			return fmt.Errorf("cannot load volume data %s: %v", v.FileName(".dat"), err)
 		} else {
-			return fmt.Errorf("load data file %s.dat: %v", fileName, err)
+			return fmt.Errorf("load data file %s: %v", v.FileName(".dat"), err)
 		}
 	}
 
@@ -72,21 +71,27 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind
 		err = v.readSuperBlock()
 	} else {
 		if !v.SuperBlock.Initialized() {
-			return fmt.Errorf("volume %s.dat not initialized", fileName)
+			return fmt.Errorf("volume %s not initialized", v.FileName(".dat"))
 		}
 		err = v.maybeWriteSuperBlock()
 	}
 	if err == nil && alsoLoadIndex {
+		// adjust for existing volumes with .idx together with .dat files
+		if v.dirIdx != v.dir {
+			if util.FileExists(v.DataFileName()+".idx") {
+				v.dirIdx = v.dir
+			}
+		}
 		var indexFile *os.File
 		if v.noWriteOrDelete {
-			glog.V(0).Infoln("open to read file", fileName+".idx")
-			if indexFile, err = os.OpenFile(fileName+".idx", os.O_RDONLY, 0644); err != nil {
-				return fmt.Errorf("cannot read Volume Index %s.idx: %v", fileName, err)
+			glog.V(0).Infoln("open to read file", v.FileName(".idx"))
+			if indexFile, err = os.OpenFile(v.FileName(".idx"), os.O_RDONLY, 0644); err != nil {
+				return fmt.Errorf("cannot read Volume Index %s: %v",  v.FileName(".idx"), err)
 			}
 		} else {
-			glog.V(1).Infoln("open to write file", fileName+".idx")
-			if indexFile, err = os.OpenFile(fileName+".idx", os.O_RDWR|os.O_CREATE, 0644); err != nil {
-				return fmt.Errorf("cannot write Volume Index %s.idx: %v", fileName, err)
+			glog.V(1).Infoln("open to write file", v.FileName(".idx"))
+			if indexFile, err = os.OpenFile(v.FileName(".idx"), os.O_RDWR|os.O_CREATE, 0644); err != nil {
+				return fmt.Errorf("cannot write Volume Index %s: %v",  v.FileName(".idx"), err)
 			}
 		}
 		if v.lastAppendAtNs, err = CheckAndFixVolumeDataIntegrity(v, indexFile); err != nil {
@@ -95,45 +100,45 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind
 		}
 
 		if v.noWriteOrDelete || v.noWriteCanDelete {
-			if v.nm, err = NewSortedFileNeedleMap(fileName, indexFile); err != nil {
-				glog.V(0).Infof("loading sorted db %s error: %v", fileName+".sdx", err)
+			if v.nm, err = NewSortedFileNeedleMap(v.IndexFileName(), indexFile); err != nil {
+				glog.V(0).Infof("loading sorted db %s error: %v",  v.FileName(".sdx"), err)
 			}
 		} else {
 			switch needleMapKind {
 			case NeedleMapInMemory:
-				glog.V(0).Infoln("loading index", fileName+".idx", "to memory")
+				glog.V(0).Infoln("loading index", v.FileName(".idx"), "to memory")
 				if v.nm, err = LoadCompactNeedleMap(indexFile); err != nil {
-					glog.V(0).Infof("loading index %s to memory error: %v", fileName+".idx", err)
+					glog.V(0).Infof("loading index %s to memory error: %v", v.FileName(".idx"), err)
 				}
 			case NeedleMapLevelDb:
-				glog.V(0).Infoln("loading leveldb", fileName+".ldb")
+				glog.V(0).Infoln("loading leveldb", v.FileName(".ldb"))
 				opts := &opt.Options{
 					BlockCacheCapacity:            2 * 1024 * 1024, // default value is 8MiB
 					WriteBuffer:                   1 * 1024 * 1024, // default value is 4MiB
 					CompactionTableSizeMultiplier: 10,              // default value is 1
 				}
-				if v.nm, err = NewLevelDbNeedleMap(fileName+".ldb", indexFile, opts); err != nil {
-					glog.V(0).Infof("loading leveldb %s error: %v", fileName+".ldb", err)
+				if v.nm, err = NewLevelDbNeedleMap(v.FileName(".ldb"), indexFile, opts); err != nil {
+					glog.V(0).Infof("loading leveldb %s error: %v", v.FileName(".ldb"), err)
 				}
 			case NeedleMapLevelDbMedium:
-				glog.V(0).Infoln("loading leveldb medium", fileName+".ldb")
+				glog.V(0).Infoln("loading leveldb medium", v.FileName(".ldb"))
 				opts := &opt.Options{
 					BlockCacheCapacity:            4 * 1024 * 1024, // default value is 8MiB
 					WriteBuffer:                   2 * 1024 * 1024, // default value is 4MiB
 					CompactionTableSizeMultiplier: 10,              // default value is 1
 				}
-				if v.nm, err = NewLevelDbNeedleMap(fileName+".ldb", indexFile, opts); err != nil {
-					glog.V(0).Infof("loading leveldb %s error: %v", fileName+".ldb", err)
+				if v.nm, err = NewLevelDbNeedleMap(v.FileName(".ldb"), indexFile, opts); err != nil {
+					glog.V(0).Infof("loading leveldb %s error: %v", v.FileName(".ldb"), err)
 				}
 			case NeedleMapLevelDbLarge:
-				glog.V(0).Infoln("loading leveldb large", fileName+".ldb")
+				glog.V(0).Infoln("loading leveldb large", v.FileName(".ldb"))
 				opts := &opt.Options{
 					BlockCacheCapacity:            8 * 1024 * 1024, // default value is 8MiB
 					WriteBuffer:                   4 * 1024 * 1024, // default value is 4MiB
 					CompactionTableSizeMultiplier: 10,              // default value is 1
 				}
-				if v.nm, err = NewLevelDbNeedleMap(fileName+".ldb", indexFile, opts); err != nil {
-					glog.V(0).Infof("loading leveldb %s error: %v", fileName+".ldb", err)
+				if v.nm, err = NewLevelDbNeedleMap(v.FileName(".ldb"), indexFile, opts); err != nil {
+					glog.V(0).Infof("loading leveldb %s error: %v", v.FileName(".ldb"), err)
 				}
 			}
 		}