VolumeInfo readonly for both disk permission and disk free space watermark

weed/command/master.go

@@ -135,7 +135,7 @@ func checkPeers(masterIp string, masterPort int, peers string) (masterAddress st
 
 	peerCount := len(cleanedPeers)
 	if !hasSelf {
-		peerCount += 1
+		peerCount++
 	}
 	if peerCount%2 == 0 {
 		glog.Fatalf("Only odd number of masters are supported!")

weed/storage/volume_info.go

@@ -71,7 +71,14 @@ func NewVolumeInfo(m *master_pb.VolumeInformationMessage) (vi VolumeInfo, err er
 
 func (vi VolumeInfo) String() string {
 	return fmt.Sprintf("Id:%d, Size:%d, ReplicaPlacement:%s, Collection:%s, Version:%v, FileCount:%d, DeleteCount:%d, DeletedByteCount:%d, ReadOnly:%v",
-		vi.Id, vi.Size, vi.ReplicaPlacement, vi.Collection, vi.Version, vi.FileCount, vi.DeleteCount, vi.DeletedByteCount, vi.ReadOnly)
+		vi.Id, vi.Size, vi.ReplicaPlacement, vi.Collection, vi.Version, vi.FileCount, vi.DeleteCount, vi.DeletedByteCount, vi.IsReadOnly())
+}
+
+// IsReadOnly tells the current vi is readonly or not, including the following conditions:
+// 1. the data file is readonly;
+// 2. the disk free space is under the watermark;
+func (vi VolumeInfo) IsReadOnly() bool {
+	return vi.ReadOnly || vi.DiskWatermark > 0 && vi.DiskFree <= vi.DiskWatermark
 }
 
 /*VolumesInfo sorting*/

weed/topology/data_node.go

@@ -38,7 +38,7 @@ func (dn *DataNode) AddOrUpdateVolume(v storage.VolumeInfo) (isNew bool) {
 	if _, ok := dn.volumes[v.Id]; !ok {
 		dn.volumes[v.Id] = v
 		dn.UpAdjustVolumeCountDelta(1)
-		if !v.ReadOnly {
+		if !v.IsReadOnly() {
 			dn.UpAdjustActiveVolumeCountDelta(1)
 		}
 		dn.UpAdjustMaxVolumeId(v.Id)

weed/topology/volume_growth.go

@@ -35,6 +35,16 @@ func (o *VolumeGrowOption) String() string {
 	return fmt.Sprintf("Collection:%s, ReplicaPlacement:%v, Ttl:%v, DataCenter:%s, Rack:%s, DataNode:%s", o.Collection, o.ReplicaPlacement, o.Ttl, o.DataCenter, o.Rack, o.DataNode)
 }
 
+func (o *VolumeGrowOption) MatchesDataCenter(dn *DataNode) bool {
+	return o.DataCenter == "" || dn.GetDataCenter().Id() == NodeId(o.DataCenter)
+}
+
+func (o *VolumeGrowOption) MatchesRackDataNode(dn *DataNode) bool {
+	matchesRack := o.Rack == "" || dn.GetRack().Id() == NodeId(o.Rack)
+	matchesDataNode := o.DataNode == "" || dn.Id() == NodeId(o.DataNode)
+	return matchesRack && matchesDataNode
+}
+
 func NewDefaultVolumeGrowth() *VolumeGrowth {
 	return &VolumeGrowth{}
 }

weed/topology/volume_layout.go

@@ -56,7 +56,7 @@ func (vl *VolumeLayout) RegisterVolume(v *storage.VolumeInfo, dn *DataNode) {
 	// glog.V(4).Infof("volume %d added to %s len %d copy %d", v.Id, dn.Id(), vl.vid2location[v.Id].Length(), v.ReplicaPlacement.GetCopyCount())
 	for _, dn := range vl.vid2location[v.Id].list {
 		if vInfo, err := dn.GetVolumesById(v.Id); err == nil {
-			if vInfo.ReadOnly {
+			if vInfo.IsReadOnly() {
 				glog.V(3).Infof("vid %d removed from writable", v.Id)
 				vl.removeFromWritable(v.Id)
 				vl.readonlyVolumes[v.Id] = true
@@ -111,7 +111,7 @@ func (vl *VolumeLayout) isOversized(v *storage.VolumeInfo) bool {
 func (vl *VolumeLayout) isWritable(v *storage.VolumeInfo) bool {
 	return !vl.isOversized(v) &&
 		v.Version == storage.CurrentVersion &&
-		!v.ReadOnly
+		!v.IsReadOnly()
 }
 
 func (vl *VolumeLayout) isEmpty() bool {
@@ -164,13 +164,11 @@ func (vl *VolumeLayout) PickForWrite(count uint64, option *VolumeGrowOption) (*s
 	for _, v := range vl.writables {
 		volumeLocationList := vl.vid2location[v]
 		for _, dn := range volumeLocationList.list {
-			if dn.GetDataCenter().Id() == NodeId(option.DataCenter) {
-				if option.Rack != "" && dn.GetRack().Id() != NodeId(option.Rack) {
-					continue
-				}
-				if option.DataNode != "" && dn.Id() != NodeId(option.DataNode) {
+			if option.MatchesDataCenter(dn) {
+				if !option.MatchesRackDataNode(dn) {
 					continue
 				}
+
 				counter++
 				if rand.Intn(counter) < 1 {
 					vid, locationList = v, volumeLocationList
@@ -191,13 +189,11 @@ func (vl *VolumeLayout) GetActiveVolumeCount(option *VolumeGrowOption) int {
 	counter := 0
 	for _, v := range vl.writables {
 		for _, dn := range vl.vid2location[v].list {
-			if dn.GetDataCenter().Id() == NodeId(option.DataCenter) {
-				if option.Rack != "" && dn.GetRack().Id() != NodeId(option.Rack) {
-					continue
-				}
-				if option.DataNode != "" && dn.Id() != NodeId(option.DataNode) {
+			if option.MatchesDataCenter(dn) {
+				if !option.MatchesRackDataNode(dn) {
 					continue
 				}
+
 				counter++
 			}
 		}