mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2025-09-24 18:03:38 +08:00
fix volume not found if marked as read only
fix https://github.com/seaweedfs/seaweedfs/issues/4088
This commit is contained in:
@@ -309,12 +309,19 @@ func (ms *MasterServer) VolumeMarkReadonly(ctx context.Context, req *master_pb.V
|
|||||||
replicaPlacement, _ := super_block.NewReplicaPlacementFromByte(byte(req.ReplicaPlacement))
|
replicaPlacement, _ := super_block.NewReplicaPlacementFromByte(byte(req.ReplicaPlacement))
|
||||||
vl := ms.Topo.GetVolumeLayout(req.Collection, replicaPlacement, needle.LoadTTLFromUint32(req.Ttl), types.ToDiskType(req.DiskType))
|
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))
|
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 {
|
for _, dn := range dataNodes {
|
||||||
if dn.Ip == req.Ip && dn.Port == int(req.Port) {
|
if dn.Ip == req.Ip && dn.Port == int(req.Port) {
|
||||||
if req.IsReadonly {
|
if req.IsReadonly {
|
||||||
vl.SetVolumeUnavailable(dn, needle.VolumeId(req.VolumeId))
|
vl.SetVolumeReadOnly(dn, needle.VolumeId(req.VolumeId))
|
||||||
} else {
|
} else {
|
||||||
vl.SetVolumeAvailable(dn, needle.VolumeId(req.VolumeId), false)
|
vl.SetVolumeWritable(dn, needle.VolumeId(req.VolumeId))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@@ -395,6 +395,31 @@ func (vl *VolumeLayout) setVolumeWritable(vid needle.VolumeId) bool {
|
|||||||
return true
|
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 {
|
func (vl *VolumeLayout) SetVolumeUnavailable(dn *DataNode, vid needle.VolumeId) bool {
|
||||||
vl.accessLock.Lock()
|
vl.accessLock.Lock()
|
||||||
defer vl.accessLock.Unlock()
|
defer vl.accessLock.Unlock()
|
||||||
|
Reference in New Issue
Block a user