volume server options add diskWaterMark (default 10g)

weed/command/server.go

@@ -92,7 +92,7 @@ func init() {
 	serverOptions.v.fixJpgOrientation = cmdServer.Flag.Bool("volume.images.fix.orientation", false, "Adjust jpg orientation when uploading.")
 	serverOptions.v.readRedirect = cmdServer.Flag.Bool("volume.read.redirect", true, "Redirect moved or non-local volumes.")
 	serverOptions.v.publicUrl = cmdServer.Flag.String("volume.publicUrl", "", "publicly accessible address")
-
+	serverOptions.v.diskWaterMark = cmdVolume.Flag.String("volume.diskWaterMark", "10GB", "disk watermark low to switch volume to read-only(eg 5G,100M,5GiB,100MiB)")
 }
 
 func runServer(cmd *Command, args []string) bool {

weed/command/volume.go

@@ -14,6 +14,7 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/server"
 	"github.com/chrislusf/seaweedfs/weed/storage"
 	"github.com/chrislusf/seaweedfs/weed/util"
+	"github.com/dustin/go-humanize"
 	"google.golang.org/grpc/reflection"
 )
 
@@ -41,6 +42,7 @@ type VolumeServerOptions struct {
 	readRedirect          *bool
 	cpuProfile            *string
 	memProfile            *string
+	diskWaterMark         *string
 }
 
 func init() {
@@ -61,6 +63,7 @@ func init() {
 	v.readRedirect = cmdVolume.Flag.Bool("read.redirect", true, "Redirect moved or non-local volumes.")
 	v.cpuProfile = cmdVolume.Flag.String("cpuprofile", "", "cpu profile output file")
 	v.memProfile = cmdVolume.Flag.String("memprofile", "", "memory profile output file")
+	v.diskWaterMark = cmdVolume.Flag.String("diskWaterMark", "10g", "disk watermark low to switch volume to read-only(eg 5G,100M,5GiB,100MiB)")
 }
 
 var cmdVolume = &Command{
@@ -133,26 +136,22 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v
 		publicVolumeMux = http.NewServeMux()
 	}
 
-	volumeNeedleMapKind := storage.NeedleMapInMemory
-	switch *v.indexType {
-	case "leveldb":
-		volumeNeedleMapKind = storage.NeedleMapLevelDb
-	case "boltdb":
-		volumeNeedleMapKind = storage.NeedleMapBoltDb
-	case "btree":
-		volumeNeedleMapKind = storage.NeedleMapBtree
-	}
+	volumeNeedleMapKind := storage.ParseVolumeNeedleMapKind(*v.indexType)
 
 	masters := *v.masters
 
+	diskWaterMarkBytes, err := humanize.ParseBytes(*v.diskWaterMark)
+	if err != nil {
+		glog.Fatalf("Check diskWaterMark %s error:%v ", *v.diskWaterMark, err)
+	}
+
 	volumeServer := weed_server.NewVolumeServer(volumeMux, publicVolumeMux,
 		*v.ip, *v.port, *v.publicUrl,
 		v.folders, v.folderMaxLimits,
 		volumeNeedleMapKind,
 		strings.Split(masters, ","), *v.pulseSeconds, *v.dataCenter, *v.rack,
 		v.whiteList,
-		*v.fixJpgOrientation, *v.readRedirect,
-	)
+		*v.fixJpgOrientation, *v.readRedirect, diskWaterMarkBytes)
 
 	listeningAddress := *v.bindIp + ":" + strconv.Itoa(*v.port)
 	glog.V(0).Infof("Start Seaweed volume server %s at %s", util.VERSION, listeningAddress)

weed/server/volume_server.go

@@ -30,7 +30,8 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
 	dataCenter string, rack string,
 	whiteList []string,
 	fixJpgOrientation bool,
-	readRedirect bool) *VolumeServer {
+	readRedirect bool,
+	diskWaterMark uint64) *VolumeServer {
 	vs := &VolumeServer{
 		pulseSeconds:      pulseSeconds,
 		dataCenter:        dataCenter,
@@ -40,7 +41,7 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
 		ReadRedirect:      readRedirect,
 	}
 	vs.MasterNodes = masterNodes
-	vs.store = storage.NewStore(port, ip, publicUrl, folders, maxCounts, vs.needleMapKind)
+	vs.store = storage.NewStore(port, ip, publicUrl, folders, maxCounts, vs.needleMapKind, diskWaterMark)
 
 	vs.guard = security.NewGuard(whiteList, "")
 

weed/storage/needle_map.go

@@ -20,6 +20,19 @@ const (
 	NeedleMapBtree
 )
 
+func ParseVolumeNeedleMapKind(indexType string) NeedleMapType {
+	switch indexType {
+	case "leveldb":
+		return NeedleMapLevelDb
+	case "boltdb":
+		return NeedleMapBoltDb
+	case "btree":
+		return NeedleMapBtree
+	default:
+		return NeedleMapInMemory
+	}
+}
+
 type NeedleMapper interface {
 	Put(key NeedleId, offset Offset, size uint32) error
 	Get(key NeedleId) (element *needle.NeedleValue, ok bool)

weed/storage/store.go

@@ -37,8 +37,9 @@ func (s *Store) String() (str string) {
 	return
 }
 
-func NewStore(port int, ip, publicUrl string, dirNames []string, maxVolumeCounts []int, needleMapKind NeedleMapType) (s *Store) {
-	s = &Store{Port: port, Ip: ip, PublicUrl: publicUrl, NeedleMapType: needleMapKind}
+func NewStore(port int, ip, publicUrl string, dirNames []string, maxVolumeCounts []int, needleMapKind NeedleMapType,
+	diskWaterMark uint64) (s *Store) {
+	s = &Store{Port: port, Ip: ip, PublicUrl: publicUrl, NeedleMapType: needleMapKind, DiskWatermark: diskWaterMark}
 	s.Locations = make([]*DiskLocation, 0)
 	for i := 0; i < len(dirNames); i++ {
 		location := NewDiskLocation(dirNames[i], maxVolumeCounts[i])