mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-10-21 22:12:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

master: retry unenforced quarantine marks and restore server writability on release (#11763)

Browse Source 
Dial the advertised grpc port, return mark errors, and keep a
quarantine entry until the writable push succeeds so a released
copy does not stay read-only on the server.
This commit is contained in:
Chris Lu
2026-10-20 05:38:25 -07:00
parent bf048c88b8
commit 9815bb4e59
2 changed files with 26 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
weed/server/master_server.go
+8 -7
View File
@@ -165,20 +165,21 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.Se
ms.Topo.SetGrowthFreezeConfig(
v.GetBool("master.volume_growth.freeze_on_topology_mismatch"),
v.GetDuration("master.volume_growth.state_loss_recovery_wait"))
ms.Topo.QuarantineVolumeFn = func(dn *topology.DataNode, vid needle.VolumeId, readOnly bool) {
if err := operation.WithVolumeServerClient(false, pb.ServerAddress(dn.Url()), ms.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
ms.Topo.QuarantineVolumeFn = func(dn *topology.DataNode, vid needle.VolumeId, readOnly bool) error {
err := operation.WithVolumeServerClient(false, dn.ServerAddress(), ms.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
var err error
if readOnly {
_, err = client.VolumeMarkReadonly(ctx, &volume_server_pb.VolumeMarkReadonlyRequest{VolumeId: uint32(vid), CanDelete: true})
} else {
_, err = client.VolumeMarkWritable(ctx, &volume_server_pb.VolumeMarkWritableRequest{VolumeId: uint32(vid)})
_, err := client.VolumeMarkReadonly(ctx, &volume_server_pb.VolumeMarkReadonlyRequest{VolumeId: uint32(vid), CanDelete: true})
return err
}
_, err := client.VolumeMarkWritable(ctx, &volume_server_pb.VolumeMarkWritableRequest{VolumeId: uint32(vid)})
return err
}); err != nil {
})
if err != nil {
glog.Warningf("quarantine volume %d on %s readOnly=%v: %v", vid, dn.Id(), readOnly, err)
}
return err
}
ms.vg = topology.NewDefaultVolumeGrowth()
glog.V(0).Infoln("Volume Size Limit is", ms.option.VolumeSizeLimitMB, "MB")
weed/topology/topology.go
+18 -8
View File
@@ -109,7 +109,7 @@ type Topology struct {
// QuarantineVolumeFn pushes the quarantine onto the volume server itself
// (VolumeMarkReadonly/VolumeMarkWritable). nil keeps quarantine at the
// layout level only.
QuarantineVolumeFn func(dn *DataNode, vid needle.VolumeId, readOnly bool)
QuarantineVolumeFn func(dn *DataNode, vid needle.VolumeId, readOnly bool) error
}
type quarantinedVolume struct {
@@ -138,12 +138,15 @@ func (t *Topology) enforceQuarantine(dn *DataNode, vi storage.VolumeInfo) storag
state := v.(quarantinedVolumeState)
if !state.enforced {
state.reported = vi.ReadOnly
t.quarantinedVolumeIds.Store(q, state)
} else if !vi.ReadOnly && t.QuarantineVolumeFn != nil {
// The server lost the pushed mark (restart, manual unquarantine);
// re-push it while the conflict stands.
t.QuarantineVolumeFn(dn, vi.Id, true)
}
if t.QuarantineVolumeFn != nil && (!state.enforced || !vi.ReadOnly) {
// The mark never landed, or the server lost it (restart, manual
// unquarantine); push it while the conflict stands.
if t.QuarantineVolumeFn(dn, vi.Id, true) == nil {
state.enforced = true
}
}
t.quarantinedVolumeIds.Store(q, state)
}
return vi
}
@@ -706,8 +709,7 @@ func (t *Topology) flagDuplicateVolumeId(v storage.VolumeInfo, dn *DataNode) {
if vl != nil {
vl.SetVolumeQuarantined(n, v.Id)
}
if t.QuarantineVolumeFn != nil {
t.QuarantineVolumeFn(n, v.Id, true)
if t.QuarantineVolumeFn != nil && t.QuarantineVolumeFn(n, v.Id, true) == nil {
if val, ok := t.quarantinedVolumeIds.Load(q); ok {
state := val.(quarantinedVolumeState)
state.enforced = true
@@ -745,6 +747,13 @@ func (t *Topology) releaseResolvedQuarantine(vid needle.VolumeId) {
vl.SetReplicaReadOnlyFlag(dn, vid, true)
} else {
vl.SetVolumeWritable(dn, vid)
if state.enforced && t.QuarantineVolumeFn != nil {
if err := t.QuarantineVolumeFn(dn, vid, false); err != nil {
// The server still rejects writes; keep the mark so a
// later release pass retries the writable push.
return true
}
}
}
t.quarantinedVolumeIds.Delete(key)
return true
@@ -781,6 +790,7 @@ func volumeLayoutKey(collection string, rp *super_block.ReplicaPlacement, ttl *n
return fmt.Sprintf("collection=%q rp=%s ttl=%s disk=%s", collection, rp, ttl, diskType)
}
// findDuplicateVolumeLayouts reports every layout that already holds v.Id on
// data nodes other than dn. Volume ids are global, so the same id held by two
// nodes under two layouts means the allocator reused an id.