diff --git a/weed/storage/types/volume_disk_type.go b/weed/storage/types/volume_disk_type.go index 04c519070..3a5e14055 100644 --- a/weed/storage/types/volume_disk_type.go +++ b/weed/storage/types/volume_disk_type.go @@ -4,6 +4,12 @@ import ( "strings" ) +// DiskId identifies a single physical disk on a volume server, matching the +// uint32 index into Store.Locations that the volume server assigns per mount +// point. It is carried on the wire as uint32 in VolumeEcShardInformationMessage +// and VolumeInformationMessage. +type DiskId uint32 + type DiskType string const ( diff --git a/weed/topology/data_node_ec.go b/weed/topology/data_node_ec.go index ee428fd7c..cbda6bfec 100644 --- a/weed/topology/data_node_ec.go +++ b/weed/topology/data_node_ec.go @@ -16,14 +16,44 @@ func (dn *DataNode) GetEcShards() (ret []*erasure_coding.EcVolumeInfo) { return ret } +// ecShardKey identifies a per-physical-disk EC shard entry on a DataNode. +// A single volume's EC shards can live on multiple physical disks of one +// DataNode (e.g. a 10+4 volume spread across 4 mount points), so entries +// must be tracked per (volume, physical disk) rather than per volume alone. +// See issue #9212. +type ecShardKey struct { + vid needle.VolumeId + diskId types.DiskId +} + func (dn *DataNode) UpdateEcShards(actualShards []*erasure_coding.EcVolumeInfo) (newShards, deletedShards []*erasure_coding.EcVolumeInfo) { - // prepare the new ec shard map - actualEcShardMap := make(map[needle.VolumeId]*erasure_coding.EcVolumeInfo) + // Aggregate incoming messages by (volume, physical disk). Duplicates for + // the same (vid, diskId) are merged; entries for the same vid on + // different disks stay separate so per-physical-disk attribution is + // preserved all the way to DiskInfo.EcShardInfos — which the admin + // shell's ec.balance / ec.rebuild planners read via eci.DiskId. + actualByKey := make(map[ecShardKey]*erasure_coding.EcVolumeInfo, len(actualShards)) for _, ecShards := range actualShards { - actualEcShardMap[ecShards.VolumeId] = ecShards + key := ecShardKey{vid: ecShards.VolumeId, diskId: types.DiskId(ecShards.DiskId)} + if existing, ok := actualByKey[key]; ok { + existing.ShardsInfo.Add(ecShards.ShardsInfo) + continue + } + // Clone so subsequent merges for the same key don't mutate the + // caller's EcVolumeInfo, and so the diff below is stable. + clone := *ecShards + clone.ShardsInfo = ecShards.ShardsInfo.Copy() + actualByKey[key] = &clone } - existingEcShards := dn.GetEcShards() + // Hold dn.Lock for the full read-diff-write cycle so concurrent heartbeats, + // getOrCreateDisk calls, and UpAdjustDiskUsageDelta updates on this data + // node are serialized with us — matching UpdateVolumes' locking model. + // Internal helpers below assume the caller holds dn.Lock. + dn.Lock() + defer dn.Unlock() + + existingEcShards := dn.getEcShardsLocked() // find out the newShards and deletedShards for _, ecShards := range existingEcShards { @@ -31,8 +61,8 @@ func (dn *DataNode) UpdateEcShards(actualShards []*erasure_coding.EcVolumeInfo) var newShardCount, deletedShardCount int disk := dn.getOrCreateDisk(ecShards.DiskType) - vid := ecShards.VolumeId - if actualEcShards, ok := actualEcShardMap[vid]; !ok { + key := ecShardKey{vid: ecShards.VolumeId, diskId: types.DiskId(ecShards.DiskId)} + if actualEcShards, ok := actualByKey[key]; !ok { // dn registered ec shards not found in the new set of ec shards deletedShards = append(deletedShards, ecShards) deletedShardCount += ecShards.ShardsInfo.Count() @@ -58,8 +88,12 @@ func (dn *DataNode) UpdateEcShards(actualShards []*erasure_coding.EcVolumeInfo) } - for _, ecShards := range actualShards { - if dn.HasEcShards(ecShards.VolumeId) { + existingKeys := make(map[ecShardKey]struct{}, len(existingEcShards)) + for _, ev := range existingEcShards { + existingKeys[ecShardKey{vid: ev.VolumeId, diskId: types.DiskId(ev.DiskId)}] = struct{}{} + } + for key, ecShards := range actualByKey { + if _, found := existingKeys[key]; found { continue } @@ -73,36 +107,60 @@ func (dn *DataNode) UpdateEcShards(actualShards []*erasure_coding.EcVolumeInfo) if len(newShards) > 0 || len(deletedShards) > 0 { // if changed, set to the new ec shard map - dn.doUpdateEcShards(actualShards) + dn.doUpdateEcShardsLocked(actualByKey) } return } +// getEcShardsLocked returns the flat list of per-physical-disk EC shard +// entries across all children. Caller MUST hold dn.Lock (or RLock); each +// disk's ecShardsLock is taken internally via Disk.GetEcShards. +func (dn *DataNode) getEcShardsLocked() (ret []*erasure_coding.EcVolumeInfo) { + for _, c := range dn.children { + disk := c.(*Disk) + ret = append(ret, disk.GetEcShards()...) + } + return ret +} + func (dn *DataNode) HasEcShards(volumeId needle.VolumeId) (found bool) { dn.RLock() defer dn.RUnlock() for _, c := range dn.children { disk := c.(*Disk) - _, found = disk.ecShards[volumeId] - if found { - return + disk.ecShardsLock.RLock() + byDisk, ok := disk.ecShards[volumeId] + has := ok && len(byDisk) > 0 + disk.ecShardsLock.RUnlock() + if has { + return true } } return } -func (dn *DataNode) doUpdateEcShards(actualShards []*erasure_coding.EcVolumeInfo) { - dn.Lock() +// doUpdateEcShardsLocked rewrites disk.ecShards from actualByKey. Caller +// MUST hold dn.Lock; each disk's ecShardsLock is taken internally around +// the read-modify-write of its ecShards map. +func (dn *DataNode) doUpdateEcShardsLocked(actualByKey map[ecShardKey]*erasure_coding.EcVolumeInfo) { for _, c := range dn.children { disk := c.(*Disk) - disk.ecShards = make(map[needle.VolumeId]*erasure_coding.EcVolumeInfo) + disk.ecShardsLock.Lock() + disk.ecShards = make(map[needle.VolumeId]map[types.DiskId]*erasure_coding.EcVolumeInfo) + disk.ecShardsLock.Unlock() } - for _, shard := range actualShards { + for _, shard := range actualByKey { disk := dn.getOrCreateDisk(shard.DiskType) - disk.ecShards[shard.VolumeId] = shard + disk.ecShardsLock.Lock() + byDisk, ok := disk.ecShards[shard.VolumeId] + if !ok { + byDisk = make(map[types.DiskId]*erasure_coding.EcVolumeInfo, 1) + disk.ecShards[shard.VolumeId] = byDisk + } + byDisk[types.DiskId(shard.DiskId)] = shard + disk.ecShardsLock.Unlock() } - dn.Unlock() } func (dn *DataNode) DeltaUpdateEcShards(newShards, deletedShards []*erasure_coding.EcVolumeInfo) { diff --git a/weed/topology/data_node_ec_multi_disk_test.go b/weed/topology/data_node_ec_multi_disk_test.go new file mode 100644 index 000000000..698a4d114 --- /dev/null +++ b/weed/topology/data_node_ec_multi_disk_test.go @@ -0,0 +1,177 @@ +package topology + +import ( + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/sequence" + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" +) + +// TestEcShardsAcrossMultipleDisksOnSameNode reproduces issue #9212. +// When a volume server reports EC shards of the same volume spread across +// multiple physical disks on the same node, the master must register ALL +// shards, not only a subset. +func TestEcShardsAcrossMultipleDisksOnSameNode(t *testing.T) { + topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false) + dc := topo.GetOrCreateDataCenter("dc1") + rack := dc.GetOrCreateRack("rack1") + maxVolumeCounts := map[string]uint32{"": 100} + dn := rack.GetOrCreateDataNode("127.0.0.1", 34534, 0, "127.0.0.1", "", maxVolumeCounts) + + const vid = uint32(15) + const collection = "grafana-loki" + const diskType = "" // HDD / default type — all 4 disks share this type + + // Volume 15 has its 14 EC shards spread across 3 physical disks on this node. + // Mirrors the "volume-2" row from issue #9212: + // /data1 (diskId 0): ec02, ec06, ec10 + // /data2 (diskId 1): ec01, ec04, ec09 + // /data4 (diskId 3): ec08, ec12 + disk0 := buildEcShardMessage(vid, collection, diskType, 0, []erasure_coding.ShardId{2, 6, 10}) + disk1 := buildEcShardMessage(vid, collection, diskType, 1, []erasure_coding.ShardId{1, 4, 9}) + disk3 := buildEcShardMessage(vid, collection, diskType, 3, []erasure_coding.ShardId{8, 12}) + + msgs := []*master_pb.VolumeEcShardInformationMessage{disk0, disk1, disk3} + topo.SyncDataNodeEcShards(msgs, dn) + + locs, ok := topo.LookupEcShards(needle.VolumeId(vid)) + if !ok { + t.Fatalf("volume %d: no ec shard locations registered at all", vid) + } + + // All 8 shards should be visible to the master: 1,2,4,6,8,9,10,12. + wantShards := []erasure_coding.ShardId{1, 2, 4, 6, 8, 9, 10, 12} + var gotShards []erasure_coding.ShardId + for shardId, dataNodes := range locs.Locations { + if len(dataNodes) > 0 { + gotShards = append(gotShards, erasure_coding.ShardId(shardId)) + } + } + + if len(gotShards) != len(wantShards) { + t.Errorf("volume %d: topology.LookupEcShards sees %d shards %v, want %d shards %v", + vid, len(gotShards), gotShards, len(wantShards), wantShards) + } + + for _, want := range wantShards { + if len(locs.Locations[want]) == 0 { + t.Errorf("volume %d: shard %d missing from topology (bug #9212)", vid, want) + } + } + + // The DataNode's own view (what drives volume.list output, admin UI, and + // ec.rebuild dry-run diagnostics) must also see all shards across all disks. + dnShards := dn.GetEcShards() + dnShardCount := 0 + dnShardBitmap := erasure_coding.ShardBits(0) + for _, ev := range dnShards { + if ev.VolumeId == needle.VolumeId(vid) { + dnShardCount += ev.ShardsInfo.Count() + dnShardBitmap |= erasure_coding.ShardBits(ev.ShardsInfo.Bitmap()) + } + } + if dnShardCount != len(wantShards) { + t.Errorf("volume %d: DataNode.GetEcShards reports %d shards (bitmap=0b%b), want %d shards %v (bug #9212)", + vid, dnShardCount, dnShardBitmap, len(wantShards), wantShards) + } + + // Per-physical-disk attribution must survive all the way to the protobuf + // DiskInfo.EcShardInfos consumed by ec.balance / ec.rebuild planners. The + // admin shell groups shards by eci.DiskId, so each physical disk needs a + // separate message with its own shard subset. + wantPerDisk := map[uint32][]erasure_coding.ShardId{ + 0: {2, 6, 10}, + 1: {1, 4, 9}, + 3: {8, 12}, + } + dnInfo := dn.ToDataNodeInfo() + gotPerDisk := map[uint32][]erasure_coding.ShardId{} + for _, diskInfo := range dnInfo.DiskInfos { + for _, eci := range diskInfo.EcShardInfos { + if eci.Id != vid { + continue + } + si := erasure_coding.ShardsInfoFromVolumeEcShardInformationMessage(eci) + gotPerDisk[eci.DiskId] = append(gotPerDisk[eci.DiskId], si.Ids()...) + } + } + for diskId, want := range wantPerDisk { + got := gotPerDisk[diskId] + if !shardSetEqual(got, want) { + t.Errorf("volume %d diskId %d: DiskInfo.EcShardInfos report shards %v, want %v (per-physical-disk attribution lost)", + vid, diskId, got, want) + } + } + for diskId := range gotPerDisk { + if _, ok := wantPerDisk[diskId]; !ok { + t.Errorf("volume %d: unexpected diskId %d in DiskInfo.EcShardInfos, shards=%v", + vid, diskId, gotPerDisk[diskId]) + } + } +} + +func shardSetEqual(a, b []erasure_coding.ShardId) bool { + if len(a) != len(b) { + return false + } + seen := make(map[erasure_coding.ShardId]int, len(a)) + for _, id := range a { + seen[id]++ + } + for _, id := range b { + seen[id]-- + if seen[id] < 0 { + return false + } + } + return true +} + +// TestEcShardsAfterRestartHeartbeat simulates the exact issue #9212 flow: +// the volume server starts up, loads shards from each disk, and sends a +// single full-sync heartbeat containing one VolumeEcShardInformationMessage +// per (disk, volume). The master must end up with all shards visible per +// DataNode, not just the subset belonging to whichever disk happened to be +// iterated last. +func TestEcShardsAfterRestartHeartbeat(t *testing.T) { + topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false) + dc := topo.GetOrCreateDataCenter("dc1") + rack := dc.GetOrCreateRack("rack1") + dn := rack.GetOrCreateDataNode("127.0.0.1", 34534, 0, "127.0.0.1", "", map[string]uint32{"": 100}) + + // Matches "volume-1" from the bug report: ec13 on /data1, ec03 on /data3. + msgs := []*master_pb.VolumeEcShardInformationMessage{ + buildEcShardMessage(15, "grafana-loki", "", 0, []erasure_coding.ShardId{13}), + buildEcShardMessage(15, "grafana-loki", "", 2, []erasure_coding.ShardId{3}), + } + topo.SyncDataNodeEcShards(msgs, dn) + + dnShards := dn.GetEcShards() + var combined erasure_coding.ShardBits + for _, ev := range dnShards { + if ev.VolumeId == 15 { + combined |= erasure_coding.ShardBits(ev.ShardsInfo.Bitmap()) + } + } + if combined.Count() != 2 { + t.Errorf("volume 15: DataNode sees %d shards (bitmap=0b%b) after full sync of 2 disks; want both shards 3 and 13 visible (bug #9212)", + combined.Count(), combined) + } +} + +func buildEcShardMessage(vid uint32, collection, diskType string, diskId uint32, shardIds []erasure_coding.ShardId) *master_pb.VolumeEcShardInformationMessage { + si := erasure_coding.NewShardsInfo() + for _, sid := range shardIds { + si.Set(erasure_coding.NewShardInfo(sid, erasure_coding.ShardSize(1024))) + } + return &master_pb.VolumeEcShardInformationMessage{ + Id: vid, + Collection: collection, + DiskType: diskType, + DiskId: diskId, + EcIndexBits: si.Bitmap(), + ShardSizes: si.SizesInt64(), + } +} diff --git a/weed/topology/disk.go b/weed/topology/disk.go index b5ce0b60c..351bb147f 100644 --- a/weed/topology/disk.go +++ b/weed/topology/disk.go @@ -18,8 +18,13 @@ import ( type Disk struct { NodeImpl - volumes map[needle.VolumeId]storage.VolumeInfo - ecShards map[needle.VolumeId]*erasure_coding.EcVolumeInfo + volumes map[needle.VolumeId]storage.VolumeInfo + // ecShards is nested so the same volume can retain separate entries per + // physical disk id. A single topology Disk represents one DiskType on a + // DataNode and may front multiple physical disks of that type, so EC + // shards of one volume can legitimately live on several of them. The + // outer key is the volume id; the inner key is the physical disk id. + ecShards map[needle.VolumeId]map[types.DiskId]*erasure_coding.EcVolumeInfo ecShardsLock sync.RWMutex } @@ -35,7 +40,7 @@ func NewDisk(diskType string) *Disk { s.nodeType = "Disk" s.diskUsages = newDiskUsages() s.volumes = make(map[needle.VolumeId]storage.VolumeInfo, 2) - s.ecShards = make(map[needle.VolumeId]*erasure_coding.EcVolumeInfo, 2) + s.ecShards = make(map[needle.VolumeId]map[types.DiskId]*erasure_coding.EcVolumeInfo, 2) s.NodeImpl.value = s return s } diff --git a/weed/topology/disk_ec.go b/weed/topology/disk_ec.go index 6ad3ac958..d92bedf0d 100644 --- a/weed/topology/disk_ec.go +++ b/weed/topology/disk_ec.go @@ -9,9 +9,17 @@ import ( func (d *Disk) GetEcShards() (ret []*erasure_coding.EcVolumeInfo) { d.ecShardsLock.RLock() defer d.ecShardsLock.RUnlock() - ret = make([]*erasure_coding.EcVolumeInfo, 0, len(d.ecShards)) - for _, ecVolumeInfo := range d.ecShards { - ret = append(ret, ecVolumeInfo) + // Total entries = sum of per-volume per-disk entries, which typically + // exceeds the number of unique volumes once shards span multiple disks. + total := 0 + for _, byDisk := range d.ecShards { + total += len(byDisk) + } + ret = make([]*erasure_coding.EcVolumeInfo, 0, total) + for _, byDisk := range d.ecShards { + for _, ecVolumeInfo := range byDisk { + ret = append(ret, ecVolumeInfo) + } } return ret } @@ -20,9 +28,16 @@ func (d *Disk) AddOrUpdateEcShard(s *erasure_coding.EcVolumeInfo) { d.ecShardsLock.Lock() defer d.ecShardsLock.Unlock() + byDisk, ok := d.ecShards[s.VolumeId] + if !ok { + byDisk = make(map[types.DiskId]*erasure_coding.EcVolumeInfo, 1) + d.ecShards[s.VolumeId] = byDisk + } + + diskId := types.DiskId(s.DiskId) delta := 0 - if existing, ok := d.ecShards[s.VolumeId]; !ok { - d.ecShards[s.VolumeId] = s + if existing, ok := byDisk[diskId]; !ok { + byDisk[diskId] = s delta = s.ShardsInfo.Count() } else { oldCount := existing.ShardsInfo.Count() @@ -41,21 +56,30 @@ func (d *Disk) DeleteEcShard(s *erasure_coding.EcVolumeInfo) { d.ecShardsLock.Lock() defer d.ecShardsLock.Unlock() - if existing, ok := d.ecShards[s.VolumeId]; ok { - oldCount := existing.ShardsInfo.Count() - existing.ShardsInfo.Subtract(s.ShardsInfo) - delta := existing.ShardsInfo.Count() - oldCount + byDisk, ok := d.ecShards[s.VolumeId] + if !ok { + return + } + diskId := types.DiskId(s.DiskId) + existing, ok := byDisk[diskId] + if !ok { + return + } + oldCount := existing.ShardsInfo.Count() + existing.ShardsInfo.Subtract(s.ShardsInfo) + delta := existing.ShardsInfo.Count() - oldCount - if delta != 0 { - d.UpAdjustDiskUsageDelta(types.ToDiskType(string(d.Id())), &DiskUsageCounts{ - ecShardCount: int64(delta), - }) - } - if existing.ShardsInfo.Count() == 0 { + if delta != 0 { + d.UpAdjustDiskUsageDelta(types.ToDiskType(string(d.Id())), &DiskUsageCounts{ + ecShardCount: int64(delta), + }) + } + if existing.ShardsInfo.Count() == 0 { + delete(byDisk, diskId) + if len(byDisk) == 0 { delete(d.ecShards, s.VolumeId) } } - } func (d *Disk) HasVolumesById(id needle.VolumeId) (hasVolumeId bool) { @@ -73,8 +97,7 @@ func (d *Disk) HasVolumesById(id needle.VolumeId) (hasVolumeId bool) { // check whether ec shards has this volume id d.ecShardsLock.RLock() - _, ok = d.ecShards[id] - if ok { + if byDisk, ok := d.ecShards[id]; ok && len(byDisk) > 0 { hasVolumeId = true } d.ecShardsLock.RUnlock()