topology: follow a volume that moved between a server's disks (#10628)

* topology: follow a volume that moved between a server's disks

The heartbeat diff asked only whether a volume id was reported anywhere on the
node, so a volume that moved to a disk of another type stayed on the disk it
left as well. The master then held two copies of it forever: the volume count
was overstated, and GetVolumesById returned whichever disk the map iterated
first, so lookups could hand back the disk the volume had already left.

Track which disk types the heartbeat named each volume on, and treat a volume
named on another disk as absent from this one. Disk types are interned to an
index because a server reports a handful of them across hundreds of thousands
of volumes.

A volume named on two disks at once is a stale twin rather than a move, and is
still kept on both -- dropping one would tell the master a replica vanished.
Only a volume named twice on one disk type is unrepresentable, so that is now
what marks the node, rather than any repeat of an id.

* master: do not tell clients a moved volume left the node

A volume moved between a node's disks is removed from one and added to the
other, so it lands in both lists of the same heartbeat. Clients apply additions
before deletions, so the removal wins and they end up with no location for a
volume that never went anywhere.

Skip removals for volumes the node still holds, as the ec shard paths already
do, and update the topology before judging the delta removals so an unmount
that really did happen is still reported.

* trim the comments on this change to the parts that are not evident

* master: judge a volume removal on normal replicas alone

HasVolumesById answers for ec shards as well, so a replica encoded into ec
shards looked like it was still on the node and clients were never told the
normal location had gone. They hold normal and ec locations separately and
prefer the normal one from the same generation, so that location would have
gone on shadowing the shards.

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))
 			}
 		}

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")
+	}
+}

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 {

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()

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) }

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())
+	}
+}