if replicated volume has one copy in readonly mode at one node,it should be removed from writable list

2 files changed, 24 insertions, 0 deletions

diff --git a/weed/topology/data_node.go b/weed/topology/data_node.go
index 1404d4aa8..da6ed2895 100644
--- a/weed/topology/data_node.go
+++ b/weed/topology/data_node.go
@@ -79,6 +79,17 @@ func (dn *DataNode) GetVolumes() (ret []storage.VolumeInfo) {
 	return ret
 }
 
+func (dn *DataNode) GetVolumesById(id storage.VolumeId) (storage.VolumeInfo, error) {
+	dn.RLock()
+	defer dn.RUnlock()
+	v_info, ok := dn.volumes[id]
+	if ok {
+		return v_info, nil
+	} else {
+		return storage.VolumeInfo{}, fmt.Errorf("volumeInfo not found")
+	}
+}
+
 func (dn *DataNode) GetDataCenter() *DataCenter {
 	return dn.Parent().Parent().(*NodeImpl).value.(*DataCenter)
 }
diff --git a/weed/topology/volume_layout.go b/weed/topology/volume_layout.go
index 066f5f69a..af8503b29 100644
--- a/weed/topology/volume_layout.go
+++ b/weed/topology/volume_layout.go
@@ -45,6 +45,19 @@ func (vl *VolumeLayout) RegisterVolume(v *storage.VolumeInfo, dn *DataNode) {
 	}
 	vl.vid2location[v.Id].Set(dn)
 	glog.V(4).Infoln("volume", v.Id, "added to dn", dn.Id(), "len", vl.vid2location[v.Id].Length(), "copy", v.ReplicaPlacement.GetCopyCount())
+	for _, dn := range vl.vid2location[v.Id].list {
+		if v_info, err := dn.GetVolumesById(v.Id); err == nil {
+			if v_info.ReadOnly {
+				glog.V(3).Infof("vid %d removed from writable", v.Id)
+				vl.removeFromWritable(v.Id)
+				return
+			}
+		} else {
+			glog.V(3).Infof("vid %d removed from writable", v.Id)
+			vl.removeFromWritable(v.Id)
+			return
+		}
+	}
 	if vl.vid2location[v.Id].Length() == vl.rp.GetCopyCount() && vl.isWritable(v) {
 		if _, ok := vl.oversizedVolumes[v.Id]; !ok {
 			vl.addToWritable(v.Id)