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: quarantine every conflicting layout at the layout level (#11763)

Browse Source 
- collect all conflicting layouts, not just the first, so a vid spanning
  three layouts does not leave one writable
- mark quarantine on the location list only: folding it into the stored
  report made the volume digest permanently disagree with the server and
  forced a full volume list every heartbeat
- isAllWritable now honors the location read-only mark so a quarantined
  copy stays out of writables through resyncs and re-registration
- re-check the conflict before keeping marks, and do not clear a node's
  quarantine while it still holds the vid in another layout
- log the full layout keys on both sides of the conflict
This commit is contained in:
Chris Lu
2026-10-19 22:07:35 -07:00
parent a3fb27ae2d
commit 7ffffa9ea4
3 changed files with 109 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
weed/topology/duplicate_volume_id_test.go
+26 -35
View File
@@ -7,8 +7,20 @@ import (
"github.com/seaweedfs/seaweedfs/weed/storage"
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
"github.com/seaweedfs/seaweedfs/weed/storage/types"
)
func isWritable(t *testing.T, topo *Topology, v storage.VolumeInfo) bool {
t.Helper()
vl := topo.FindVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, types.ToDiskType(v.DiskType))
for _, id := range vl.CloneWritableVolumes() {
if id == v.Id {
return true
}
}
return false
}
func TestDuplicateVolumeIdAcrossCollections(t *testing.T) {
topo := NewTopology("topo", nil, 128, 3, false)
rp, _ := super_block.NewReplicaPlacementFromString("000")
@@ -25,20 +37,11 @@ func TestDuplicateVolumeIdAcrossCollections(t *testing.T) {
topo.RegisterVolumeLayout(v2, dn2)
topo.flagDuplicateVolumeId(v2, dn2)
vi, err := dn2.GetVolumesById(7)
if err != nil {
t.Fatalf("volume info not stored on dn2: %v", err)
if isWritable(t, topo, v2) {
t.Fatal("duplicate newcomer should be quarantined out of the writable list")
}
if !vi.ReadOnly {
t.Fatal("duplicate newcomer should be marked read-only")
}
vi, err = dn1.GetVolumesById(7)
if err != nil {
t.Fatalf("volume info not stored on dn1: %v", err)
}
if !vi.ReadOnly {
t.Fatal("either copy could be the impostor; the earlier copy must go read-only too")
if isWritable(t, topo, v1) {
t.Fatal("either copy could be the impostor; the earlier copy must be quarantined too")
}
}
@@ -47,7 +50,11 @@ func TestSameLayoutReplicaIsNotDuplicate(t *testing.T) {
rp, _ := super_block.NewReplicaPlacementFromString("001")
dn1 := NewDataNode("server1")
dn1.Ip = "10.0.0.1"
dn1.Port = 8080
dn2 := NewDataNode("server2")
dn2.Ip = "10.0.0.2"
dn2.Port = 8080
v := storage.VolumeInfo{Id: 7, Collection: "a", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL}
dn1.UpdateVolumes([]storage.VolumeInfo{v})
@@ -56,11 +63,7 @@ func TestSameLayoutReplicaIsNotDuplicate(t *testing.T) {
topo.RegisterVolumeLayout(v, dn2)
topo.flagDuplicateVolumeId(v, dn2)
vi, err := dn2.GetVolumesById(7)
if err != nil {
t.Fatalf("volume info not stored on dn2: %v", err)
}
if vi.ReadOnly {
if !isWritable(t, topo, v) {
t.Fatal("a same-layout replica is a legitimate copy, not a duplicate")
}
}
@@ -80,11 +83,7 @@ func TestSameNodeVolumeMoveIsNotDuplicate(t *testing.T) {
topo.RegisterVolumeLayout(moved, dn)
topo.flagDuplicateVolumeId(moved, dn)
vi, err := dn.GetVolumesById(7)
if err != nil {
t.Fatalf("volume info not stored on dn: %v", err)
}
if vi.ReadOnly {
if !isWritable(t, topo, moved) {
t.Fatal("a same-node disk move is not a duplicate")
}
}
@@ -105,11 +104,7 @@ func TestQuarantineReleasedWhenConflictResolves(t *testing.T) {
topo.UnRegisterVolumeLayout(v2, dn2)
vi, err := dn1.GetVolumesById(7)
if err != nil {
t.Fatalf("volume info not stored on dn1: %v", err)
}
if vi.ReadOnly {
if !isWritable(t, topo, v1) {
t.Fatal("the surviving copy must be released once the conflict is gone")
}
}
@@ -129,16 +124,12 @@ func TestQuarantineSurvivesResync(t *testing.T) {
topo.flagDuplicateVolumeId(v2, dn2)
// a resync from the same server re-reports the volume as writable;
// the quarantine must keep it read-only
// the quarantine must keep it out of the writable list
topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{{
Id: 7, Collection: "b", ReplicaPlacement: 0, Version: 3,
}}, dn2)
vi, err := dn2.GetVolumesById(7)
if err != nil {
t.Fatalf("volume info not stored on dn2: %v", err)
}
if !vi.ReadOnly {
t.Fatal("resync must not clear the quarantine's read-only mark")
if isWritable(t, topo, v2) {
t.Fatal("resync must not clear the quarantine")
}
}
weed/topology/topology.go
+67 -25
View File
@@ -9,6 +9,7 @@ import (
"os"
"path/filepath"
"slices"
"strings"
"sync"
"sync/atomic"
"time"
@@ -103,6 +104,11 @@ type Topology struct {
// vid collided across layouts. Kept separate from the reported flag so
// later heartbeats from the same server cannot quietly undo it.
quarantinedVolumeIds sync.Map
// 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)
}
type quarantinedVolume struct {
@@ -110,19 +116,27 @@ type quarantinedVolume struct {
vid needle.VolumeId
}
type quarantinedVolumeState struct {
reported bool
enforced bool
}
func (t *Topology) isQuarantined(dn *DataNode, vid needle.VolumeId) bool {
_, ok := t.quarantinedVolumeIds.Load(quarantinedVolume{dn.Id(), vid})
return ok
}
// enforceQuarantine keeps a quarantined copy read-only even when the server's
// own reports claim it writable again. The entry also tracks the latest
// reported flag so a later release restores the server's true state.
// enforceQuarantine keeps the latest reported read-only flag for a quarantined
// copy so a later release restores the server's true state. The stored report
// keeps the server's flag: quarantine lives at the layout level, and folding
// it into the record here would make the master's volume digest permanently
// disagree with the server's and force a full volume list on every heartbeat.
func (t *Topology) enforceQuarantine(dn *DataNode, vi storage.VolumeInfo) storage.VolumeInfo {
q := quarantinedVolume{dn.Id(), vi.Id}
if _, ok := t.quarantinedVolumeIds.Load(q); ok {
t.quarantinedVolumeIds.Store(q, vi.ReadOnly)
vi.ReadOnly = true
if v, ok := t.quarantinedVolumeIds.Load(q); ok {
state := v.(quarantinedVolumeState)
state.reported = vi.ReadOnly
t.quarantinedVolumeIds.Store(q, state)
}
return vi
}
@@ -644,6 +658,9 @@ func (t *Topology) RegisterVolumeLayout(v storage.VolumeInfo, dn *DataNode) {
for {
vl := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType)
if vl.RegisterVolume(&v, dn) {
if t.isQuarantined(dn, v.Id) {
vl.SetVolumeQuarantined(dn, v.Id)
}
vl.EnsureCorrectWritables(&v)
go t.flagDuplicateVolumeId(v, dn)
return
@@ -659,25 +676,35 @@ func (t *Topology) RegisterVolumeLayout(v storage.VolumeInfo, dn *DataNode) {
// quarantined: either could be the impostor, and arrival order must not
// decide which copy stays writable.
func (t *Topology) flagDuplicateVolumeId(v storage.VolumeInfo, dn *DataNode) {
dupCollection, dupLayout, nodes := t.findDuplicateVolumeLayout(v, dn)
if len(nodes) == 0 {
hits := t.findDuplicateVolumeLayouts(v, dn)
if len(hits) == 0 {
return
}
// A conflict could resolve between the scan above and the marks below;
// re-run the release so a stale quarantine is cleared instead of kept.
defer t.releaseResolvedQuarantine(v.Id)
stats.MasterDuplicateVolumeIds.WithLabelValues(v.Collection, v.Id.String()).Inc()
glog.Errorf("duplicate volume id %d: %s reports it under collection %q but %v also serves it under collection %q; marking all copies read-only", v.Id, dn.Id(), v.Collection, nodes, dupCollection)
var otherKeys []string
for _, h := range hits {
otherKeys = append(otherKeys, h.layoutKey())
}
glog.Errorf("duplicate volume id %d: %s reports it under %s but other data nodes also serve it under %s; marking all copies read-only", v.Id, dn.Id(), volumeLayoutKey(v.Collection, v.ReplicaPlacement, v.Ttl, types.ToDiskType(v.DiskType)), strings.Join(otherKeys, ", "))
quarantine := func(n *DataNode, vl *VolumeLayout) {
q := quarantinedVolume{n.Id(), v.Id}
reported := false
if vi, err := n.GetVolumesById(v.Id); err == nil {
reported = vi.ReadOnly
}
t.quarantinedVolumeIds.LoadOrStore(quarantinedVolume{n.Id(), v.Id}, reported)
t.quarantinedVolumeIds.LoadOrStore(q, quarantinedVolumeState{reported: reported})
if vl != nil {
vl.SetVolumeReadOnly(n, v.Id)
vl.SetVolumeQuarantined(n, v.Id)
}
}
quarantine(dn, t.FindVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, types.ToDiskType(v.DiskType)))
for _, n := range nodes {
quarantine(n, dupLayout)
for _, h := range hits {
for _, n := range h.nodes {
quarantine(n, h.vl)
}
}
}
@@ -694,11 +721,11 @@ func (t *Topology) releaseResolvedQuarantine(vid needle.VolumeId) {
t.quarantinedVolumeIds.Delete(key)
return true
}
if _, _, nodes := t.findDuplicateVolumeLayout(vi, dn); len(nodes) > 0 {
if hits := t.findDuplicateVolumeLayouts(vi, dn); len(hits) > 0 {
return true
}
reported, _ := val.(bool)
if reported {
state, _ := val.(quarantinedVolumeState)
if state.reported {
dn.SetVolumeReadOnly(vid, true)
vl.SetReplicaReadOnlyFlag(dn, vid, true)
} else {
@@ -723,15 +750,31 @@ func (t *Topology) findQuarantinedVolume(dnId NodeId, vid needle.VolumeId) (*Dat
return nil, nil, storage.VolumeInfo{}
}
// findDuplicateVolumeLayout reports the layout and the other data nodes
// already holding v.Id under a different layout key. Volume ids are global,
// so the same id held by two nodes under two layouts means the allocator
// reused an id.
func (t *Topology) findDuplicateVolumeLayout(v storage.VolumeInfo, dn *DataNode) (string, *VolumeLayout, []*DataNode) {
// duplicateHit is one layout that holds v.Id on data nodes other than the
// reporter's.
type duplicateHit struct {
collection string
vl *VolumeLayout
nodes []*DataNode
}
func (h duplicateHit) layoutKey() string {
return volumeLayoutKey(h.collection, h.vl.rp, h.vl.ttl, h.vl.diskType)
}
func volumeLayoutKey(collection string, rp *super_block.ReplicaPlacement, ttl *needle.TTL, diskType types.DiskType) string {
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.
func (t *Topology) findDuplicateVolumeLayouts(v storage.VolumeInfo, dn *DataNode) []duplicateHit {
if v.ReplicaPlacement == nil {
return "", nil, nil
return nil
}
diskType := types.ToDiskType(v.DiskType)
var hits []duplicateHit
for _, item := range t.collectionMap.Items() {
c := item.(*Collection)
var same *VolumeLayout
@@ -749,16 +792,15 @@ func (t *Topology) findDuplicateVolumeLayout(v storage.VolumeInfo, dn *DataNode)
}
}
if len(others) > 0 {
return c.Name, vl, others
hits = append(hits, duplicateHit{collection: c.Name, vl: vl, nodes: others})
}
}
}
return "", nil, nil
return hits
}
func (t *Topology) UnRegisterVolumeLayout(v storage.VolumeInfo, dn *DataNode) {
glog.Infof("removing volume info: %+v from %v", v, dn.id)
t.quarantinedVolumeIds.Delete(quarantinedVolume{dn.Id(), v.Id})
defer t.releaseResolvedQuarantine(v.Id)
diskType := types.ToDiskType(v.DiskType)
volumeLayout := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType)
weed/topology/volume_layout.go
+16
View File
@@ -411,6 +411,9 @@ func (vl *VolumeLayout) ensureCorrectWritables(vid needle.VolumeId) {
// or one on a server in maintenance mode, holds the whole volume out.
func (vl *VolumeLayout) isAllWritable(vid needle.VolumeId) bool {
if location, ok := vl.vid2location[vid]; ok {
if location.AnyReadOnly() {
return false
}
for _, dn := range location.list {
if dn.InMaintenanceMode() {
return false
@@ -1030,6 +1033,19 @@ func (vl *VolumeLayout) SetVolumeWritable(dn *DataNode, vid needle.VolumeId) boo
return false
}
// SetVolumeQuarantined marks dn's replica read-only at the layout level only.
// The node's stored report is left untouched so the volume digest still tracks
// what the server reports and quarantined copies do not force endless full
// heartbeats.
func (vl *VolumeLayout) SetVolumeQuarantined(dn *DataNode, vid needle.VolumeId) {
vl.accessLock.Lock()
defer vl.accessLock.Unlock()
if location, ok := vl.vid2location[vid]; ok {
location.SetReadOnly(dn, true)
}
vl.removeFromWritable(vid)
}
// SetReplicaReadOnlyFlag records what dn's latest heartbeat said about its
// replica of vid, and nothing more: the writable list is left to
// EnsureCorrectWritables and its capacity guards. Until now the flag only moved