diff --git a/weed/topology/duplicate_volume_id_test.go b/weed/topology/duplicate_volume_id_test.go index 3c0bf5cfc1..949c39dddd 100644 --- a/weed/topology/duplicate_volume_id_test.go +++ b/weed/topology/duplicate_volume_id_test.go @@ -3,6 +3,7 @@ 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" @@ -79,3 +80,32 @@ func TestSameNodeVolumeMoveIsNotDuplicate(t *testing.T) { t.Fatal("a same-node disk move is not a duplicate") } } + +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 read-only + 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") + } +} diff --git a/weed/topology/topology.go b/weed/topology/topology.go index 3a9ba38bfa..4bf5fcc43e 100644 --- a/weed/topology/topology.go +++ b/weed/topology/topology.go @@ -99,6 +99,30 @@ 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 +} + +type quarantinedVolume struct { + dnId NodeId + vid needle.VolumeId +} + +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. +func (t *Topology) enforceQuarantine(dn *DataNode, vi storage.VolumeInfo) storage.VolumeInfo { + if t.isQuarantined(dn, vi.Id) { + vi.ReadOnly = true + } + return vi } func NewTopology(id string, seq sequence.Sequencer, volumeSizeLimit uint64, pulse int, replicationAsMin bool) *Topology { @@ -631,6 +655,7 @@ func (t *Topology) flagDuplicateVolumeId(v storage.VolumeInfo, dn *DataNode) { } 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 %s's copy read-only", v.Id, dn.Id(), v.Collection, nodes, dupCollection, dn.Id()) + t.quarantinedVolumeIds.Store(quarantinedVolume{dn.Id(), v.Id}, struct{}{}) if vl := t.FindVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, types.ToDiskType(v.DiskType)); vl != nil { vl.SetVolumeReadOnly(dn, v.Id) } @@ -671,6 +696,7 @@ func (t *Topology) findDuplicateVolumeLayout(v storage.VolumeInfo, dn *DataNode) 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}) diskType := types.ToDiskType(v.DiskType) volumeLayout := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType) volumeLayout.UnRegisterVolume(&v, dn) @@ -795,6 +821,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 { @@ -807,7 +836,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 @@ -831,6 +860,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 @@ -854,7 +884,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) @@ -915,7 +945,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 { @@ -935,11 +965,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) {