diff --git a/weed/server/master_grpc_server.go b/weed/server/master_grpc_server.go index 5ee4d49fc..37b5e468c 100644 --- a/weed/server/master_grpc_server.go +++ b/weed/server/master_grpc_server.go @@ -28,6 +28,16 @@ import ( "github.com/seaweedfs/seaweedfs/weed/topology" ) +// A volume moved between the node's disks appears in both lists, and clients +// apply additions before deletions, so passing the removal on would drop a +// location that is still good. Not HasVolumesById, which answers for ec shards +// too: clients hold those separately and prefer the normal location, so a +// replica that became ec shards has to be reported gone. +func shouldBroadcastVolumeRemoval(dn *topology.DataNode, vid needle.VolumeId) bool { + _, err := dn.GetVolumesById(vid) + return err != nil +} + func (ms *MasterServer) RegisterUuids(heartbeat *master_pb.Heartbeat) (duplicated_uuids []string, err error) { ms.Topo.UuidAccessLock.Lock() defer ms.Topo.UuidAccessLock.Unlock() @@ -207,15 +217,19 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ stats.MasterReceivedHeartbeatCounter.WithLabelValues("deletedVolumes").Inc() } if len(heartbeat.NewVolumes) > 0 || len(heartbeat.DeletedVolumes) > 0 { + // first, so the removals below see where the volumes ended up + ms.Topo.IncrementalSyncDataNodeRegistration(heartbeat.NewVolumes, heartbeat.DeletedVolumes, dn) + // process delta volume ids if exists for fast volume id updates for _, volInfo := range heartbeat.NewVolumes { message.NewVids = append(message.NewVids, volInfo.Id) } for _, volInfo := range heartbeat.DeletedVolumes { + if !shouldBroadcastVolumeRemoval(dn, needle.VolumeId(volInfo.Id)) { + continue + } message.DeletedVids = append(message.DeletedVids, volInfo.Id) } - // update master internal volume layouts - ms.Topo.IncrementalSyncDataNodeRegistration(heartbeat.NewVolumes, heartbeat.DeletedVolumes, dn) } if len(heartbeat.Volumes) > 0 || heartbeat.HasNoVolumes { @@ -234,6 +248,9 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ } for _, v := range deletedVolumes { glog.V(1).Infof("master see deleted volume %d from %s", uint32(v.Id), dn.Url()) + if !shouldBroadcastVolumeRemoval(dn, v.Id) { + continue + } message.DeletedVids = append(message.DeletedVids, uint32(v.Id)) } } diff --git a/weed/server/master_grpc_server_volume_move_test.go b/weed/server/master_grpc_server_volume_move_test.go new file mode 100644 index 000000000..46ae5bbbd --- /dev/null +++ b/weed/server/master_grpc_server_volume_move_test.go @@ -0,0 +1,98 @@ +package weed_server + +import ( + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/sequence" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/topology" +) + +func moveTestNode(t *testing.T) (*topology.Topology, *topology.DataNode) { + t.Helper() + topo := topology.NewTopology("test", sequence.NewMemorySequencer(), 32*1024*1024*1024, 5, false) + dn := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1"). + GetOrCreateDataNode("127.0.0.1", 8080, 18080, "", "", map[string]uint32{"": 100, "ssd": 100}) + return topo, dn +} + +func moveTestVolume(diskType string, diskId uint32) *master_pb.VolumeInformationMessage { + return &master_pb.VolumeInformationMessage{ + Id: 1, Size: 1024, Collection: "c", Version: 3, DiskType: diskType, DiskId: diskId, + } +} + +// Clients apply additions before deletions, so a move reported as both would +// leave them with no location for a volume that never went anywhere. +func TestVolumeMovedBetweenDisksIsNotBroadcastAsRemoved(t *testing.T) { + topo, dn := moveTestNode(t) + topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{moveTestVolume("", 0)}, dn) + + _, deleted := topo.SyncDataNodeRegistration( + []*master_pb.VolumeInformationMessage{moveTestVolume("ssd", 1)}, dn) + + if len(deleted) != 1 { + t.Fatalf("expected the move to remove the volume from the disk it left, got %d removals", len(deleted)) + } + if shouldBroadcastVolumeRemoval(dn, needle.VolumeId(1)) { + t.Error("clients would be told a volume left a node that still has it") + } +} + +func TestVolumeGoneFromTheNodeIsBroadcastAsRemoved(t *testing.T) { + topo, dn := moveTestNode(t) + topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{moveTestVolume("", 0)}, dn) + + if _, deleted := topo.SyncDataNodeRegistration(nil, dn); len(deleted) != 1 { + t.Fatalf("expected the volume to be removed, got %d removals", len(deleted)) + } + if !shouldBroadcastVolumeRemoval(dn, needle.VolumeId(1)) { + t.Error("clients were not told about a volume that really did leave the node") + } +} + +func TestVolumeRemountedOnAnotherDiskIsNotBroadcastAsRemoved(t *testing.T) { + topo, dn := moveTestNode(t) + topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{moveTestVolume("", 0)}, dn) + + topo.IncrementalSyncDataNodeRegistration( + []*master_pb.VolumeShortInformationMessage{{Id: 1, Collection: "c", DiskType: "ssd"}}, + []*master_pb.VolumeShortInformationMessage{{Id: 1, Collection: "c", DiskType: ""}}, + dn) + + if shouldBroadcastVolumeRemoval(dn, needle.VolumeId(1)) { + t.Error("clients would be told a volume left a node that still has it on another disk") + } +} + +// Fails if the topology update stops running before the removals are judged. +func TestVolumeUnmountedViaDeltaIsBroadcastAsRemoved(t *testing.T) { + topo, dn := moveTestNode(t) + topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{moveTestVolume("", 0)}, dn) + + topo.IncrementalSyncDataNodeRegistration(nil, + []*master_pb.VolumeShortInformationMessage{{Id: 1, Collection: "c", DiskType: ""}}, dn) + + if !shouldBroadcastVolumeRemoval(dn, needle.VolumeId(1)) { + t.Error("clients were not told about a volume that really was unmounted") + } +} + +// Clients track normal and ec locations separately, and prefer the normal one +// when both come from the same generation. A replica that became ec shards has +// genuinely left, so the removal must still go out. +func TestVolumeReplacedByEcShardsIsBroadcastAsRemoved(t *testing.T) { + topo, dn := moveTestNode(t) + topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{moveTestVolume("", 0)}, dn) + topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{ + {Id: 1, Collection: "c", EcIndexBits: 0x3fff}, + }, dn) + + if _, deleted := topo.SyncDataNodeRegistration(nil, dn); len(deleted) != 1 { + t.Fatalf("expected the normal volume to be removed, got %d removals", len(deleted)) + } + if !shouldBroadcastVolumeRemoval(dn, needle.VolumeId(1)) { + t.Error("clients kept a normal-volume location for a replica that became ec shards") + } +} diff --git a/weed/topology/data_node.go b/weed/topology/data_node.go index e796fcfb5..7226aac8d 100644 --- a/weed/topology/data_node.go +++ b/weed/topology/data_node.go @@ -83,9 +83,9 @@ func (dn *DataNode) doAddOrUpdateVolume(v storage.VolumeInfo) (isNew, isChanged // used in master to notify master clients of these changes. func (dn *DataNode) UpdateVolumes(actualVolumes []storage.VolumeInfo) (newVolumes, deletedVolumes, changedVolumes []storage.VolumeInfo) { - actualVolumeIds := make(map[needle.VolumeId]struct{}, len(actualVolumes)) + reported := newReportedVolumes(len(actualVolumes)) for _, v := range actualVolumes { - actualVolumeIds[v.Id] = struct{}{} + reported.add(v.Id, v.DiskType) } // A volume id mounted on two disks of one server -- a stale twin re-attached @@ -93,7 +93,7 @@ func (dn *DataNode) UpdateVolumes(actualVolumes []storage.VolumeInfo) (newVolume // id alone and keeps only the last copy. Its digest can then never equal the // server's however often the list is resent, so record it and let the // heartbeat fall back to the full list for this node. - dn.duplicateVolumeIds.Store(len(actualVolumeIds) < len(actualVolumes)) + dn.duplicateVolumeIds.Store(reported.duplicated) dn.Lock() defer dn.Unlock() @@ -101,7 +101,7 @@ func (dn *DataNode) UpdateVolumes(actualVolumes []storage.VolumeInfo) (newVolume keptCount := 0 for _, c := range dn.children { disk := c.(*Disk) - for _, v := range disk.RemoveVolumesNotIn(actualVolumeIds) { + for _, v := range disk.RemoveVolumesNotIn(reported) { glog.V(0).Infoln("Deleting volume id:", v.Id) deletedVolumes = append(deletedVolumes, v) @@ -120,7 +120,7 @@ func (dn *DataNode) UpdateVolumes(actualVolumes []storage.VolumeInfo) (newVolume // Everything still on the node is also in this heartbeat, so the remainder // is what the node is about to gain. A steady-state heartbeat gains nothing // and must not allocate here; a reconnecting server gains all of them. - if addedCount := len(actualVolumes) - keptCount; addedCount > 0 { + if addedCount := reported.count() - keptCount; addedCount > 0 { newVolumes = make([]storage.VolumeInfo, 0, addedCount) } for _, v := range actualVolumes { diff --git a/weed/topology/disk.go b/weed/topology/disk.go index 4cc939421..f6755f706 100644 --- a/weed/topology/disk.go +++ b/weed/topology/disk.go @@ -235,14 +235,15 @@ func (d *Disk) VolumeCount() int { return len(d.volumes) } -// RemoveVolumesNotIn drops the volumes whose ids are absent from keep and -// returns them, so a heartbeat can be diffed without first copying the whole -// volume map out. -func (d *Disk) RemoveVolumesNotIn(keep map[needle.VolumeId]struct{}) (removed []storage.VolumeInfo) { +// RemoveVolumesNotIn drops the volumes the heartbeat did not name on this disk +// and returns them, so a heartbeat can be diffed without copying the volume map +// out. A volume named on another disk has moved, and counts as absent here. +func (d *Disk) RemoveVolumesNotIn(reported *reportedVolumes) (removed []storage.VolumeInfo) { + diskTypeIndex := reported.diskTypeIndex(string(d.Id())) d.Lock() defer d.Unlock() for vid, v := range d.volumes { - if _, ok := keep[vid]; !ok { + if !reported.namedOn(vid, diskTypeIndex) { removed = append(removed, v) delete(d.volumes, vid) d.volumeDigest ^= v.ReportHash() diff --git a/weed/topology/reported_volumes.go b/weed/topology/reported_volumes.go new file mode 100644 index 000000000..53432cc1c --- /dev/null +++ b/weed/topology/reported_volumes.go @@ -0,0 +1,79 @@ +package topology + +import ( + "github.com/seaweedfs/seaweedfs/weed/storage/needle" +) + +// reportedVolumes records which disk types a heartbeat named each volume on, so +// a volume that moved between a server's disks is dropped from the one it left. +// Disk types are interned because a string header per volume would cost more +// than the map holding them. +type reportedVolumes struct { + diskTypes []string + byVolume map[needle.VolumeId]int32 + // extra covers volumes named on several disk types, a stale twin rather than + // a move. Nil otherwise. + extra map[needle.VolumeId][]int32 + // duplicated: named twice on one disk type, which the master stores once. + duplicated bool +} + +func newReportedVolumes(size int) *reportedVolumes { + return &reportedVolumes{byVolume: make(map[needle.VolumeId]int32, size)} +} + +func (r *reportedVolumes) add(vid needle.VolumeId, diskType string) { + index := r.internDiskType(diskType) + existing, seen := r.byVolume[vid] + if !seen { + r.byVolume[vid] = index + return + } + if existing == index || r.hasExtra(vid, index) { + r.duplicated = true + return + } + if r.extra == nil { + r.extra = make(map[needle.VolumeId][]int32) + } + r.extra[vid] = append(r.extra[vid], index) +} + +func (r *reportedVolumes) internDiskType(diskType string) int32 { + if index := r.diskTypeIndex(diskType); index >= 0 { + return index + } + r.diskTypes = append(r.diskTypes, diskType) + return int32(len(r.diskTypes) - 1) +} + +// diskTypeIndex returns -1 when the heartbeat named no volume on the type. +func (r *reportedVolumes) diskTypeIndex(diskType string) int32 { + for i, known := range r.diskTypes { + if known == diskType { + return int32(i) + } + } + return -1 +} + +func (r *reportedVolumes) namedOn(vid needle.VolumeId, index int32) bool { + if index < 0 { + return false + } + if stored, ok := r.byVolume[vid]; ok && stored == index { + return true + } + return r.hasExtra(vid, index) +} + +func (r *reportedVolumes) hasExtra(vid needle.VolumeId, index int32) bool { + for _, other := range r.extra[vid] { + if other == index { + return true + } + } + return false +} + +func (r *reportedVolumes) count() int { return len(r.byVolume) } diff --git a/weed/topology/volume_disk_move_test.go b/weed/topology/volume_disk_move_test.go new file mode 100644 index 000000000..24f6806a2 --- /dev/null +++ b/weed/topology/volume_disk_move_test.go @@ -0,0 +1,112 @@ +package topology + +import ( + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" +) + +func diskMoveNode(t *testing.T) (*Topology, *DataNode) { + t.Helper() + topo := NewTopology("move", nil, 32*1024*1024*1024, 5, false) + dn := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1"). + GetOrCreateDataNode("127.0.0.1", 8080, 18080, "", "", map[string]uint32{"": 100, "ssd": 100}) + return topo, dn +} + +func diskMoveVolume(id uint32, diskType string, diskId uint32) *master_pb.VolumeInformationMessage { + return &master_pb.VolumeInformationMessage{ + Id: id, Size: 1024, Collection: "c", Version: 3, DiskType: diskType, DiskId: diskId, + } +} + +func heldCopies(dn *DataNode) int { + total := 0 + for _, c := range dn.Children() { + total += c.(*Disk).VolumeCount() + } + return total +} + +func TestVolumeMovedBetweenDisks(t *testing.T) { + for _, tc := range []struct { + name string + from, to string + }{ + {"SameDiskType", "", ""}, + {"DifferentDiskType", "", "ssd"}, + {"BackAgain", "ssd", ""}, + } { + t.Run(tc.name, func(t *testing.T) { + topo, dn := diskMoveNode(t) + topo.SyncDataNodeRegistration( + []*master_pb.VolumeInformationMessage{diskMoveVolume(1, tc.from, 0)}, dn) + topo.SyncDataNodeRegistration( + []*master_pb.VolumeInformationMessage{diskMoveVolume(1, tc.to, 1)}, dn) + + if got := heldCopies(dn); got != 1 { + t.Errorf("master holds %d copies of a volume that moved, want 1", got) + } + stored, err := dn.GetVolumesById(needle.VolumeId(1)) + if err != nil { + t.Fatalf("moved volume is no longer on the node: %v", err) + } + if stored.DiskType != tc.to || stored.DiskId != 1 { + t.Errorf("master has the volume on disk %q/%d, server reports %q/1", + stored.DiskType, stored.DiskId, tc.to) + } + if !dn.HasConsistentVolumeIndex() { + t.Error("the move left the lookup index disagreeing with the disks") + } + }) + } +} + +// A stale twin is two reports, not a move: dropping one would tell the master a +// replica vanished. +func TestVolumeReportedOnTwoDiskTypesIsKept(t *testing.T) { + topo, dn := diskMoveNode(t) + topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{ + diskMoveVolume(1, "", 0), diskMoveVolume(1, "ssd", 1), + }, dn) + + if got := heldCopies(dn); got != 2 { + t.Errorf("master holds %d copies of a volume reported on two disks, want 2", got) + } + if dn.HasDuplicateVolumeIds() { + t.Error("a volume on two disk types is representable, so it should not disable digest comparison") + } + + topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{diskMoveVolume(1, "ssd", 1)}, dn) + if got := heldCopies(dn); got != 1 { + t.Errorf("master holds %d copies after the twin was unmounted, want 1", got) + } +} + +func TestVolumeReportedTwiceOnOneDiskTypeIsFlagged(t *testing.T) { + topo, dn := diskMoveNode(t) + first := diskMoveVolume(1, "ssd", 0) + second := diskMoveVolume(1, "ssd", 1) + topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{first, second}, dn) + + if !dn.HasDuplicateVolumeIds() { + t.Error("a volume reported twice on one disk type went unflagged") + } +} + +func TestVolumeDigestSurvivesADiskMove(t *testing.T) { + topo, dn := diskMoveNode(t) + topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{diskMoveVolume(1, "", 0)}, dn) + + moved := diskMoveVolume(1, "ssd", 1) + topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{moved}, dn) + + reference, referenceNode := diskMoveNode(t) + reference.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{moved}, referenceNode) + + if dn.VolumeDigest() != referenceNode.VolumeDigest() { + t.Errorf("after a disk move the digest is %d, a server holding only the moved volume reports %d", + dn.VolumeDigest(), referenceNode.VolumeDigest()) + } +}