diff --git a/weed/server/master_grpc_server_volume.go b/weed/server/master_grpc_server_volume.go index 77154972b..06dbb4bb5 100644 --- a/weed/server/master_grpc_server_volume.go +++ b/weed/server/master_grpc_server_volume.go @@ -309,12 +309,19 @@ func (ms *MasterServer) VolumeMarkReadonly(ctx context.Context, req *master_pb.V replicaPlacement, _ := super_block.NewReplicaPlacementFromByte(byte(req.ReplicaPlacement)) vl := ms.Topo.GetVolumeLayout(req.Collection, replicaPlacement, needle.LoadTTLFromUint32(req.Ttl), types.ToDiskType(req.DiskType)) dataNodes := ms.Topo.Lookup(req.Collection, needle.VolumeId(req.VolumeId)) + if req.IsReadonly { + for _, dn := range dataNodes { + if dn.Ip == req.Ip && dn.Port == int(req.Port) { + + } + } + } for _, dn := range dataNodes { if dn.Ip == req.Ip && dn.Port == int(req.Port) { if req.IsReadonly { - vl.SetVolumeUnavailable(dn, needle.VolumeId(req.VolumeId)) + vl.SetVolumeReadOnly(dn, needle.VolumeId(req.VolumeId)) } else { - vl.SetVolumeAvailable(dn, needle.VolumeId(req.VolumeId), false) + vl.SetVolumeWritable(dn, needle.VolumeId(req.VolumeId)) } } } diff --git a/weed/topology/volume_layout.go b/weed/topology/volume_layout.go index 377122cae..d1bc28ef8 100644 --- a/weed/topology/volume_layout.go +++ b/weed/topology/volume_layout.go @@ -395,6 +395,31 @@ func (vl *VolumeLayout) setVolumeWritable(vid needle.VolumeId) bool { return true } +func (vl *VolumeLayout) SetVolumeReadOnly(dn *DataNode, vid needle.VolumeId) bool { + vl.accessLock.Lock() + defer vl.accessLock.Unlock() + + if _, ok := vl.vid2location[vid]; ok { + vl.readonlyVolumes.Add(vid, dn) + return vl.removeFromWritable(vid) + } + return true +} + +func (vl *VolumeLayout) SetVolumeWritable(dn *DataNode, vid needle.VolumeId) bool { + vl.accessLock.Lock() + defer vl.accessLock.Unlock() + + if _, ok := vl.vid2location[vid]; ok { + vl.readonlyVolumes.Remove(vid, dn) + } + + if vl.enoughCopies(vid) { + return vl.setVolumeWritable(vid) + } + return false +} + func (vl *VolumeLayout) SetVolumeUnavailable(dn *DataNode, vid needle.VolumeId) bool { vl.accessLock.Lock() defer vl.accessLock.Unlock()