diff --git a/weed/topology/duplicate_volume_id_test.go b/weed/topology/duplicate_volume_id_test.go index fad5ad5174..79d498409c 100644 --- a/weed/topology/duplicate_volume_id_test.go +++ b/weed/topology/duplicate_volume_id_test.go @@ -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") } } diff --git a/weed/topology/topology.go b/weed/topology/topology.go index 29fe5de6e9..d5e139a829 100644 --- a/weed/topology/topology.go +++ b/weed/topology/topology.go @@ -9,6 +9,7 @@ import ( "os" "path/filepath" "slices" + "strings" "sync" "sync/atomic" "time" @@ -104,6 +105,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 { @@ -111,19 +117,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 } @@ -645,6 +659,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 @@ -660,25 +677,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) + } } } @@ -695,11 +722,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 { @@ -724,15 +751,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 @@ -750,16 +793,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) diff --git a/weed/topology/volume_layout.go b/weed/topology/volume_layout.go index 39ac21c5e7..4219c8eff3 100644 --- a/weed/topology/volume_layout.go +++ b/weed/topology/volume_layout.go @@ -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