From c4e188505341473ce79d4dd587b111c0c0629fd6 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 22 Apr 2026 10:30:13 -0700 Subject: [PATCH] fix(ec): honor disk_id in ReceiveFile so EC shards respect admin placement (#9184) (#9185) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * test(volume_server): reproduce #9184 EC ReceiveFile disk-placement bug The plugin-worker EC task sends shards via ReceiveFile, which picks Locations[0] as the target directory regardless of the admin planner's TargetDisk assignment. ReceiveFileInfo has no disk_id field, so there is no wire channel to honor the plan. Adds StartSingleVolumeClusterWithDataDirs to the integration framework so tests can launch a volume server with N data directories. The new repro asserts the current (buggy) behavior: sending three distinct EC shards via ReceiveFile leaves all three files in dir[0] and the other dirs empty. When the fix adds disk_id to ReceiveFileInfo, this assertion must flip to verify the planned placement is respected. * fix(ec): honor disk_id in ReceiveFile so EC shards respect admin placement Before this change, VolumeServer.ReceiveFile for EC shards always selected the first HDD location (Locations[0]). The plugin-worker EC task had no way to pass the admin planner's per-shard disk assignment — ReceiveFileInfo carried no disk_id field — so every received EC shard piled onto a single disk per destination server. On multi-disk servers this caused uneven load (one disk absorbing all EC shard I/O), frequent ENOSPC retries, and a growing EC backlog under sustained ingest (see issue #9184). Changes: - proto: add disk_id to ReceiveFileInfo, mirroring VolumeEcShardsCopyRequest.disk_id. - worker: DistributeEcShards tracks the planner-assigned disk per shard; sendShardFileToDestination forwards that disk id. Metadata files (ecx/ecj/vif) inherit the disk of the first data shard targeting the same node so they land next to the shards. - server: ReceiveFile honors disk_id when > 0 with bounds validation; disk_id=0 (unset) falls back to the same auto-selection pattern as VolumeEcShardsCopy (prefer disk that already has shards for this volume, then any HDD with free space, then any location with free space). Tests updated: - TestReceiveFileEcShardHonorsDiskID asserts three shards sent with disk_id={1,2,0} land on data dirs 1, 2, and 0 respectively. - TestReceiveFileEcShardRejectsInvalidDiskID pins the out-of-range disk_id rejection path. * fix(volume-rust): honor disk_id in ReceiveFile for EC shards Mirror the Go-side change: when disk_id > 0 place the EC shard on the requested disk; when unset, auto-select with the same preference order as volume_ec_shards_copy (disk already holding shards, then any HDD, then any disk). * fix(volume): compare disk_id as uint32 to avoid 32-bit overflow On 32-bit Go builds `int(fileInfo.DiskId) >= len(Locations)` can wrap a high-bit uint32 to a negative int, bypassing the bounds check before the index operation. Compare in the uint32 domain instead. * test(ec): fail invalid-disk_id test on transport error Previously a transport-level error from CloseAndRecv silently passed the test by returning early, masking any real gRPC failure. Fail loudly so only the structured ReceiveFileResponse rejection path counts as a pass. * docs(test): explain why DiskId=0 auto-selects dir 0 in EC placement test Documents the load-bearing assumption that shards are never mounted in this test, so loc.FindEcVolume always returns false and auto-select falls through to the first HDD. Saves future readers from re-deriving the expected directory for the DiskId=0 case. * fix(test): preserve baseDir/volume path for single-dir clusters StartSingleVolumeClusterWithDataDirs started naming the data directory volume0 even in the dataDirCount=1 case, which broke Scrub tests that reach into baseDir/volume via CorruptDatFile / CorruptEcShardFile / CorruptEcxFile. Keep the legacy name for single-dir clusters; only use the indexed "volumeN" layout when multiple disks are requested. --- seaweed-volume/proto/volume_server.proto | 1 + seaweed-volume/src/server/grpc_server.rs | 34 +++- test/volume_server/framework/cluster.go | 48 ++++- .../grpc/ec_receive_disk_placement_test.go | 178 ++++++++++++++++++ weed/pb/volume_server.proto | 1 + weed/pb/volume_server_pb/volume_server.pb.go | 24 ++- weed/server/volume_grpc_copy.go | 34 +++- .../erasure_coding/shard_distribution.go | 23 ++- 8 files changed, 310 insertions(+), 33 deletions(-) create mode 100644 test/volume_server/grpc/ec_receive_disk_placement_test.go diff --git a/seaweed-volume/proto/volume_server.proto b/seaweed-volume/proto/volume_server.proto index bc5d79c69..cb71d92af 100644 --- a/seaweed-volume/proto/volume_server.proto +++ b/seaweed-volume/proto/volume_server.proto @@ -339,6 +339,7 @@ message ReceiveFileInfo { bool is_ec_volume = 4; uint32 shard_id = 5; uint64 file_size = 6; + uint32 disk_id = 7; // EC shard disk; 0 = auto-select (see VolumeEcShardsCopyRequest.disk_id) } message ReceiveFileResponse { diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index ecdee7894..c06ec5f6e 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -1425,13 +1425,33 @@ impl VolumeServer for VolumeGrpcService { // Determine file path let path = if info.is_ec_volume { let store = self.state.store.read().unwrap(); - // Go prefers a HardDriveType location, then falls back to first - let dir = store - .locations - .iter() - .find(|loc| loc.disk_type == DiskType::HardDrive) - .or_else(|| store.locations.first()) - .map(|loc| loc.directory.clone()); + // 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. + let vid = VolumeId(info.volume_id); + let dir = if info.disk_id > 0 { + let count = store.locations.len(); + if (info.disk_id as usize) >= count { + resp_error = Some(format!( + "invalid disk_id {}: only have {} disks", + info.disk_id, count + )); + break; + } + Some(store.locations[info.disk_id as usize].directory.clone()) + } else { + let loc_idx = store + .find_free_location_predicate(|loc| loc.has_ec_volume(vid)) + .or_else(|| { + store.find_free_location_predicate(|loc| { + loc.disk_type == DiskType::HardDrive + }) + }) + .or_else(|| { + store.find_free_location_predicate(|_| true) + }); + loc_idx.map(|i| store.locations[i].directory.clone()) + }; drop(store); let dir = match dir { Some(d) => d, diff --git a/test/volume_server/framework/cluster.go b/test/volume_server/framework/cluster.go index 352e7f77c..437f17d3e 100644 --- a/test/volume_server/framework/cluster.go +++ b/test/volume_server/framework/cluster.go @@ -54,12 +54,25 @@ type Cluster struct { masterCmd *exec.Cmd volumeCmd *exec.Cmd + volumeDataDirs []string + cleanupOnce sync.Once } // StartSingleVolumeCluster boots one master and one volume server. func StartSingleVolumeCluster(t testing.TB, profile matrix.Profile) *Cluster { + return StartSingleVolumeClusterWithDataDirs(t, profile, 1) +} + +// StartSingleVolumeClusterWithDataDirs boots one master and one volume server +// with dataDirCount separate data directories (passed to -dir as a comma list). +// Each directory becomes its own DiskLocation on the volume server, letting +// tests exercise multi-disk EC placement paths. +func StartSingleVolumeClusterWithDataDirs(t testing.TB, profile matrix.Profile, dataDirCount int) *Cluster { t.Helper() + if dataDirCount < 1 { + t.Fatalf("dataDirCount must be >= 1, got %d", dataDirCount) + } weedBinary, err := FindOrBuildWeedBinary() if err != nil { @@ -74,8 +87,19 @@ func StartSingleVolumeCluster(t testing.TB, profile matrix.Profile) *Cluster { configDir := filepath.Join(baseDir, "config") logsDir := filepath.Join(baseDir, "logs") masterDataDir := filepath.Join(baseDir, "master") - volumeDataDir := filepath.Join(baseDir, "volume") - for _, dir := range []string{configDir, logsDir, masterDataDir, volumeDataDir} { + volumeDataDirs := make([]string, dataDirCount) + // Single-dir layout stays at baseDir/volume so existing fixtures + // (CorruptDatFile etc.) that hardcode that path keep working. Only + // multi-dir clusters get the "volumeN" layout. + for i := 0; i < dataDirCount; i++ { + if dataDirCount == 1 { + volumeDataDirs[i] = filepath.Join(baseDir, "volume") + } else { + volumeDataDirs[i] = filepath.Join(baseDir, fmt.Sprintf("volume%d", i)) + } + } + setupDirs := append([]string{configDir, logsDir, masterDataDir}, volumeDataDirs...) + for _, dir := range setupDirs { if mkErr := os.MkdirAll(dir, 0o755); mkErr != nil { t.Fatalf("create %s: %v", dir, mkErr) } @@ -120,11 +144,12 @@ func StartSingleVolumeCluster(t testing.TB, profile matrix.Profile) *Cluster { t.Fatalf("wait for master readiness: %v\nmaster log tail:\n%s", err, masterLog) } - if err = c.startVolume(volumeDataDir); err != nil { + if err = c.startVolume(volumeDataDirs); err != nil { masterLog := c.tailLog("master.log") c.Stop() t.Fatalf("start volume: %v\nmaster log tail:\n%s", err, masterLog) } + c.volumeDataDirs = volumeDataDirs if err = c.waitForHTTP(c.VolumeAdminURL() + "/status"); err != nil { volumeLog := c.tailLog("volume.log") c.Stop() @@ -184,12 +209,16 @@ func (c *Cluster) startMaster(dataDir string) error { return c.masterCmd.Start() } -func (c *Cluster) startVolume(dataDir string) error { +func (c *Cluster) startVolume(dataDirs []string) error { logFile, err := os.Create(filepath.Join(c.logsDir, "volume.log")) if err != nil { return err } + maxPerDir := make([]string, len(dataDirs)) + for i := range dataDirs { + maxPerDir[i] = "16" + } args := []string{ "-config_dir=" + c.configDir, "volume", @@ -197,8 +226,8 @@ func (c *Cluster) startVolume(dataDir string) error { "-port=" + strconv.Itoa(c.volumePort), "-port.grpc=" + strconv.Itoa(c.volumeGrpcPort), "-port.public=" + strconv.Itoa(c.volumePubPort), - "-dir=" + dataDir, - "-max=16", + "-dir=" + strings.Join(dataDirs, ","), + "-max=" + strings.Join(maxPerDir, ","), "-master=127.0.0.1:" + strconv.Itoa(c.masterPort), "-readMode=" + c.profile.ReadMode, "-concurrentUploadLimitMB=" + strconv.Itoa(c.profile.ConcurrentUploadLimitMB), @@ -415,3 +444,10 @@ func (c *Cluster) VolumePublicURL() string { func (c *Cluster) BaseDir() string { return c.baseDir } + +// VolumeDataDirs returns the data directories the volume server was started with. +// Index 0 corresponds to DiskLocation 0, index 1 to DiskLocation 1, and so on. +// Tests can scan these directories to verify where files physically landed. +func (c *Cluster) VolumeDataDirs() []string { + return append([]string(nil), c.volumeDataDirs...) +} diff --git a/test/volume_server/grpc/ec_receive_disk_placement_test.go b/test/volume_server/grpc/ec_receive_disk_placement_test.go new file mode 100644 index 000000000..de949f888 --- /dev/null +++ b/test/volume_server/grpc/ec_receive_disk_placement_test.go @@ -0,0 +1,178 @@ +package volume_server_grpc_test + +import ( + "context" + "fmt" + "os" + "path/filepath" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/test/volume_server/framework" + "github.com/seaweedfs/seaweedfs/test/volume_server/matrix" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" +) + +// Sends EC shards with disk_id={1, 2, 0} and verifies each lands on the +// requested disk (0 → auto-select → disk 0 for a fresh volume). +func TestReceiveFileEcShardHonorsDiskID(t *testing.T) { + if testing.Short() { + t.Skip("skipping integration test in short mode") + } + + const dataDirCount = 3 + clusterHarness := framework.StartSingleVolumeClusterWithDataDirs(t, matrix.P1(), dataDirCount) + conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress()) + defer conn.Close() + + dataDirs := clusterHarness.VolumeDataDirs() + if len(dataDirs) != dataDirCount { + t.Fatalf("expected %d data dirs, got %d: %v", dataDirCount, len(dataDirs), dataDirs) + } + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + + const volumeID = uint32(9184) + const collection = "ec-disk-placement" + + // DiskId=0 triggers auto-select. Since this test never calls + // VolumeEcShardsMount, loc.FindEcVolume returns false on every disk, so + // the "already holds shards" preference is skipped and auto-select falls + // through to the first HDD (dir 0). If shards get mounted between uploads + // in a future revision, the expected dir for DiskId=0 becomes the first + // disk that already holds a shard. + shards := []struct { + shardID uint32 + requestedDisk uint32 + expectedDirIdx int + }{ + {shardID: 0, requestedDisk: 1, expectedDirIdx: 1}, + {shardID: 5, requestedDisk: 2, expectedDirIdx: 2}, + {shardID: 13, requestedDisk: 0, expectedDirIdx: 0}, + } + shardExt := func(id uint32) string { return fmt.Sprintf(".ec%02d", id) } + + for _, s := range shards { + payload := []byte(fmt.Sprintf("ec-shard-payload-%02d", s.shardID)) + stream, err := grpcClient.ReceiveFile(ctx) + if err != nil { + t.Fatalf("ReceiveFile stream create for shard %d: %v", s.shardID, err) + } + if err = stream.Send(&volume_server_pb.ReceiveFileRequest{ + Data: &volume_server_pb.ReceiveFileRequest_Info{ + Info: &volume_server_pb.ReceiveFileInfo{ + VolumeId: volumeID, + Ext: shardExt(s.shardID), + Collection: collection, + IsEcVolume: true, + ShardId: s.shardID, + FileSize: uint64(len(payload)), + DiskId: s.requestedDisk, + }, + }, + }); err != nil { + t.Fatalf("ReceiveFile send info for shard %d: %v", s.shardID, err) + } + if err = stream.Send(&volume_server_pb.ReceiveFileRequest{ + Data: &volume_server_pb.ReceiveFileRequest_FileContent{FileContent: payload}, + }); err != nil { + t.Fatalf("ReceiveFile send content for shard %d: %v", s.shardID, err) + } + resp, err := stream.CloseAndRecv() + if err != nil { + t.Fatalf("ReceiveFile close for shard %d: %v", s.shardID, err) + } + if resp.GetError() != "" { + t.Fatalf("ReceiveFile shard %d error response: %s", s.shardID, resp.GetError()) + } + if resp.GetBytesWritten() != uint64(len(payload)) { + t.Fatalf("ReceiveFile shard %d bytes_written mismatch: got %d want %d", + s.shardID, resp.GetBytesWritten(), len(payload)) + } + } + + // Scan every data dir and record which shard files live where. + perDirShards := make(map[int][]uint32) + for dirIdx, dir := range dataDirs { + for _, s := range shards { + shardPath := filepath.Join(dir, fmt.Sprintf("%s_%d%s", collection, volumeID, shardExt(s.shardID))) + if _, err := os.Stat(shardPath); err == nil { + perDirShards[dirIdx] = append(perDirShards[dirIdx], s.shardID) + } else if !os.IsNotExist(err) { + t.Fatalf("unexpected stat error for %s: %v", shardPath, err) + } + } + } + + for _, s := range shards { + found := false + for _, got := range perDirShards[s.expectedDirIdx] { + if got == s.shardID { + found = true + break + } + } + if !found { + t.Fatalf("shard %d (requestedDisk=%d) expected on dir[%d], full layout: %v", + s.shardID, s.requestedDisk, s.expectedDirIdx, perDirShards) + } + for dirIdx, ids := range perDirShards { + if dirIdx == s.expectedDirIdx { + continue + } + for _, id := range ids { + if id == s.shardID { + t.Fatalf("shard %d leaked onto dir[%d]: full layout: %v", s.shardID, dirIdx, perDirShards) + } + } + } + } + + t.Logf("disk_id honored: shard placement = %v", perDirShards) +} + +func TestReceiveFileEcShardRejectsInvalidDiskID(t *testing.T) { + if testing.Short() { + t.Skip("skipping integration test in short mode") + } + + const dataDirCount = 2 + clusterHarness := framework.StartSingleVolumeClusterWithDataDirs(t, matrix.P1(), dataDirCount) + conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress()) + defer conn.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + const volumeID = uint32(91840) + const collection = "ec-invalid-disk" + payload := []byte("invalid-disk-id-payload") + + stream, err := grpcClient.ReceiveFile(ctx) + if err != nil { + t.Fatalf("ReceiveFile stream create: %v", err) + } + if err = stream.Send(&volume_server_pb.ReceiveFileRequest{ + Data: &volume_server_pb.ReceiveFileRequest_Info{ + Info: &volume_server_pb.ReceiveFileInfo{ + VolumeId: volumeID, + Ext: ".ec00", + Collection: collection, + IsEcVolume: true, + ShardId: 0, + FileSize: uint64(len(payload)), + DiskId: 99, + }, + }, + }); err != nil { + t.Fatalf("ReceiveFile send info: %v", err) + } + resp, err := stream.CloseAndRecv() + if err != nil { + t.Fatalf("ReceiveFile close: %v", err) + } + if resp.GetError() == "" { + t.Fatalf("expected invalid disk_id rejection, got success response: %+v", resp) + } +} diff --git a/weed/pb/volume_server.proto b/weed/pb/volume_server.proto index a3becf41b..0b5e1f323 100644 --- a/weed/pb/volume_server.proto +++ b/weed/pb/volume_server.proto @@ -339,6 +339,7 @@ message ReceiveFileInfo { bool is_ec_volume = 4; uint32 shard_id = 5; uint64 file_size = 6; + uint32 disk_id = 7; // EC shard disk; 0 = auto-select (see VolumeEcShardsCopyRequest.disk_id) } message ReceiveFileResponse { diff --git a/weed/pb/volume_server_pb/volume_server.pb.go b/weed/pb/volume_server_pb/volume_server.pb.go index ab631cf57..15eb6ac63 100644 --- a/weed/pb/volume_server_pb/volume_server.pb.go +++ b/weed/pb/volume_server_pb/volume_server.pb.go @@ -1,19 +1,18 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.6 -// protoc v6.33.4 +// protoc-gen-go v1.36.6 +// protoc v6.33.4 // source: volume_server.proto package volume_server_pb import ( - reflect "reflect" - sync "sync" - unsafe "unsafe" - remote_pb "github.com/seaweedfs/seaweedfs/weed/pb/remote_pb" protoreflect "google.golang.org/protobuf/reflect/protoreflect" protoimpl "google.golang.org/protobuf/runtime/protoimpl" + reflect "reflect" + sync "sync" + unsafe "unsafe" ) const ( @@ -2334,6 +2333,7 @@ type ReceiveFileInfo struct { IsEcVolume bool `protobuf:"varint,4,opt,name=is_ec_volume,json=isEcVolume,proto3" json:"is_ec_volume,omitempty"` ShardId uint32 `protobuf:"varint,5,opt,name=shard_id,json=shardId,proto3" json:"shard_id,omitempty"` FileSize uint64 `protobuf:"varint,6,opt,name=file_size,json=fileSize,proto3" json:"file_size,omitempty"` + DiskId uint32 `protobuf:"varint,7,opt,name=disk_id,json=diskId,proto3" json:"disk_id,omitempty"` // EC shard disk; 0 = auto-select (see VolumeEcShardsCopyRequest.disk_id) unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -2410,6 +2410,13 @@ func (x *ReceiveFileInfo) GetFileSize() uint64 { return 0 } +func (x *ReceiveFileInfo) GetDiskId() uint32 { + if x != nil { + return x.DiskId + } + return 0 +} + type ReceiveFileResponse struct { state protoimpl.MessageState `protogen:"open.v1"` BytesWritten uint64 `protobuf:"varint,1,opt,name=bytes_written,json=bytesWritten,proto3" json:"bytes_written,omitempty"` @@ -6883,7 +6890,7 @@ const file_volume_server_proto_rawDesc = "" + "\x12ReceiveFileRequest\x127\n" + "\x04info\x18\x01 \x01(\v2!.volume_server_pb.ReceiveFileInfoH\x00R\x04info\x12#\n" + "\ffile_content\x18\x02 \x01(\fH\x00R\vfileContentB\x06\n" + - "\x04data\"\xba\x01\n" + + "\x04data\"\xd3\x01\n" + "\x0fReceiveFileInfo\x12\x1b\n" + "\tvolume_id\x18\x01 \x01(\rR\bvolumeId\x12\x10\n" + "\x03ext\x18\x02 \x01(\tR\x03ext\x12\x1e\n" + @@ -6893,7 +6900,8 @@ const file_volume_server_proto_rawDesc = "" + "\fis_ec_volume\x18\x04 \x01(\bR\n" + "isEcVolume\x12\x19\n" + "\bshard_id\x18\x05 \x01(\rR\ashardId\x12\x1b\n" + - "\tfile_size\x18\x06 \x01(\x04R\bfileSize\"P\n" + + "\tfile_size\x18\x06 \x01(\x04R\bfileSize\x12\x17\n" + + "\adisk_id\x18\a \x01(\rR\x06diskId\"P\n" + "\x13ReceiveFileResponse\x12#\n" + "\rbytes_written\x18\x01 \x01(\x04R\fbytesWritten\x12\x14\n" + "\x05error\x18\x02 \x01(\tR\x05error\"`\n" + diff --git a/weed/server/volume_grpc_copy.go b/weed/server/volume_grpc_copy.go index 317ca5907..508bf26f3 100644 --- a/weed/server/volume_grpc_copy.go +++ b/weed/server/volume_grpc_copy.go @@ -560,18 +560,34 @@ func (vs *VolumeServer) ReceiveFile(stream volume_server_pb.VolumeServer_Receive glog.V(1).Infof("ReceiveFile: volume %d, ext %s, collection %s, shard %d, size %d", fileInfo.VolumeId, fileInfo.Ext, fileInfo.Collection, fileInfo.ShardId, fileInfo.FileSize) - // Create file path based on file info if fileInfo.IsEcVolume { - // Find storage location for EC shard + // 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. var targetLocation *storage.DiskLocation - for _, location := range vs.store.Locations { - if location.DiskType == types.HardDriveType { - targetLocation = location - break + if fileInfo.DiskId > 0 { + if fileInfo.DiskId >= uint32(len(vs.store.Locations)) { + glog.Errorf("ReceiveFile: invalid disk_id %d: only have %d disks", fileInfo.DiskId, len(vs.store.Locations)) + return stream.SendAndClose(&volume_server_pb.ReceiveFileResponse{ + Error: fmt.Sprintf("invalid disk_id %d: only have %d disks", fileInfo.DiskId, len(vs.store.Locations)), + }) + } + 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 + }) } - } - if targetLocation == nil && len(vs.store.Locations) > 0 { - targetLocation = vs.store.Locations[0] // Fall back to first available location } if targetLocation == nil { glog.Errorf("ReceiveFile: no storage location available") diff --git a/weed/storage/erasure_coding/shard_distribution.go b/weed/storage/erasure_coding/shard_distribution.go index 85ed14e56..6a8756bc8 100644 --- a/weed/storage/erasure_coding/shard_distribution.go +++ b/weed/storage/erasure_coding/shard_distribution.go @@ -110,6 +110,9 @@ func DistributeEcShards(volumeID uint32, collection string, targets []*worker_pb log := ensureLogger(logger) shardAssignment := make(map[string][]string) + // node → shardType → planner-assigned diskID. Metadata files (ecx/ecj/vif) + // inherit the first data shard's disk so they land next to the shards. + shardDisks := make(map[string]map[string]uint32) for _, target := range targets { if len(target.ShardIds) == 0 { @@ -134,6 +137,15 @@ func DistributeEcShards(volumeID uint32, collection string, targets []*worker_pb } } + if shardDisks[target.Node] == nil { + shardDisks[target.Node] = make(map[string]uint32) + } + for _, shardType := range assignedShards { + if _, already := shardDisks[target.Node][shardType]; !already { + shardDisks[target.Node][shardType] = target.DiskId + } + } + existing := shardAssignment[target.Node] if len(existing) == 0 { shardAssignment[target.Node] = assignedShards @@ -183,7 +195,11 @@ func DistributeEcShards(volumeID uint32, collection string, targets []*worker_pb }).Info("Starting shard file transfer") } - if err := sendShardFileToDestination(volumeID, collection, dialOption, destNode, filePath, shardType); err != nil { + diskID := uint32(0) + if byShard, ok := shardDisks[destNode]; ok { + diskID = byShard[shardType] + } + if err := sendShardFileToDestination(volumeID, collection, dialOption, destNode, diskID, filePath, shardType); err != nil { return nil, fmt.Errorf("failed to send %s to %s: %w", shardType, destNode, err) } @@ -277,8 +293,8 @@ func MountEcShards(volumeID uint32, collection string, shardAssignment map[strin return nil } -// sendShardFileToDestination sends a single shard file to a destination server using ReceiveFile API. -func sendShardFileToDestination(volumeID uint32, collection string, dialOption grpc.DialOption, destServer, filePath, shardType string) error { +// diskID=0 leaves disk placement to the server's auto-select. +func sendShardFileToDestination(volumeID uint32, collection string, dialOption grpc.DialOption, destServer string, diskID uint32, filePath, shardType string) error { return operation.WithVolumeServerClient(false, pb.ServerAddress(destServer), dialOption, func(client volume_server_pb.VolumeServerClient) error { file, err := os.Open(filePath) @@ -324,6 +340,7 @@ func sendShardFileToDestination(volumeID uint32, collection string, dialOption g IsEcVolume: true, ShardId: shardId, FileSize: uint64(fileInfo.Size()), + DiskId: diskID, }, }, })