From 5fbe39320c38950b004e993b79d2f4a0962fbb92 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 27 Apr 2026 15:59:57 -0700 Subject: [PATCH] fix(volume_server): pin EC shard auto-select to the .ecx-owning disk (#9212) (#9245) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(volume_server): pin EC shard auto-select to the .ecx-owning disk (#9212) ec.rebuild only sets CopyEcxFile=true on the first shard sent to the rebuilder; subsequent shards rely on VolumeEcShardsCopy / ReceiveFile auto-select to land on the same disk. The old auto-select used FindEcVolume (in-memory) to detect the "already has this volume" case. Mid-rebuild, no EC volume has been mounted yet on the destination, so FindEcVolume returns nothing and the fallback picks "any HDD with free space" — which can split shards from their .ecx across disks of the same node and feed the orphan-shard layout reported in #9212 / fixed on the loader side in #9244. Add Store.FindEcShardTargetLocation as the canonical placement primitive: prefer a mounted EC volume, then a disk that has the .ecx on disk, then any HDD, then any disk. DiskLocation.HasEcxFileOnDisk is the new on-disk check, and it looks at IdxDirectory first with a fallback to Directory to handle .ecx written before -dir.idx was configured. Both VolumeEcShardsCopy and ReceiveFile now route through the new helper, dropping their duplicated 4-level fallback ladder. No protocol changes; explicit DiskId callers are unaffected. * fix(volume_server): treat directories named *.ecx as no-match in HasEcxFileOnDisk os.Stat(".ecx") succeeds for both files and directories. If something happens to leave a directory named X.ecx in the data or idx folder, HasEcxFileOnDisk would currently report true and FindEcShardTargetLocation would route shards to that disk — where NewEcVolume's eventual OpenFile(O_RDWR) on the same path errors out. Add a !info.IsDir() check on both stat sites. Cheap and conservative. Suggested in PR #9245 review by @gemini-code-assist. * refactor(volume_server): collapse EC placement helper to a single pass FindEcShardTargetLocation called FindFreeLocation up to four times. Each call iterates s.Locations and acquires VolumesLen / EcShardCount RLocks per disk — for a typical 4-disk node that's 32 RLock cycles per placement decision. Walk s.Locations once, score each disk by tier (mounted > .ecx-on-disk > HDD > any-disk), break ties by free count. The free-slot math is factored into a small helper that mirrors FindFreeLocation's formula without re-entering the location's locks. Behaviour is unchanged: each existing tier still wins over later tiers, and within a tier the disk with the most free count still wins, matching the original max-tracking in FindFreeLocation. Suggested in PR #9245 review by @gemini-code-assist. * refactor(volume_server): thread dataShardCount as a parameter through EC placement ecFreeShardCount and FindEcShardTargetLocation referenced erasure_coding.DataShardsCount directly. Take it as a parameter so custom-ratio builds (e.g. enterprise) can swap the default without touching the helper itself, and so unit tests can pin a specific ratio independent of the package constant. Default callsites in VolumeEcShardsCopy and ReceiveFile now pass the package default explicitly; tests pass a literal 10 for clarity. * fix(volume_server): treat MaxVolumeCount=0 as unlimited in EC placement ecFreeShardCount computed `MaxVolumeCount - VolumesLen()` and went negative when MaxVolumeCount was 0 — the "unlimited disk" sentinel already honoured by Store.hasFreeDiskLocation and friends. With a negative free count, FindEcShardTargetLocation's `freeCount <= 0` guard skipped the disk entirely, so unlimited disks could never receive EC shards via the placement helper. Special-case MaxVolumeCount<=0: report a synthetic large free count that decrements with current usage, so unlimited disks are eligible and tie-breaks still prefer the less-loaded one. Added TestFindEcShardTargetLocation_HonoursUnlimitedDisk as the regression. Reported in PR #9245 review by @gemini-code-assist. * fix(volume_server): account in shard slots, not volume slots, in ecFreeShardCount FindFreeLocation in store.go ends with `free /= DataShardsCount`, converting "shard slots free" back to "volume-equivalent slots." The truncation is harmless there, but my new ecFreeShardCount inherited the same final divide and re-introduced exactly the orphan-shard hazard #9245 was meant to prevent: with MaxVolumeCount=1, VolumesLen=0, EcShardCount=1 the formula reports 0 even though the disk has room for 9 more shards, so subsequent shards route off the .ecx-owning disk into the HDD-fallback tier. Drop the trailing divide and return the count directly in shard slots. Same shape, finer granularity; tie-breaks still order by free count. The unlimited branch's "used" calculation is updated to match (mix volume-slots and shard-slots in shard units). Added TestFindEcShardTargetLocation_TightProvisioningKeepsEcxDisk as the regression. Reported in PR #9245 review by @coderabbitai. --- weed/server/volume_grpc_copy.go | 26 +-- weed/server/volume_grpc_erasure_coding.go | 27 +-- weed/storage/disk_location_ec.go | 22 ++ weed/storage/store_ec.go | 104 ++++++++++ weed/storage/store_ec_target_location_test.go | 193 ++++++++++++++++++ 5 files changed, 338 insertions(+), 34 deletions(-) create mode 100644 weed/storage/store_ec_target_location_test.go diff --git a/weed/server/volume_grpc_copy.go b/weed/server/volume_grpc_copy.go index e1aa7812c..c0af5677e 100644 --- a/weed/server/volume_grpc_copy.go +++ b/weed/server/volume_grpc_copy.go @@ -572,8 +572,12 @@ func (vs *VolumeServer) ReceiveFile(stream volume_server_pb.VolumeServer_Receive } // disk_id=0 means "unset" (protobuf default), so auto-select - // mirrors VolumeEcShardsCopy: prefer a disk already holding - // this volume's shards, then any HDD, then any disk. + // using the same primitive as VolumeEcShardsCopy: prefer a + // disk that has the EC volume mounted, then a disk that owns + // the .ecx on disk (the volume hasn't been mounted yet — + // relevant when shards stream in mid-rebuild before any + // mount has happened; see #9212), then any HDD, then any + // disk. var targetLocation *storage.DiskLocation if fileInfo.DiskId > 0 { if fileInfo.DiskId >= uint32(len(vs.store.Locations)) { @@ -584,20 +588,10 @@ func (vs *VolumeServer) ReceiveFile(stream volume_server_pb.VolumeServer_Receive } targetLocation = vs.store.Locations[fileInfo.DiskId] } else { - targetLocation = vs.store.FindFreeLocation(func(loc *storage.DiskLocation) bool { - _, found := loc.FindEcVolume(needle.VolumeId(fileInfo.VolumeId)) - return found - }) - if targetLocation == nil { - targetLocation = vs.store.FindFreeLocation(func(loc *storage.DiskLocation) bool { - return loc.DiskType == types.HardDriveType - }) - } - if targetLocation == nil { - targetLocation = vs.store.FindFreeLocation(func(loc *storage.DiskLocation) bool { - return true - }) - } + // Pass the build's default data-shard count for the helper's + // free-slot maths; it's a parameter so custom-ratio builds + // (e.g. enterprise) can swap it without touching this file. + targetLocation = vs.store.FindEcShardTargetLocation(fileInfo.Collection, needle.VolumeId(fileInfo.VolumeId), erasure_coding.DataShardsCount) } if targetLocation == nil { glog.Errorf("ReceiveFile: no storage location available") diff --git a/weed/server/volume_grpc_erasure_coding.go b/weed/server/volume_grpc_erasure_coding.go index b3f10efd3..ec05a209d 100644 --- a/weed/server/volume_grpc_erasure_coding.go +++ b/weed/server/volume_grpc_erasure_coding.go @@ -257,24 +257,15 @@ func (vs *VolumeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_serv location = vs.store.Locations[req.DiskId] glog.V(1).Infof("Using disk %d for EC shard copy: %s", req.DiskId, location.Directory) } else { - // Prefer a location that already has shards for this volume, - // so all shards end up on the same disk for rebuild. - location = vs.store.FindFreeLocation(func(loc *storage.DiskLocation) bool { - _, found := loc.FindEcVolume(needle.VolumeId(req.VolumeId)) - return found - }) - if location == nil { - // Fall back to any HDD location with free space - location = vs.store.FindFreeLocation(func(loc *storage.DiskLocation) bool { - return loc.DiskType == types.HardDriveType - }) - } - if location == nil { - // Fall back to any location with free space - location = vs.store.FindFreeLocation(func(loc *storage.DiskLocation) bool { - return true - }) - } + // Auto-select the target disk: prefer a disk that already has the + // EC volume mounted, then a disk that owns the .ecx on disk (the + // volume hasn't been mounted yet — relevant for ec.rebuild, where + // only the first shard carries .ecx and subsequent shards must + // land on the same disk; see #9212), then any HDD, then any disk. + // Pass the build's default data-shard count for free-slot maths; + // the helper takes it as a parameter so custom-ratio builds (e.g. + // enterprise) can swap it without touching this file. + location = vs.store.FindEcShardTargetLocation(req.Collection, needle.VolumeId(req.VolumeId), erasure_coding.DataShardsCount) if location == nil { return nil, fmt.Errorf("no space left") } diff --git a/weed/storage/disk_location_ec.go b/weed/storage/disk_location_ec.go index cc6ef8f8a..dcbc9217d 100644 --- a/weed/storage/disk_location_ec.go +++ b/weed/storage/disk_location_ec.go @@ -92,6 +92,28 @@ func (l *DiskLocation) FindEcShard(vid needle.VolumeId, shardId erasure_coding.S return nil, false } +// HasEcxFileOnDisk reports whether this disk has a sealed .ecx index file +// for the given (collection, vid). Unlike FindEcVolume this does not +// require the EC volume to be mounted in memory, which makes it the right +// primitive for placement decisions during ec.balance / ec.rebuild flows +// where shards may arrive before any mount has happened on the receiving +// disk. Without checking the on-disk state, auto-select can split shards +// from the .ecx that travels with the first shard, which is the source of +// the orphan-shard layout reported in #9212. +func (l *DiskLocation) HasEcxFileOnDisk(collection string, vid needle.VolumeId) bool { + idxBase := erasure_coding.EcShardFileName(collection, l.IdxDirectory, int(vid)) + if info, err := os.Stat(idxBase + ".ecx"); err == nil && !info.IsDir() { + return true + } + if l.IdxDirectory != l.Directory { + dataBase := erasure_coding.EcShardFileName(collection, l.Directory, int(vid)) + if info, err := os.Stat(dataBase + ".ecx"); err == nil && !info.IsDir() { + return true + } + } + return false +} + func (l *DiskLocation) LoadEcShard(collection string, vid needle.VolumeId, shardId erasure_coding.ShardId) (*erasure_coding.EcVolume, error) { ecVolumeShard, err := erasure_coding.NewEcVolumeShard(l.DiskType, l.Directory, collection, vid, shardId) diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go index 0c1cc91b5..0a074f6b1 100644 --- a/weed/storage/store_ec.go +++ b/weed/storage/store_ec.go @@ -22,6 +22,110 @@ import ( "github.com/seaweedfs/seaweedfs/weed/storage/types" ) +// FindEcShardTargetLocation returns the disk that should receive a new +// shard / index file for (collection, vid). The selection order is: +// +// 1. a disk that already has the EC volume mounted (in-memory state), +// 2. a disk that owns the .ecx file on disk (volume not mounted yet), +// 3. any HDD with free space, +// 4. any disk with free space. +// +// Step 2 is the missing primitive that pinned subsequent shards to the +// first-shard disk during ec.rebuild. ec.rebuild only sets CopyEcxFile=true +// for the first shard, then relies on auto-select to land later shards on +// the same disk. Without an on-disk check, FindEcVolume returns nothing +// (no mount yet) and the fallback picks "any HDD with free space" — which +// can split shards from their index files across disks of the same node +// and lose them at startup. See issue #9212 and the orphan-shard +// reconciliation in #9244. +// +// dataShardCount is the data-shard count for this volume's EC layout (10 +// for the OSS default, but custom ratios are supported via .vif). Callers +// pass it explicitly so this helper stays free of package-level constants +// — easier to mirror into builds that ship a different default ratio. +// +// Implementation walks s.Locations once and scores each disk by tier; the +// highest-tier disk wins, ties broken by free count. The earlier waterfall +// across four FindFreeLocation passes was equivalent but acquired +// volumesLock and ecVolumesLock RLocks (via VolumesLen / EcShardCount) up +// to four times per disk per call. +func (s *Store) FindEcShardTargetLocation(collection string, vid needle.VolumeId, dataShardCount int) *DiskLocation { + const ( + tierAnyDisk = iota + 1 + tierHDD + tierEcxOnDisk + tierMounted + ) + + var ( + best *DiskLocation + bestTier int + bestFree int32 + ) + for _, loc := range s.Locations { + if loc.isDiskSpaceLow { + continue + } + freeCount := ecFreeShardCount(loc, dataShardCount) + if freeCount <= 0 { + continue + } + tier := tierAnyDisk + if loc.DiskType == types.HardDriveType { + tier = tierHDD + } + if loc.HasEcxFileOnDisk(collection, vid) { + tier = tierEcxOnDisk + } + if _, mounted := loc.FindEcVolume(vid); mounted { + tier = tierMounted + } + if best == nil || tier > bestTier || (tier == bestTier && freeCount > bestFree) { + best = loc + bestTier = tier + bestFree = freeCount + } + } + return best +} + +// ecFreeShardCount returns the free EC shard capacity of loc, expressed +// in shard slots (not volume-equivalent slots). dataShardCount is the +// data-shard count of the EC layout being placed — see +// FindEcShardTargetLocation's docstring for why it's a parameter. +// +// FindFreeLocation in store.go does the same math but divides by +// DataShardsCount at the end. That truncation can exclude a disk that +// still has room for several individual shards (e.g. MaxVolumeCount=1, +// EcShardCount=1, dataShardCount=10 → reports 0 despite 9 free shard +// slots), which in this helper would re-route subsequent shards off the +// .ecx-owning disk and re-introduce the orphan-shard layout #9212 is +// trying to prevent. So we keep the result in shard slots throughout. +// +// MaxVolumeCount == 0 is the "unlimited" sentinel used elsewhere in the +// store (see hasFreeDiskLocation). Reporting a synthetic large free +// count keeps unlimited disks eligible while still letting tie-breaks +// prefer the less-loaded one. +func ecFreeShardCount(loc *DiskLocation, dataShardCount int) int32 { + if dataShardCount <= 0 { + return 0 + } + if loc.MaxVolumeCount <= 0 { + const unlimitedFree = int32(1 << 30) + used := int32(loc.VolumesLen())*int32(dataShardCount) + int32(loc.EcShardCount()) + if used >= unlimitedFree { + return 1 + } + return unlimitedFree - used + } + free := (loc.MaxVolumeCount - int32(loc.VolumesLen())) * int32(dataShardCount) + free -= int32(loc.EcShardCount()) + if free < 0 { + return 0 + } + return free +} + func (s *Store) CollectErasureCodingHeartbeat() *master_pb.Heartbeat { var ecShardMessages []*master_pb.VolumeEcShardInformationMessage collectionEcShardSize := make(map[string]int64) diff --git a/weed/storage/store_ec_target_location_test.go b/weed/storage/store_ec_target_location_test.go new file mode 100644 index 000000000..92283647e --- /dev/null +++ b/weed/storage/store_ec_target_location_test.go @@ -0,0 +1,193 @@ +package storage + +import ( + "os" + "path/filepath" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +// dataShardCount is the data-shard count threaded into +// FindEcShardTargetLocation by these tests. Kept as a literal so the test +// stays valid when enterprise builds use a different default ratio. +const dataShardCount = 10 + +// TestFindEcShardTargetLocation_PinsToEcxOnDisk reproduces the placement +// half of issue #9212. ec.rebuild copies the .ecx alongside the first +// shard, then sends subsequent shards with CopyEcxFile=false relying on +// the volume server's auto-select to land them on the same disk. The +// volume isn't mounted yet, so FindEcVolume can't see the .ecx — without +// an on-disk check the selection falls back to "any HDD with free space" +// and shards end up split from their index files across disks of the +// same node. +// +// The fix: FindEcShardTargetLocation also looks for the .ecx on disk +// before falling through to the generic disk-space heuristic. +func TestFindEcShardTargetLocation_PinsToEcxOnDisk(t *testing.T) { + store := newEcTargetTestStore(t, 3) + collection := "grafana-loki" + vid := needle.VolumeId(1093) + + // Drop a sealed .ecx onto disk 2. Nothing is mounted yet — this is + // the state right after ec.rebuild's first VolumeEcShardsCopy with + // CopyEcxFile=true and before any VolumeEcShardsMount has run. + base := erasure_coding.EcShardFileName(collection, store.Locations[2].IdxDirectory, int(vid)) + if err := os.WriteFile(base+".ecx", make([]byte, 20), 0o644); err != nil { + t.Fatalf("seed .ecx on disk 2: %v", err) + } + + got := store.FindEcShardTargetLocation(collection, vid, dataShardCount) + if got == nil { + t.Fatalf("FindEcShardTargetLocation returned nil; expected disk 2") + } + if got != store.Locations[2] { + t.Errorf("placement leaked off the .ecx-owning disk: got %s, want %s (issue #9212)", + got.Directory, store.Locations[2].Directory) + } +} + +// TestFindEcShardTargetLocation_PrefersMountedOverEcx checks that an +// already-mounted EC volume on disk 1 wins over a stray .ecx on disk 2. +// This protects the post-startup steady state from being perturbed by +// leftover index files from a prior failed move. +func TestFindEcShardTargetLocation_PrefersMountedOverEcx(t *testing.T) { + store := newEcTargetTestStore(t, 3) + collection := "grafana-loki" + vid := needle.VolumeId(2222) + + // Mount a placeholder EC volume on disk 1 so FindEcVolume returns it. + loc1 := store.Locations[1] + loc1.ecVolumesLock.Lock() + loc1.ecVolumes[vid] = &erasure_coding.EcVolume{VolumeId: vid, Collection: collection} + loc1.ecVolumesLock.Unlock() + + // Drop a stray .ecx on disk 2 to make sure it does NOT win. + base := erasure_coding.EcShardFileName(collection, store.Locations[2].IdxDirectory, int(vid)) + if err := os.WriteFile(base+".ecx", make([]byte, 20), 0o644); err != nil { + t.Fatalf("seed .ecx on disk 2: %v", err) + } + + got := store.FindEcShardTargetLocation(collection, vid, dataShardCount) + if got != loc1 { + t.Errorf("placement should follow mounted EC volume on disk 1, got %v", got) + } +} + +// TestFindEcShardTargetLocation_FallsThroughToHddWhenNothingMatches keeps +// the existing fallback behaviour intact for the cold-volume case (no +// mount, no .ecx anywhere on this server). +func TestFindEcShardTargetLocation_FallsThroughToHddWhenNothingMatches(t *testing.T) { + store := newEcTargetTestStore(t, 2) + collection := "grafana-loki" + vid := needle.VolumeId(3333) + + got := store.FindEcShardTargetLocation(collection, vid, dataShardCount) + if got == nil { + t.Fatalf("FindEcShardTargetLocation returned nil; expected an HDD fallback") + } + if got.DiskType != types.HardDriveType { + t.Errorf("fallback should pick an HDD; got disk type %q", got.DiskType) + } +} + +// TestFindEcShardTargetLocation_HonoursUnlimitedDisk pins the +// MaxVolumeCount==0 ("unlimited") convention shared with +// hasFreeDiskLocation. ecFreeShardCount used to return a negative free +// count for unlimited disks, which made FindEcShardTargetLocation skip +// them entirely. PR #9245 review by @gemini-code-assist. +func TestFindEcShardTargetLocation_HonoursUnlimitedDisk(t *testing.T) { + store := newEcTargetTestStore(t, 1) + store.Locations[0].MaxVolumeCount = 0 // unlimited + + got := store.FindEcShardTargetLocation("grafana-loki", needle.VolumeId(4444), dataShardCount) + if got == nil { + t.Fatalf("FindEcShardTargetLocation returned nil for an unlimited (MaxVolumeCount=0) disk") + } + if got != store.Locations[0] { + t.Errorf("expected the only (unlimited) disk to be picked; got %v", got) + } +} + +// TestFindEcShardTargetLocation_TightProvisioningKeepsEcxDisk pins the +// truncation hazard PR #9245 review by @coderabbitai surfaced. +// +// With MaxVolumeCount=1, VolumesLen=0, and one EC shard already on the +// disk, the previous formula (free = (1*10 - 1) / 10 = 0) would treat +// the disk as full and route subsequent shards to a different disk — +// exactly the orphan-shard layout this PR exists to prevent. The fix +// keeps the free count in shard slots, so 9 free slots is reported as +// 9 rather than rounded down to 0. +func TestFindEcShardTargetLocation_TightProvisioningKeepsEcxDisk(t *testing.T) { + store := newEcTargetTestStore(t, 2) + store.Locations[0].MaxVolumeCount = 1 + store.Locations[1].MaxVolumeCount = 1 + + collection := "grafana-loki" + vid := needle.VolumeId(5555) + + // Seed disk 1 with a single EC shard for this volume so it owns the + // .ecx and has 9 free shard slots remaining; the old formula would + // have rounded that to 0. + loc1 := store.Locations[1] + loc1.ecVolumesLock.Lock() + loc1.ecVolumes[vid] = &erasure_coding.EcVolume{ + VolumeId: vid, + Collection: collection, + Shards: []*erasure_coding.EcVolumeShard{{VolumeId: vid, ShardId: 0, Collection: collection}}, + } + loc1.ecVolumesLock.Unlock() + + got := store.FindEcShardTargetLocation(collection, vid, dataShardCount) + if got != loc1 { + t.Errorf("expected the .ecx-owning disk (1 shard placed, 9 free shard slots) to be picked; got %v", got) + } +} + +// newEcTargetTestStore is a leaner cousin of the helper in +// store_load_balancing_test.go: it spins up an in-memory Store with N +// HDD disk locations under a single t.TempDir and consumes any heartbeat +// channel traffic so the placement helpers can be exercised directly. +func newEcTargetTestStore(t *testing.T, numDirs int) *Store { + t.Helper() + tempDir := t.TempDir() + dirs := make([]string, 0, numDirs) + maxCounts := make([]int32, 0, numDirs) + minFreeSpaces := make([]util.MinFreeSpace, 0, numDirs) + diskTypes := make([]types.DiskType, 0, numDirs) + for i := 0; i < numDirs; i++ { + dir := filepath.Join(tempDir, "data", filepath.Base(t.Name())+"-"+string(rune('a'+i))) + if err := os.MkdirAll(dir, 0o755); err != nil { + t.Fatalf("mkdir %s: %v", dir, err) + } + dirs = append(dirs, dir) + maxCounts = append(maxCounts, 100) + minFreeSpaces = append(minFreeSpaces, util.MinFreeSpace{}) + diskTypes = append(diskTypes, types.HardDriveType) + } + store := NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id", + dirs, maxCounts, minFreeSpaces, "", NeedleMapInMemory, diskTypes, nil, 3, + ) + done := make(chan struct{}) + go func() { + for { + select { + case <-store.NewVolumesChan: + case <-store.NewEcShardsChan: + case <-store.DeletedVolumesChan: + case <-store.DeletedEcShardsChan: + case <-store.StateUpdateChan: + case <-done: + return + } + } + }() + t.Cleanup(func() { + store.Close() + close(done) + }) + return store +}