From 4d0674266449e579cb20d8be6ea3461c768f3af4 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 20 Oct 2026 07:37:33 -0700 Subject: [PATCH] Detect and quarantine duplicate volume ids (#11763) (#11828) * master: detect and quarantine duplicate volume ids across layouts (#11763) * shell: volume.check.duplicate_ids audit command (#11763) * master: persist duplicate-vid quarantine across heartbeats (#11763) * master: quarantine all copies of a duplicated volume id Near-simultaneous registrations let each side quarantine only the other, so arrival order decided which copy stayed writable. A confirmed cross-layout collision means either copy could be the impostor, so flag every colliding copy read-only regardless of registration order. * topology: return a copy from VolumeLayout.Lookup Callers iterating the returned slice raced with registrations and removals replacing the location list in place. * master: release duplicate-vid quarantine when the conflict resolves A cross-node volume.move briefly shows the same id on two nodes under different disk layouts; permanent quarantine would pin a valid target read-only. Keep the first-reported state per quarantined copy and lift the marks once the conflicting copy unregisters, so only a persistent collision stays read-only. * shell: include ec shards in volume.check.duplicate_ids Ec-only volumes share the global id space. Compare shard collections against regular ones while allowing shards and regular copies of the same encoded volume. * master: keep quarantine release current and cover all removal paths Track the latest server-reported read-only flag while quarantined so a release restores the server s true state, and run the release check on collection deletion and data-node removal too, not only explicit volume unregistration. * shell: do not flag ec-only volumes as duplicates An ec-only volume has no regular copy, so its shard collections had nothing to compare against and always read as foreign. * master: quarantine every conflicting layout at the layout level (#11763) - 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 * master: push duplicate-vid quarantine to the volume server (#11763) A quarantined copy stayed writable on the volume server itself for a client holding an existing file id, since the master-only mark never reached the server's local write path. Quarantine now also calls VolumeMarkReadonly (can_delete so expiry still drains), is re-pushed if the server comes back reporting it writable, and VolumeMarkWritable runs when the conflict resolves and the server had reported writable. * master: retry unenforced quarantine marks and restore server writability on release (#11763) 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. * master: treat rp-only layout differences as an in-flight replication change (#11763) volume.configure.replication rolls through nodes one at a time, so a volume legitimately reports the old and new replica placement while it settles. Layouts differing only in rp within the same collection are a supported transition, not a reused id. * shell: group ec shards by collection in the duplicate id audit (#11763) EC shards of one volume legitimately span disk types when DiskTypePrefer spills, so grouping by disk type reports healthy volumes as duplicates. --- weed/server/master_server.go | 19 ++ .../command_volume_check_duplicate_ids.go | 140 +++++++++++ weed/stats/metrics.go | 9 + weed/topology/duplicate_volume_id_test.go | 187 ++++++++++++++ weed/topology/topology.go | 228 +++++++++++++++++- weed/topology/topology_event_handling.go | 2 + weed/topology/volume_layout.go | 27 ++- 7 files changed, 607 insertions(+), 5 deletions(-) create mode 100644 weed/shell/command_volume_check_duplicate_ids.go create mode 100644 weed/topology/duplicate_volume_id_test.go diff --git a/weed/server/master_server.go b/weed/server/master_server.go index b2e0de3874..b36679ebcd 100644 --- a/weed/server/master_server.go +++ b/weed/server/master_server.go @@ -15,7 +15,9 @@ import ( "time" "github.com/seaweedfs/seaweedfs/weed/cluster/maintenance" + "github.com/seaweedfs/seaweedfs/weed/operation" "github.com/seaweedfs/seaweedfs/weed/stats" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/telemetry" "github.com/seaweedfs/seaweedfs/weed/cluster" @@ -28,6 +30,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" "github.com/seaweedfs/seaweedfs/weed/security" "github.com/seaweedfs/seaweedfs/weed/sequence" "github.com/seaweedfs/seaweedfs/weed/shell" @@ -162,6 +165,22 @@ 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) 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() + if readOnly { + _, 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 + }) + 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") diff --git a/weed/shell/command_volume_check_duplicate_ids.go b/weed/shell/command_volume_check_duplicate_ids.go new file mode 100644 index 0000000000..d60474ff9e --- /dev/null +++ b/weed/shell/command_volume_check_duplicate_ids.go @@ -0,0 +1,140 @@ +package shell + +import ( + "fmt" + "io" + "slices" + + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/super_block" +) + +func init() { + Commands = append(Commands, &commandVolumeCheckDuplicateIds{}) +} + +type commandVolumeCheckDuplicateIds struct { +} + +func (c *commandVolumeCheckDuplicateIds) Name() string { + return "volume.check.duplicate_ids" +} + +func (c *commandVolumeCheckDuplicateIds) Help() string { + return `find volume ids registered under more than one layout + + volume.check.duplicate_ids + + Volume ids are global. The same id under two collections, replica + placements, TTLs, or disk types means the allocator reused an id, for + example after master state loss. See volume.growth.resume. + +` +} + +func (c *commandVolumeCheckDuplicateIds) HasTag(CommandTag) bool { + return false +} + +func (c *commandVolumeCheckDuplicateIds) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) { + topoInfo, _, err := collectTopologyInfo(commandEnv, 0) + if err != nil { + return err + } + + // vid -> layout key -> locations; ec shards tracked per collection since + // shards plus regular copies of the same encoded volume are legitimate + seen := make(map[needle.VolumeId]map[string][]string) + ecSeen := make(map[needle.VolumeId]map[string][]string) + collections := make(map[needle.VolumeId]map[string]bool) + ecCollections := make(map[needle.VolumeId]map[string]bool) + add := func(m map[needle.VolumeId]map[string][]string, id uint32, key, location string) { + layouts := m[needle.VolumeId(id)] + if layouts == nil { + layouts = make(map[string][]string) + m[needle.VolumeId(id)] = layouts + } + layouts[key] = append(layouts[key], location) + } + addCollection := func(m map[needle.VolumeId]map[string]bool, id uint32, collection string) { + if m[needle.VolumeId(id)] == nil { + m[needle.VolumeId(id)] = make(map[string]bool) + } + m[needle.VolumeId(id)][collection] = true + } + for _, dc := range topoInfo.DataCenterInfos { + for _, r := range dc.RackInfos { + for _, dn := range r.DataNodeInfos { + for diskKey, di := range dn.DiskInfos { + for _, vi := range di.VolumeInfos { + rp, _ := super_block.NewReplicaPlacementFromByte(byte(vi.ReplicaPlacement)) + add(seen, vi.Id, fmt.Sprintf("collection=%q replication=%s ttl=%s disk=%s", + vi.Collection, rp.String(), needle.LoadTTLFromUint32(vi.Ttl), vi.DiskType), + fmt.Sprintf("%s %s", dn.Id, diskKey)) + addCollection(collections, vi.Id, vi.Collection) + } + for _, si := range di.EcShardInfos { + add(ecSeen, si.Id, fmt.Sprintf("collection=%q ec_shards", si.Collection), + fmt.Sprintf("%s %s disk=%s", dn.Id, diskKey, si.DiskType)) + addCollection(ecCollections, si.Id, si.Collection) + } + } + } + } + } + + vids := make(map[needle.VolumeId]struct{}, len(seen)+len(ecSeen)) + for vid := range seen { + vids[vid] = struct{}{} + } + for vid := range ecSeen { + vids[vid] = struct{}{} + } + + duplicates := 0 + for _, vid := range sortedKeys(vids) { + layouts := seen[vid] + shardLayouts := ecSeen[vid] + ecForeign := false + if len(collections[vid]) > 0 { + for collection := range ecCollections[vid] { + if !collections[vid][collection] { + ecForeign = true + } + } + } + if len(layouts) < 2 && len(shardLayouts) < 2 && !ecForeign { + continue + } + duplicates++ + fmt.Fprintf(writer, "volume %d is registered under %d layouts:\n", vid, len(layouts)+len(shardLayouts)) + for _, key := range sortedKeysString(layouts) { + fmt.Fprintf(writer, " %-60s %v\n", key, layouts[key]) + } + for _, key := range sortedKeysString(shardLayouts) { + fmt.Fprintf(writer, " %-60s %v\n", key, shardLayouts[key]) + } + } + if duplicates == 0 { + fmt.Fprintf(writer, "no duplicate volume ids found\n") + } + return nil +} + +func sortedKeys[V any](m map[needle.VolumeId]V) []needle.VolumeId { + keys := make([]needle.VolumeId, 0, len(m)) + for k := range m { + keys = append(keys, k) + } + slices.Sort(keys) + return keys +} + +func sortedKeysString[V any](m map[string]V) []string { + keys := make([]string, 0, len(m)) + for k := range m { + keys = append(keys, k) + } + slices.Sort(keys) + return keys +} diff --git a/weed/stats/metrics.go b/weed/stats/metrics.go index 2b37256f7a..4678c227d5 100644 --- a/weed/stats/metrics.go +++ b/weed/stats/metrics.go @@ -109,6 +109,14 @@ var ( Help: "1 while volume growth is frozen after detected master state loss", }) + MasterDuplicateVolumeIds = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: Namespace, + Subsystem: subsystemMaster, + Name: "duplicate_volume_ids", + Help: "volume id registered under two different layouts", + }, []string{"collection", "volumeId"}) + MasterReceivedHeartbeatCounter = prometheus.NewCounterVec( prometheus.CounterOpts{ Namespace: Namespace, @@ -1000,6 +1008,7 @@ func init() { Gather.MustRegister(MasterRaftIsleader) Gather.MustRegister(MasterAdminLock) Gather.MustRegister(MasterVolumeGrowthFrozen) + Gather.MustRegister(MasterDuplicateVolumeIds) Gather.MustRegister(MasterReceivedHeartbeatCounter) Gather.MustRegister(MasterLeaderChangeCounter) Gather.MustRegister(MasterReplicaPlacementMismatch) diff --git a/weed/topology/duplicate_volume_id_test.go b/weed/topology/duplicate_volume_id_test.go new file mode 100644 index 0000000000..42cb6fd97a --- /dev/null +++ b/weed/topology/duplicate_volume_id_test.go @@ -0,0 +1,187 @@ +package topology + +import ( + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "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") + + dn1 := NewDataNode("server1") + dn2 := NewDataNode("server2") + + v1 := storage.VolumeInfo{Id: 7, Collection: "a", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL} + v2 := storage.VolumeInfo{Id: 7, Collection: "b", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL} + + dn1.UpdateVolumes([]storage.VolumeInfo{v1}) + dn2.UpdateVolumes([]storage.VolumeInfo{v2}) + topo.RegisterVolumeLayout(v1, dn1) + topo.RegisterVolumeLayout(v2, dn2) + topo.flagDuplicateVolumeId(v2, dn2) + + if isWritable(t, topo, v2) { + t.Fatal("duplicate newcomer should be quarantined out of the writable list") + } + if isWritable(t, topo, v1) { + t.Fatal("either copy could be the impostor; the earlier copy must be quarantined too") + } +} + +func TestSameLayoutReplicaIsNotDuplicate(t *testing.T) { + topo := NewTopology("topo", nil, 128, 3, false) + 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}) + dn2.UpdateVolumes([]storage.VolumeInfo{v}) + topo.RegisterVolumeLayout(v, dn1) + topo.RegisterVolumeLayout(v, dn2) + topo.flagDuplicateVolumeId(v, dn2) + + if !isWritable(t, topo, v) { + t.Fatal("a same-layout replica is a legitimate copy, not a duplicate") + } +} + +func TestSameNodeVolumeMoveIsNotDuplicate(t *testing.T) { + topo := NewTopology("topo", nil, 128, 3, false) + rp, _ := super_block.NewReplicaPlacementFromString("000") + dn := NewDataNode("server1") + + old := storage.VolumeInfo{Id: 7, Collection: "c", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL} + moved := storage.VolumeInfo{Id: 7, Collection: "c", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL, DiskType: "ssd"} + dn.UpdateVolumes([]storage.VolumeInfo{old}) + topo.RegisterVolumeLayout(old, dn) + + // the move registers the new layout before the old entry is removed + dn.UpdateVolumes([]storage.VolumeInfo{moved}) + topo.RegisterVolumeLayout(moved, dn) + topo.flagDuplicateVolumeId(moved, dn) + + if !isWritable(t, topo, moved) { + t.Fatal("a same-node disk move is not a duplicate") + } +} + +func TestQuarantineReleasedWhenConflictResolves(t *testing.T) { + topo := NewTopology("topo", nil, 128, 3, false) + rp, _ := super_block.NewReplicaPlacementFromString("000") + + dn1 := NewDataNode("server1") + dn2 := NewDataNode("server2") + v1 := storage.VolumeInfo{Id: 7, Collection: "a", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL} + v2 := storage.VolumeInfo{Id: 7, Collection: "b", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL} + dn1.UpdateVolumes([]storage.VolumeInfo{v1}) + dn2.UpdateVolumes([]storage.VolumeInfo{v2}) + topo.RegisterVolumeLayout(v1, dn1) + topo.RegisterVolumeLayout(v2, dn2) + topo.flagDuplicateVolumeId(v2, dn2) + + topo.UnRegisterVolumeLayout(v2, dn2) + + if !isWritable(t, topo, v1) { + t.Fatal("the surviving copy must be released once the conflict is gone") + } +} + +func TestQuarantineSurvivesResync(t *testing.T) { + topo := NewTopology("topo", nil, 128, 3, false) + rp, _ := super_block.NewReplicaPlacementFromString("000") + + dn1 := NewDataNode("server1") + dn2 := NewDataNode("server2") + v1 := storage.VolumeInfo{Id: 7, Collection: "a", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL} + v2 := storage.VolumeInfo{Id: 7, Collection: "b", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL} + dn1.UpdateVolumes([]storage.VolumeInfo{v1}) + dn2.UpdateVolumes([]storage.VolumeInfo{v2}) + topo.RegisterVolumeLayout(v1, dn1) + topo.RegisterVolumeLayout(v2, dn2) + topo.flagDuplicateVolumeId(v2, dn2) + + // a resync from the same server re-reports the volume as writable; + // the quarantine must keep it out of the writable list + topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{{ + Id: 7, Collection: "b", ReplicaPlacement: 0, Version: 3, + }}, dn2) + + if isWritable(t, topo, v2) { + t.Fatal("resync must not clear the quarantine") + } +} + +func TestReplicationChangeIsNotDuplicate(t *testing.T) { + topo := NewTopology("topo", nil, 128, 3, false) + rpOld, _ := super_block.NewReplicaPlacementFromString("000") + rpNew, _ := 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 + + vOld := storage.VolumeInfo{Id: 7, Collection: "a", ReplicaPlacement: rpOld, Ttl: needle.EMPTY_TTL} + vNew := storage.VolumeInfo{Id: 7, Collection: "a", ReplicaPlacement: rpNew, Ttl: needle.EMPTY_TTL} + + dn1.UpdateVolumes([]storage.VolumeInfo{vOld}) + dn2.UpdateVolumes([]storage.VolumeInfo{vNew}) + topo.RegisterVolumeLayout(vOld, dn1) + topo.RegisterVolumeLayout(vNew, dn2) + topo.flagDuplicateVolumeId(vNew, dn2) + + if topo.isQuarantined(dn2, vNew.Id) || topo.isQuarantined(dn1, vOld.Id) { + t.Fatal("an rp-only layout difference is a replication change in flight, not a duplicate") + } +} + +func TestDuplicateVolumeIdAcrossTtls(t *testing.T) { + topo := NewTopology("topo", nil, 128, 3, false) + rp, _ := super_block.NewReplicaPlacementFromString("000") + ttl, _ := needle.ReadTTL("3d") + + dn1 := NewDataNode("server1") + dn1.Ip = "10.0.0.1" + dn1.Port = 8080 + dn2 := NewDataNode("server2") + dn2.Ip = "10.0.0.2" + dn2.Port = 8080 + + v1 := storage.VolumeInfo{Id: 7, Collection: "a", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL} + v2 := storage.VolumeInfo{Id: 7, Collection: "a", ReplicaPlacement: rp, Ttl: ttl} + + dn1.UpdateVolumes([]storage.VolumeInfo{v1}) + dn2.UpdateVolumes([]storage.VolumeInfo{v2}) + topo.RegisterVolumeLayout(v1, dn1) + topo.RegisterVolumeLayout(v2, dn2) + topo.flagDuplicateVolumeId(v2, dn2) + + if !topo.isQuarantined(dn2, v2.Id) || !topo.isQuarantined(dn1, v1.Id) { + t.Fatal("same collection with a different ttl is still a reused volume id") + } +} diff --git a/weed/topology/topology.go b/weed/topology/topology.go index 6933981800..3c68d3586b 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" @@ -99,6 +100,55 @@ type Topology struct { reportedTopologyIdsLock sync.Mutex pendingReportedTopologyIds map[string]struct{} + + // quarantinedVolumeIds are (node, vid) pairs force-read-only because the + // 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) error +} + +type quarantinedVolume struct { + dnId NodeId + 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 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 v, ok := t.quarantinedVolumeIds.Load(q); ok { + state := v.(quarantinedVolumeState) + if !state.enforced { + state.reported = vi.ReadOnly + } + 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 } func NewTopology(id string, seq sequence.Sequencer, volumeSizeLimit uint64, pulse int, replicationAsMin bool) *Topology { @@ -591,9 +641,15 @@ func (t *Topology) DeleteCollection(collectionName string) { if !found { return } + var vids []needle.VolumeId for _, vl := range collection.(*Collection).GetAllVolumeLayouts() { + vids = append(vids, vl.ListVids()...) vl.releaseLookupOwnership() } + // a vid colliding with a copy in this collection is no longer a duplicate + for _, vid := range vids { + t.releaseResolvedQuarantine(vid) + } } func (t *Topology) DeleteLayout(collectionName string, rp *super_block.ReplicaPlacement, ttl *needle.TTL, diskType types.DiskType) { @@ -612,15 +668,172 @@ 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 } // Dropped with its collection; the next lookup creates a fresh one. } } +// flagDuplicateVolumeId reports and quarantines v.Id when another data node +// already serves it under a different layout key. It runs after registration +// so a legitimate same-node move (disk type or collection change) settles +// first and does not trip it. Every copy across the conflicting layouts is +// 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) { + 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() + 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(q, quarantinedVolumeState{reported: reported}) + if vl != nil { + vl.SetVolumeQuarantined(n, v.Id) + } + 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 + t.quarantinedVolumeIds.Store(q, state) + } + } + } + quarantine(dn, t.FindVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, types.ToDiskType(v.DiskType))) + for _, h := range hits { + for _, n := range h.nodes { + quarantine(n, h.vl) + } + } +} + +// releaseResolvedQuarantine lifts read-only marks once the conflicting copy +// is gone, so a finished move does not leave the surviving copy quarantined. +func (t *Topology) releaseResolvedQuarantine(vid needle.VolumeId) { + t.quarantinedVolumeIds.Range(func(key, val any) bool { + q := key.(quarantinedVolume) + if q.vid != vid { + return true + } + dn, vl, vi := t.findQuarantinedVolume(q.dnId, vid) + if dn == nil { + t.quarantinedVolumeIds.Delete(key) + return true + } + if hits := t.findDuplicateVolumeLayouts(vi, dn); len(hits) > 0 { + return true + } + state, _ := val.(quarantinedVolumeState) + if state.reported { + dn.SetVolumeReadOnly(vid, true) + 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 + }) +} + +func (t *Topology) findQuarantinedVolume(dnId NodeId, vid needle.VolumeId) (*DataNode, *VolumeLayout, storage.VolumeInfo) { + for _, item := range t.collectionMap.Items() { + for _, vl := range item.(*Collection).GetAllVolumeLayouts() { + for _, n := range vl.Lookup(vid) { + if n.Id() == dnId { + vi, _ := n.GetVolumesById(vid) + return n, vl, vi + } + } + } + } + return nil, nil, storage.VolumeInfo{} +} + +// 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) +} + +func ttlEqual(a, b *needle.TTL) bool { + if a == nil || b == nil { + return a == b + } + return a.ToUint32() == b.ToUint32() +} + + +// 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 + } + diskType := types.ToDiskType(v.DiskType) + var hits []duplicateHit + for _, item := range t.collectionMap.Items() { + c := item.(*Collection) + for _, vl := range c.GetAllVolumeLayouts() { + if c.Name == v.Collection && vl.diskType == diskType && ttlEqual(vl.ttl, v.Ttl) { + // The reporter's own layout, or an rp-only difference: + // a replication change rolls through the nodes one at a + // time, so the same volume legitimately reports the old + // and new rp while it settles. + continue + } + var others []*DataNode + for _, n := range vl.Lookup(v.Id) { + if n != dn { + others = append(others, n) + } + } + if len(others) > 0 { + hits = append(hits, duplicateHit{collection: c.Name, vl: vl, nodes: others}) + } + } + } + return hits +} + func (t *Topology) UnRegisterVolumeLayout(v storage.VolumeInfo, dn *DataNode) { glog.Infof("removing volume info: %+v from %v", v, dn.id) + defer t.releaseResolvedQuarantine(v.Id) diskType := types.ToDiskType(v.DiskType) volumeLayout := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType) volumeLayout.UnRegisterVolume(&v, dn) @@ -745,6 +958,9 @@ func (t *Topology) SyncDataNodeRegistration(volumes []*master_pb.VolumeInformati glog.V(0).Infof("Fail to convert joined volume information: %v", err) } } + for i := range volumeInfos { + volumeInfos[i] = t.enforceQuarantine(dn, volumeInfos[i]) + } // find out the delta volumes newVolumes, deletedVolumes, changedVolumes = dn.UpdateVolumes(volumeInfos) for _, v := range newVolumes { @@ -757,7 +973,7 @@ func (t *Topology) SyncDataNodeRegistration(volumes []*master_pb.VolumeInformati if v.ReplicaPlacement == nil { continue } - t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, types.ToDiskType(v.DiskType)).SetReplicaReadOnlyFlag(dn, v.Id, v.ReadOnly) + t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, types.ToDiskType(v.DiskType)).SetReplicaReadOnlyFlag(dn, v.Id, v.ReadOnly || t.isQuarantined(dn, v.Id)) } // Update effective sizes for all reported volumes (decay pending estimates). // If decay brings a volume eagerly removed by RecordAssign back under the @@ -781,6 +997,7 @@ func (t *Topology) SyncDataNodeRegistration(volumes []*master_pb.VolumeInformati // one, which the calls below must use too. vl = t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType) } + go t.flagDuplicateVolumeId(v, dn) // Volumes new to the disk map were registered above, so reaching // here means only the lookup index had lost it. Clients were told // it went when the node dropped out, so the repair has to tell them @@ -804,7 +1021,7 @@ func (t *Topology) IncrementalSyncDataNodeRegistration(newVolumes, deletedVolume glog.V(0).Infof("NewVolumeInfoFromShort %v: %v", v, err) continue } - newVis = append(newVis, vi) + newVis = append(newVis, t.enforceQuarantine(dn, vi)) } for _, v := range deletedVolumes { vi, err := storage.NewVolumeInfoFromShort(v) @@ -865,7 +1082,7 @@ func (t *Topology) ApplyVolumeChanges(changed []*master_pb.VolumeInformationMess glog.V(0).Infof("Fail to convert changed volume information: %v", err) continue } - volumeInfos = append(volumeInfos, vi) + volumeInfos = append(volumeInfos, t.enforceQuarantine(dn, vi)) } for _, vi := range volumeInfos { @@ -885,11 +1102,14 @@ func (t *Topology) ApplyVolumeChanges(changed []*master_pb.VolumeInformationMess // Dropped with its collection; the next lookup creates a fresh one. vl = t.GetVolumeLayout(vi.Collection, vi.ReplicaPlacement, vi.Ttl, types.ToDiskType(vi.DiskType)) } + if becameServable { + go t.flagDuplicateVolumeId(vi, dn) + } if isNew || becameServable || tierTransition || isChanged { newVolumes = append(newVolumes, vi) } if isChanged { - vl.SetReplicaReadOnlyFlag(dn, vi.Id, vi.ReadOnly) + vl.SetReplicaReadOnlyFlag(dn, vi.Id, vi.ReadOnly || t.isQuarantined(dn, vi.Id)) } vl.UpdateOversizedState(&vi, dn) if vl.UpdateVolumeSize(vi.Id, vi.Size, vi.CompactRevision, true) { diff --git a/weed/topology/topology_event_handling.go b/weed/topology/topology_event_handling.go index a2825345b2..a0df67e221 100644 --- a/weed/topology/topology_event_handling.go +++ b/weed/topology/topology_event_handling.go @@ -118,6 +118,8 @@ func (t *Topology) UnRegisterDataNode(dn *DataNode) { diskType := types.ToDiskType(v.DiskType) vl := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType) vl.SetVolumeUnavailable(dn, v.Id) + // this node may have been the conflicting copy pinning a quarantine + defer t.releaseResolvedQuarantine(v.Id) } // unregister ec shards when volume server disconnected diff --git a/weed/topology/volume_layout.go b/weed/topology/volume_layout.go index 1213b30fca..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 @@ -586,12 +589,21 @@ func (vl *VolumeLayout) isEmpty() bool { return len(vl.vid2location) == 0 } +func (vl *VolumeLayout) ListVids() (vids []needle.VolumeId) { + vl.accessLock.RLock() + defer vl.accessLock.RUnlock() + for vid := range vl.vid2location { + vids = append(vids, vid) + } + return +} + func (vl *VolumeLayout) Lookup(vid needle.VolumeId) []*DataNode { vl.accessLock.RLock() defer vl.accessLock.RUnlock() if location := vl.vid2location[vid]; location != nil { - return location.list + return append([]*DataNode(nil), location.list...) } return nil } @@ -1021,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