From 9c8d3b6a81786c84c1133c7810636a3df538b095 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 23 Aug 2026 21:48:20 -0700 Subject: [PATCH] ec: refund the cleared leftover shards' slots in the encode source health check (#10903) * erasure_coding: one home for the shard-count to volume-slots conversion * ec: refund the cleared leftover shards' slots in the encode source health check --- weed/ec/ec_encode.go | 68 ++++++++++++++----- weed/ec/ec_encode_test.go | 55 +++++++++++++++ weed/storage/erasure_coding/ec_shards_info.go | 7 ++ weed/storage/store.go | 3 +- weed/topology/disk.go | 12 +--- weed/topology/node.go | 3 +- 6 files changed, 118 insertions(+), 30 deletions(-) diff --git a/weed/ec/ec_encode.go b/weed/ec/ec_encode.go index 5ce8f4a9f..d57ce4128 100644 --- a/weed/ec/ec_encode.go +++ b/weed/ec/ec_encode.go @@ -224,6 +224,55 @@ func volumeLocations(env *Env, volumeIds []needle.VolumeId) (map[needle.VolumeId return res, nil } +// collectSourceFreeVolumeCounts maps "volumeId-serverAddress" to the free +// volume slots on the disk holding that volume, for the source health check. +// Key by dn.Address so it matches wdclient.Location.Url: in deployments where +// dn.Id is a short name (e.g. a Kubernetes StatefulSet pod name) while +// dn.Address is a FQDN:port, keying by dn.Id would never match the location +// Url during the lookup. +// +// The topology snapshot predates clearPreexistingEcShards, so shards of the +// very volumes being encoded (leftovers of an interrupted run the sweep just +// removed) may still be charged against a disk's FreeVolumeCount. Refund their +// slots, or a retried encode fails the health check on capacity the sweep +// already freed. A node the sweep skipped as unreachable keeps its charge: its +// leftovers are still there. +func collectSourceFreeVolumeCounts(topologyInfo *master_pb.TopologyInfo, volumeIds []needle.VolumeId, sweepSkippedNodes map[pb.ServerAddress]struct{}) map[string]int { + encoding := make(map[uint32]bool, len(volumeIds)) + for _, vid := range volumeIds { + encoding[uint32(vid)] = true + } + freeVolumeCountMap := make(map[string]int) // key: volumeId-serverAddress + EachDataNode(topologyInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) { + addr := dn.Address + if addr == "" { + addr = dn.Id // older nodes use ip:port as id + } + _, skipped := sweepSkippedNodes[pb.NewServerAddressFromDataNode(dn)] + for _, diskInfo := range dn.DiskInfos { + free := diskInfo.FreeVolumeCount + if !skipped { + var total, cleared int64 + for _, ecInfo := range diskInfo.EcShardInfos { + n := int64(erasure_coding.GetShardCount(ecInfo)) + total += n + if encoding[ecInfo.Id] { + cleared += n + } + } + if cleared > 0 { + free += erasure_coding.VolumeSlots(total) - erasure_coding.VolumeSlots(total-cleared) + } + } + for _, v := range diskInfo.VolumeInfos { + key := fmt.Sprintf("%d-%s", v.Id, addr) + freeVolumeCountMap[key] = int(free) + } + } + }) + return freeVolumeCountMap +} + func doEcEncode(env *Env, writer io.Writer, volumeIdToCollection map[needle.VolumeId]string, volumeIds []needle.VolumeId, maxParallelization int, topologyInfo *master_pb.TopologyInfo) (skippedNodes map[pb.ServerAddress]struct{}, err error) { if !env.isLocked() { return nil, fmt.Errorf("lock is lost") @@ -243,24 +292,7 @@ func doEcEncode(env *Env, writer io.Writer, volumeIdToCollection map[needle.Volu return nil, fmt.Errorf("clear pre-existing ec shards before encoding: %w", err) } - // Build a map of (volumeId, serverAddress) -> freeVolumeCount. - // Key by dn.Address so it matches wdclient.Location.Url. In deployments - // where dn.Id is a short name (e.g. Kubernetes StatefulSet pod name) - // while dn.Address is a FQDN:port, keying by dn.Id would never match the - // location Url during the health-check lookup below. - freeVolumeCountMap := make(map[string]int) // key: volumeId-serverAddress - EachDataNode(topologyInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) { - addr := dn.Address - if addr == "" { - addr = dn.Id // older nodes use ip:port as id - } - for _, diskInfo := range dn.DiskInfos { - for _, v := range diskInfo.VolumeInfos { - key := fmt.Sprintf("%d-%s", v.Id, addr) - freeVolumeCountMap[key] = int(diskInfo.FreeVolumeCount) - } - } - }) + freeVolumeCountMap := collectSourceFreeVolumeCounts(topologyInfo, volumeIds, skippedNodes) // Filter replicas by free capacity BEFORE marking volumes readonly so that // a failed health check does not strand volumes in readonly state. diff --git a/weed/ec/ec_encode_test.go b/weed/ec/ec_encode_test.go index d57408f5d..9f1796d43 100644 --- a/weed/ec/ec_encode_test.go +++ b/weed/ec/ec_encode_test.go @@ -475,3 +475,58 @@ func TestEcShardSummaryNamesTheShardIds(t *testing.T) { "node2:8080=2 shards [1 2]", }, ecShardSummaryByNode(byNode)) } + +// A retried ec.encode clears the leftover shards of an interrupted run before +// the source health check, but the topology snapshot predates that sweep: the +// leftovers still depress FreeVolumeCount on the source disk. The counts must +// refund those slots, or the retry refuses the very volume it just cleaned up +// after. Numbers mirror the interruption-matrix failure: 8 slots, 4 volumes, +// 8 shards of other EC volumes, plus 14 leftover shards of volume 3, so the +// master charges ceil(22/10)=3 slots and reports free=1. +func TestCollectSourceFreeVolumeCountsRefundsClearedShards(t *testing.T) { + newTopo := func() *master_pb.TopologyInfo { + return &master_pb.TopologyInfo{ + DataCenterInfos: []*master_pb.DataCenterInfo{{ + Id: "dc1", + RackInfos: []*master_pb.RackInfo{{ + Id: "rack0", + DataNodeInfos: []*master_pb.DataNodeInfo{{ + Id: "127.0.0.1:8110", + Address: "127.0.0.1:8110", + DiskInfos: map[string]*master_pb.DiskInfo{ + "hdd": { + Type: "hdd", + MaxVolumeCount: 8, + VolumeCount: 4, + FreeVolumeCount: 1, + VolumeInfos: []*master_pb.VolumeInformationMessage{ + {Id: 3}, {Id: 6}, + }, + EcShardInfos: []*master_pb.VolumeEcShardInformationMessage{ + {Id: 1, EcIndexBits: 0b00000010001, DiskId: 0}, // shards 0,4 + {Id: 1, EcIndexBits: 0b10000000100, DiskId: 1}, // shards 2,10 + {Id: 2, EcIndexBits: 0b00000010001, DiskId: 0}, + {Id: 2, EcIndexBits: 0b10000000100, DiskId: 1}, + {Id: 3, EcIndexBits: 0b11111111111111, DiskId: 0}, // interrupted run's leftovers + }, + }, + }, + }}, + }}, + }}, + } + } + + counts := collectSourceFreeVolumeCounts(newTopo(), []needle.VolumeId{3}, nil) + assert.Equal(t, 3, counts["3-127.0.0.1:8110"], + "the cleared leftovers' ceil(22/10)-ceil(8/10)=2 slots must be refunded") + + counts = collectSourceFreeVolumeCounts(newTopo(), []needle.VolumeId{6}, nil) + assert.Equal(t, 1, counts["6-127.0.0.1:8110"], + "no refund when the encoded volume has no leftover shards") + + skipped := map[pb.ServerAddress]struct{}{"127.0.0.1:8110": {}} + counts = collectSourceFreeVolumeCounts(newTopo(), []needle.VolumeId{3}, skipped) + assert.Equal(t, 1, counts["3-127.0.0.1:8110"], + "no refund on a node the sweep skipped as unreachable") +} diff --git a/weed/storage/erasure_coding/ec_shards_info.go b/weed/storage/erasure_coding/ec_shards_info.go index dc8775dbb..74ab72fc1 100644 --- a/weed/storage/erasure_coding/ec_shards_info.go +++ b/weed/storage/erasure_coding/ec_shards_info.go @@ -104,6 +104,13 @@ func GetShardCount(vi *master_pb.VolumeEcShardInformationMessage) int { return ShardBits(vi.EcIndexBits).Count() } +// VolumeSlots converts an EC shard count to the number of volume slots those +// shards occupy: every DataShardsCount shards hold one volume's worth of data, +// rounded up. +func VolumeSlots(ecShardCount int64) int64 { + return (ecShardCount + DataShardsCount - 1) / DataShardsCount +} + // EcShardsTotalSize returns the sum of all shard sizes (data + parity) in // the message. Walks vi.ShardSizes directly rather than materializing a // ShardsInfo, which is significantly cheaper for callers that only need the diff --git a/weed/storage/store.go b/weed/storage/store.go index c19c35489..24a09cf11 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -467,8 +467,7 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat { effectiveMaxCount := location.MaxVolumeCount if location.isDiskSpaceLow.Load() { usedSlots := int32(location.LocalVolumesLen()) - ecShardCount := location.EcShardCount() - usedSlots += int32((ecShardCount + erasure_coding.DataShardsCount - 1) / erasure_coding.DataShardsCount) + usedSlots += int32(erasure_coding.VolumeSlots(int64(location.EcShardCount()))) effectiveMaxCount = usedSlots } if effectiveMaxCount < 0 { diff --git a/weed/topology/disk.go b/weed/topology/disk.go index 5b1333397..635942fd4 100644 --- a/weed/topology/disk.go +++ b/weed/topology/disk.go @@ -50,12 +50,6 @@ type Disk struct { // from lingering forever. const volumeRemovalGracePeriod = 10 * time.Second -// ecShardSlots returns the number of volume slots consumed by the given -// number of EC shards, rounded up to whole-volume equivalents. -func ecShardSlots(ecShardCount int64) int64 { - return (ecShardCount + erasure_coding.DataShardsCount - 1) / erasure_coding.DataShardsCount -} - func NewDisk(diskType string) *Disk { s := &Disk{} s.id = NodeId(diskType) @@ -106,7 +100,7 @@ func (d *DiskUsages) ToDiskInfo() map[string]*master_pb.DiskInfo { m := &master_pb.DiskInfo{ VolumeCount: usage.volumeCount, MaxVolumeCount: usage.maxVolumeCount, - FreeVolumeCount: usage.maxVolumeCount - (usage.volumeCount - usage.remoteVolumeCount) - ecShardSlots(usage.ecShardCount), + FreeVolumeCount: usage.maxVolumeCount - (usage.volumeCount - usage.remoteVolumeCount) - erasure_coding.VolumeSlots(usage.ecShardCount), ActiveVolumeCount: usage.activeVolumeCount, RemoteVolumeCount: usage.remoteVolumeCount, DiskTotalBytes: uint64(max(0, usage.diskTotalBytes)), @@ -174,7 +168,7 @@ func (a *DiskUsageCounts) snapshot() DiskUsageCounts { func (a *DiskUsageCounts) FreeSpace() int64 { u := a.snapshot() - return u.maxVolumeCount + u.remoteVolumeCount - u.volumeCount - ecShardSlots(u.ecShardCount) + return u.maxVolumeCount + u.remoteVolumeCount - u.volumeCount - erasure_coding.VolumeSlots(u.ecShardCount) } func (du *DiskUsages) getOrCreateDisk(diskType types.DiskType) *DiskUsageCounts { @@ -452,7 +446,7 @@ func (d *Disk) ToDiskInfo(filter VolumeFilter) *master_pb.DiskInfo { Type: string(d.Id()), VolumeCount: diskUsage.volumeCount, MaxVolumeCount: diskUsage.maxVolumeCount, - FreeVolumeCount: diskUsage.maxVolumeCount - (diskUsage.volumeCount - diskUsage.remoteVolumeCount) - ecShardSlots(diskUsage.ecShardCount), + FreeVolumeCount: diskUsage.maxVolumeCount - (diskUsage.volumeCount - diskUsage.remoteVolumeCount) - erasure_coding.VolumeSlots(diskUsage.ecShardCount), ActiveVolumeCount: diskUsage.activeVolumeCount, RemoteVolumeCount: diskUsage.remoteVolumeCount, DiskId: diskId, diff --git a/weed/topology/node.go b/weed/topology/node.go index ebbf43946..dc5300789 100644 --- a/weed/topology/node.go +++ b/weed/topology/node.go @@ -11,6 +11,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/stats" + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/storage/types" ) @@ -308,7 +309,7 @@ func (n *NodeImpl) getOrCreateDisk(diskType types.DiskType) *DiskUsageCounts { func (n *NodeImpl) AvailableSpaceFor(option *VolumeGrowOption) int64 { t := n.getOrCreateDisk(option.DiskType) freeVolumeSlotCount := atomic.LoadInt64(&t.maxVolumeCount) + atomic.LoadInt64(&t.remoteVolumeCount) - atomic.LoadInt64(&t.volumeCount) - freeVolumeSlotCount -= ecShardSlots(atomic.LoadInt64(&t.ecShardCount)) + freeVolumeSlotCount -= erasure_coding.VolumeSlots(atomic.LoadInt64(&t.ecShardCount)) return freeVolumeSlotCount }