diff --git a/weed/ec/ec_balance_no_space_test.go b/weed/ec/ec_balance_no_space_test.go new file mode 100644 index 000000000..1b41cfde9 --- /dev/null +++ b/weed/ec/ec_balance_no_space_test.go @@ -0,0 +1,359 @@ +package ec + +import ( + "context" + "fmt" + "net" + "strings" + "sync" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding/ecbalancer" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +// ecBalanceFakeServer answers the EC-shard RPC subset a balance run drives: +// copy (rejected when full), mount, inventory, unmount, delete. The disk state +// a real server tracks lives in shards; what the planner saw is rebuilt from it +// through FetchTopology, so a landed move is visible to the next plan. +type ecBalanceFakeServer struct { + volume_server_pb.UnimplementedVolumeServerServer + + mu sync.Mutex + shards map[uint32]map[uint32]bool // vid -> shardId -> present + full bool // VolumeEcShardsCopy answers "no space left" + copyCalls int + grpcAddr string + stop func() +} + +func (f *ecBalanceFakeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_server_pb.VolumeEcShardsCopyRequest) (*volume_server_pb.VolumeEcShardsCopyResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.copyCalls++ + if f.full { + return nil, fmt.Errorf("no space left") + } + return &volume_server_pb.VolumeEcShardsCopyResponse{}, nil +} + +func (f *ecBalanceFakeServer) VolumeEcShardsMount(ctx context.Context, req *volume_server_pb.VolumeEcShardsMountRequest) (*volume_server_pb.VolumeEcShardsMountResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + for _, sid := range req.ShardIds { + f.shards[req.VolumeId][sid] = true + } + return &volume_server_pb.VolumeEcShardsMountResponse{}, nil +} + +func (f *ecBalanceFakeServer) VolumeEcShardsInfo(ctx context.Context, req *volume_server_pb.VolumeEcShardsInfoRequest) (*volume_server_pb.VolumeEcShardsInfoResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + resp := &volume_server_pb.VolumeEcShardsInfoResponse{} + for sid := range f.shards[req.VolumeId] { + resp.EcShardInfos = append(resp.EcShardInfos, &volume_server_pb.EcShardInfo{ + VolumeId: req.VolumeId, + ShardId: sid, + Size: 1, + }) + } + return resp, nil +} + +func (f *ecBalanceFakeServer) VolumeEcShardsUnmount(ctx context.Context, req *volume_server_pb.VolumeEcShardsUnmountRequest) (*volume_server_pb.VolumeEcShardsUnmountResponse, error) { + return &volume_server_pb.VolumeEcShardsUnmountResponse{}, nil +} + +func (f *ecBalanceFakeServer) VolumeEcShardsDelete(ctx context.Context, req *volume_server_pb.VolumeEcShardsDeleteRequest) (*volume_server_pb.VolumeEcShardsDeleteResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + for _, sid := range req.ShardIds { + delete(f.shards[req.VolumeId], sid) + } + return &volume_server_pb.VolumeEcShardsDeleteResponse{}, nil +} + +// nodeId returns the "host:httpPort.grpcPort" node id form the cluster uses, so +// ToGrpcAddress resolves to this server's listener. +func (f *ecBalanceFakeServer) nodeId() string { + host, port, err := net.SplitHostPort(f.grpcAddr) + if err != nil { + return f.grpcAddr + } + return fmt.Sprintf("%s:1.%s", host, port) +} + +func (f *ecBalanceFakeServer) shardCount(vid uint32) int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.shards[vid]) +} + +func startEcBalanceFakeServer(t *testing.T, vid uint32, present ...uint32) *ecBalanceFakeServer { + t.Helper() + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + + f := &ecBalanceFakeServer{ + shards: map[uint32]map[uint32]bool{vid: {}}, + grpcAddr: listener.Addr().String(), + } + for _, sid := range present { + f.shards[vid][sid] = true + } + + server := grpc.NewServer() + volume_server_pb.RegisterVolumeServerServer(server, f) + go func() { _ = server.Serve(listener) }() + f.stop = server.Stop + t.Cleanup(server.Stop) + return f +} + +// fetchEcBalanceTopology builds a FetchTopology that reports the fake servers' +// live shard inventory. +func fetchEcBalanceTopology(servers []*ecBalanceFakeServer) func(delay time.Duration) (*master_pb.TopologyInfo, uint64, error) { + return func(delay time.Duration) (*master_pb.TopologyInfo, uint64, error) { + var nodes []*master_pb.DataNodeInfo + for _, f := range servers { + di := &master_pb.DiskInfo{Type: string(types.HardDriveType), MaxVolumeCount: 100} + f.mu.Lock() + for v, set := range f.shards { + if len(set) == 0 { + continue + } + si := erasure_coding.NewShardsInfo() + for sid := range set { + si.Set(erasure_coding.NewShardInfo(erasure_coding.ShardId(sid), 1)) + } + di.EcShardInfos = append(di.EcShardInfos, &master_pb.VolumeEcShardInformationMessage{ + Id: v, + Collection: "c1", + EcIndexBits: si.Bitmap(), + ShardSizes: si.SizesInt64(), + DiskType: string(types.HardDriveType), + }) + } + f.mu.Unlock() + nodes = append(nodes, &master_pb.DataNodeInfo{ + Id: f.nodeId(), + DiskInfos: map[string]*master_pb.DiskInfo{string(types.HardDriveType): di}, + }) + } + return &master_pb.TopologyInfo{ + DataCenterInfos: []*master_pb.DataCenterInfo{{ + Id: "dc1", + RackInfos: []*master_pb.RackInfo{{ + Id: "r1", + DataNodeInfos: nodes, + }}, + }}, + }, 0, nil + } +} + +// TestEcBalanceReplanAfterNoSpaceLeft reproduces seaweedfs/seaweedfs#11753: +// the planner works from heartbeat free-slot counts that can disagree with the +// volume server's own check, so a planned target can reject the copy with +// "no space left" mid-run. The balance must re-plan without that node as a +// target rather than fail the whole run. +func TestEcBalanceReplanAfterNoSpaceLeft(t *testing.T) { + const vid = uint32(1) + var allShards []uint32 + for sid := 0; sid < erasure_coding.DataShardsCount+erasure_coding.ParityShardsCount; sid++ { + allShards = append(allShards, uint32(sid)) + } + + full := startEcBalanceFakeServer(t, vid) + accept := startEcBalanceFakeServer(t, vid) + source := startEcBalanceFakeServer(t, vid, allShards...) + full.full = true + + env := &Env{ + GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + FetchTopology: fetchEcBalanceTopology([]*ecBalanceFakeServer{source, full, accept}), + } + + err := EcBalance(env, []string{"c1"}, "", nil, types.HardDriveType, 3, 0, true, nil, nil, nil) + if err != nil { + t.Fatalf("balance should re-plan around the full target, got: %v", err) + } + if full.copyCalls == 0 { + t.Fatalf("expected the full server to have rejected at least one copy") + } + if got := full.shardCount(vid); got != 0 { + t.Fatalf("full server should receive no shards, holds %d", got) + } + total := source.shardCount(vid) + full.shardCount(vid) + accept.shardCount(vid) + if total != len(allShards) { + t.Fatalf("shards lost or duplicated: total %d, want %d", total, len(allShards)) + } + if source.shardCount(vid) == len(allShards) { + t.Fatalf("no shards were moved off the source") + } +} + +// TestEcBalanceReplanAfterNoSpaceLeftStaleSnapshot serves the pre-move +// topology on every collection after the first, simulating a master heartbeat +// that lags the landed moves. The re-plan must replay its completed moves onto +// the stale snapshot instead of re-moving shards off the deleted source. +func TestEcBalanceReplanAfterNoSpaceLeftStaleSnapshot(t *testing.T) { + const vid = uint32(1) + var allShards []uint32 + for sid := 0; sid < erasure_coding.DataShardsCount+erasure_coding.ParityShardsCount; sid++ { + allShards = append(allShards, uint32(sid)) + } + + full := startEcBalanceFakeServer(t, vid) + accept := startEcBalanceFakeServer(t, vid) + source := startEcBalanceFakeServer(t, vid, allShards...) + full.full = true + + live := fetchEcBalanceTopology([]*ecBalanceFakeServer{source, full, accept}) + var stale *master_pb.TopologyInfo + calls := 0 + fetchTopology := func(delay time.Duration) (*master_pb.TopologyInfo, uint64, error) { + topo, size, err := live(delay) + if err != nil { + return nil, 0, err + } + calls++ + if stale == nil { + stale = topo + } + if calls > 1 { + return stale, size, nil + } + return topo, size, nil + } + + env := &Env{ + GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + FetchTopology: fetchTopology, + } + + err := EcBalance(env, []string{"c1"}, "", nil, types.HardDriveType, 3, 0, true, nil, nil, nil) + if err != nil { + t.Fatalf("balance should re-plan around the full target, got: %v", err) + } + if calls < 2 { + t.Fatalf("expected a re-plan, topology collected %d time(s)", calls) + } + // Each shard lands on the accepting server exactly once; a stale re-plan + // would copy already-moved shards again. + if got, want := accept.copyCalls, accept.shardCount(vid); got != want { + t.Fatalf("accepting server saw %d copies for %d held shards: completed moves were re-planned", got, want) + } + total := source.shardCount(vid) + full.shardCount(vid) + accept.shardCount(vid) + if total != len(allShards) { + t.Fatalf("shards lost or duplicated: total %d, want %d", total, len(allShards)) + } +} + +// TestEcBalanceNoReceivingTargetFails surfaces the refusal when every +// destination that could take a shard has rejected a copy, rather than +// reporting success for a plan that moved nothing. +func TestEcBalanceNoReceivingTargetFails(t *testing.T) { + const vid = uint32(1) + var allShards []uint32 + for sid := 0; sid < erasure_coding.DataShardsCount+erasure_coding.ParityShardsCount; sid++ { + allShards = append(allShards, uint32(sid)) + } + + full := startEcBalanceFakeServer(t, vid) + source := startEcBalanceFakeServer(t, vid, allShards...) + full.full = true + + env := &Env{ + GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + FetchTopology: fetchEcBalanceTopology([]*ecBalanceFakeServer{source, full}), + } + + err := EcBalance(env, []string{"c1"}, "", nil, types.HardDriveType, 3, 0, true, nil, nil, nil) + if err == nil { + t.Fatalf("balance with no receiving target should fail") + } + if !strings.Contains(err.Error(), "no space left") { + t.Fatalf("expected the refusal to surface, got: %v", err) + } +} + +// TestApplyCompletedMovesReplaysDiskState checks that replaying a landed move +// onto a stale snapshot carries the shard's disk attribution and the per-disk +// slot accounting, so a re-plan does not treat the just-filled disk as empty. +func TestApplyCompletedMovesReplaysDiskState(t *testing.T) { + const vid = uint32(1) + + source := newEcNode("dc1", "r1", "dn1", 99) + source.Info.DiskInfos["hdd"] = &master_pb.DiskInfo{ + Type: "hdd", + EcShardInfos: []*master_pb.VolumeEcShardInformationMessage{{ + Id: vid, + Collection: "c1", + DiskId: 1, + DiskType: "hdd", + EcIndexBits: uint32(erasure_coding.ShardBits(0).Set(5)), + }}, + } + si := erasure_coding.NewShardsInfo() + si.Set(erasure_coding.NewShardInfo(5, 1)) + source.Disks = map[uint32]*EcDisk{ + 1: {DiskId: 1, DiskType: "hdd", FreeEcSlots: 99, EcShardCount: 1, EcShards: map[needle.VolumeId]*erasure_coding.ShardsInfo{1: si}}, + } + + target := newEcNode("dc1", "r1", "dn2", 100) + target.Disks = map[uint32]*EcDisk{ + 1: {DiskId: 1, DiskType: "hdd", FreeEcSlots: 10, EcShardCount: 0, EcShards: map[needle.VolumeId]*erasure_coding.ShardsInfo{}}, + 2: {DiskId: 2, DiskType: "hdd", FreeEcSlots: 10, EcShardCount: 0, EcShards: map[needle.VolumeId]*erasure_coding.ShardsInfo{}}, + } + + nodes := []*EcNode{source, target} + move := ecbalancer.Move{ + VolumeID: vid, + ShardID: 5, + Collection: "c1", + SourceNode: "dn1", + TargetNode: "dn2", + TargetDisk: 2, + Phase: "within_rack", + } + applyCompletedMoves(nodes, []ecbalancer.Move{move}, types.HardDriveType) + + if ecNodeHasShard(source, needle.VolumeId(vid), 5) { + t.Fatalf("source still holds shard 5") + } + if d := source.Disks[1]; d.EcShardCount != 0 || d.FreeEcSlots != 100 { + t.Fatalf("source disk counts not restored: %+v", d) + } + if !ecNodeHasShard(target, needle.VolumeId(vid), 5) { + t.Fatalf("target missing shard 5") + } + if d := target.Disks[2]; d.EcShardCount != 1 || d.FreeEcSlots != 9 { + t.Fatalf("target disk 2 not charged: %+v", d) + } + if d := target.Disks[1]; d.EcShardCount != 0 { + t.Fatalf("shard attributed to wrong disk: %+v", d) + } + if target.FreeEcSlot != 99 { + t.Fatalf("target free slots = %d, want 99", target.FreeEcSlot) + } + + // Replaying again (or onto a fresher snapshot) must not double-count. + applyCompletedMoves(nodes, []ecbalancer.Move{move}, types.HardDriveType) + if d := target.Disks[2]; d.EcShardCount != 1 || d.FreeEcSlots != 9 { + t.Fatalf("double-applied move: %+v", d) + } + if target.FreeEcSlot != 99 { + t.Fatalf("target free slots = %d, want 99", target.FreeEcSlot) + } +} diff --git a/weed/ec/ec_common.go b/weed/ec/ec_common.go index e7cb27b17..8cabfe426 100644 --- a/weed/ec/ec_common.go +++ b/weed/ec/ec_common.go @@ -8,6 +8,7 @@ import ( "sort" "strconv" "strings" + "sync" "time" "github.com/seaweedfs/seaweedfs/weed/glog" @@ -762,6 +763,26 @@ type ecBalancer struct { // migratingVolumeIds are mid-encode volumes whose shards are ingested from // every disk-type bucket; see EcBalance. migratingVolumeIds map[uint32]bool + // noReceiveTargets are node ids the planner may not pick as destinations; + // noReceiveDisks narrows that to individual disks on a node: a server-side + // auto-pick refusal means no disk qualifies, but an explicit-disk refusal + // only convicts that disk. See EcBalance. + noReceiveTargets map[string]bool + noReceiveDisks map[string]map[uint32]bool + // mu guards the execution journal written by concurrent move tasks. + mu sync.Mutex + // rejectedTargets and rejectedDisks collect the destinations that refused a + // copy this attempt; completedMoves records applied moves so a retry can + // replay them onto a stale-heartbeat snapshot; sawOtherErr marks any + // failure that is not a "no space left" refusal and makes the run + // non-retryable. + rejectedTargets map[string]bool + rejectedDisks map[string]map[uint32]bool + completedMoves []ecbalancer.Move + sawOtherErr bool + // plannedMoves is the move count of the last plan, for the empty-replan + // check in EcBalance. + plannedMoves int } // excludeNodes is a set of server addresses kept out of the balance as copy/move @@ -782,32 +803,6 @@ type ecBalancer struct { // so a plain ec.balance -diskType=X never drags deliberately tiered shards of // other types onto X disks. func EcBalance(env *Env, collections []string, dc string, ecReplicaPlacement *super_block.ReplicaPlacement, diskType types.DiskType, maxParallelization int, ioBytePerSecond int64, applyBalancing bool, excludeNodes map[pb.ServerAddress]struct{}, volumeIds []needle.VolumeId, migratingVolumeIds []needle.VolumeId) (err error) { - // collect all ec nodes - allEcNodes, totalFreeEcSlots, err := CollectEcNodesForDC(env, dc, diskType) - if err != nil { - return err - } - - // Drop excluded nodes (and the slots they contribute) before planning so they - // can be neither a target nor a source for any move this balance plans. - if len(excludeNodes) > 0 { - kept := allEcNodes[:0] - var excludedFreeSlots int - for _, en := range allEcNodes { - if _, skip := excludeNodes[pb.NewServerAddressFromDataNode(en.Info)]; skip { - excludedFreeSlots += en.FreeEcSlot - glog.V(0).Infof("EC balance excluding node %s: skipped as unreachable by the encode orphan sweep", en.Info.Id) - continue - } - kept = append(kept, en) - } - allEcNodes = kept - totalFreeEcSlots -= excludedFreeSlots - } - - if totalFreeEcSlots < 1 { - return fmt.Errorf("no free ec shard slots. only %d left", totalFreeEcSlots) - } var volumeIdFilter map[uint32]bool if len(volumeIds) > 0 { @@ -824,22 +819,277 @@ func EcBalance(env *Env, collections []string, dc string, ecReplicaPlacement *su } } - ecb := &ecBalancer{ - env: env, - ecNodes: allEcNodes, - replicaPlacement: ecReplicaPlacement, - applyBalancing: applyBalancing, - maxParallelization: maxParallelization, - ioBytePerSecond: ioBytePerSecond, - diskType: diskType, - volumeIds: volumeIdFilter, - migratingVolumeIds: migrating, - } + // The planner's free-slot numbers come from heartbeat snapshots that can lag + // the volume server's own accounting, so a planned target can refuse a copy + // with "no space left" while shards move concurrently (issue #11753). The + // server's refusal is authoritative: re-collect the topology, mark the + // refusing targets no-receive, and plan again. They stay usable as sources. + noReceiveTargets := make(map[string]bool) + noReceiveDisks := make(map[string]map[uint32]bool) + var completed []ecbalancer.Move + var lastErr error + for { + // collect all ec nodes + allEcNodes, totalFreeEcSlots, err := CollectEcNodesForDC(env, dc, diskType) + if err != nil { + return err + } - if len(collections) == 0 { - glog.V(1).Infof("WARNING: No collections to balance EC volumes across.\n") + // Drop excluded nodes (and the slots they contribute) before planning so they + // can be neither a target nor a source for any move this balance plans. + if len(excludeNodes) > 0 { + kept := allEcNodes[:0] + var excludedFreeSlots int + for _, en := range allEcNodes { + if _, skip := excludeNodes[pb.NewServerAddressFromDataNode(en.Info)]; skip { + excludedFreeSlots += en.FreeEcSlot + glog.V(0).Infof("EC balance excluding node %s: skipped as unreachable by the encode orphan sweep", en.Info.Id) + continue + } + kept = append(kept, en) + } + allEcNodes = kept + totalFreeEcSlots -= excludedFreeSlots + } + // A refused destination's slots cannot take a copy, so they do not + // count toward the free-space precheck. + for _, en := range allEcNodes { + if noReceiveTargets[en.Info.Id] { + totalFreeEcSlots -= en.FreeEcSlot + continue + } + for diskId := range noReceiveDisks[en.Info.Id] { + if d := en.Disks[diskId]; d != nil { + totalFreeEcSlots -= d.FreeEcSlots + } + } + } + + if totalFreeEcSlots < 1 { + if lastErr != nil { + return fmt.Errorf("no receiving targets remain: %w", lastErr) + } + return fmt.Errorf("no free ec shard slots. only %d left", totalFreeEcSlots) + } + + // A fresh collection can still describe an applied move through a + // heartbeat that predates its source delete; replay the journal. + applyCompletedMoves(allEcNodes, completed, diskType) + + ecb := &ecBalancer{ + env: env, + ecNodes: allEcNodes, + replicaPlacement: ecReplicaPlacement, + applyBalancing: applyBalancing, + maxParallelization: maxParallelization, + ioBytePerSecond: ioBytePerSecond, + diskType: diskType, + volumeIds: volumeIdFilter, + migratingVolumeIds: migrating, + noReceiveTargets: noReceiveTargets, + noReceiveDisks: noReceiveDisks, + } + + if len(collections) == 0 { + glog.V(1).Infof("WARNING: No collections to balance EC volumes across.\n") + } + err = ecb.balance(collections) + if !applyBalancing { + return err + } + if err == nil { + if lastErr != nil && ecb.plannedMoves == 0 { + return fmt.Errorf("no receiving targets remain: %w", lastErr) + } + return nil + } + + // Replan while the failed attempt surfaced newly-refusing targets; a node + // once marked can no longer be picked, so the loop terminates. Any other + // failure aborts instead of being hidden by the retry. + newRejects := ecb.mergeRejectedTargets(noReceiveTargets, noReceiveDisks) + if len(newRejects) == 0 || ecb.hasOtherError() { + return err + } + lastErr = err + completed = append(completed, ecb.completedMoves...) + fmt.Printf("ec.balance: %v reported no space left; re-collecting topology and re-planning without them as targets\n", newRejects) + } +} + +// applyCompletedMoves replays already-applied moves onto a freshly collected +// node list: heartbeats can lag, so the snapshot may still show a moved shard +// on its deleted source or lack it on its mounted target. +func applyCompletedMoves(ecNodes []*EcNode, moves []ecbalancer.Move, diskType types.DiskType) { + if len(moves) == 0 { + return + } + byID := make(map[string]*EcNode, len(ecNodes)) + for _, en := range ecNodes { + byID[en.Info.Id] = en + } + for _, m := range moves { + vid := needle.VolumeId(m.VolumeID) + shardId := erasure_coding.ShardId(m.ShardID) + if src := byID[m.SourceNode]; src != nil { + removeEcShard(src, vid, shardId) + } + if m.Phase == "dedup" { + continue + } + if dst := byID[m.TargetNode]; dst != nil && !ecNodeHasShard(dst, vid, shardId) { + addEcShard(dst, m.Collection, vid, shardId, m.TargetDisk, diskType) + } + } +} + +func ecNodeHasShard(ecNode *EcNode, vid needle.VolumeId, shardId erasure_coding.ShardId) bool { + for _, diskInfo := range ecNode.Info.DiskInfos { + for _, eci := range diskInfo.EcShardInfos { + if needle.VolumeId(eci.Id) == vid && erasure_coding.ShardBits(eci.EcIndexBits).Has(shardId) { + return true + } + } + } + return false +} + +// removeEcShard drops the shard from the node model, including the per-disk +// bookkeeping of the disk that reported it. +func removeEcShard(en *EcNode, vid needle.VolumeId, shardId erasure_coding.ShardId) { + for _, diskInfo := range en.Info.DiskInfos { + for _, eci := range diskInfo.EcShardInfos { + if needle.VolumeId(eci.Id) != vid || !erasure_coding.ShardBits(eci.EcIndexBits).Has(shardId) { + continue + } + if d := en.Disks[eci.DiskId]; d != nil { + if si := d.EcShards[vid]; si != nil { + si.Delete(shardId) + } + d.EcShardCount-- + d.FreeEcSlots++ + } + } + } + en.DeleteEcVolumeShards(vid, []erasure_coding.ShardId{shardId}) +} + +// addEcShard places the shard on the disk the move used when that disk is +// known; otherwise (TargetDisk 0, server-picked) it falls back to the node's +// disk-type bucket. +func addEcShard(en *EcNode, collection string, vid needle.VolumeId, shardId erasure_coding.ShardId, diskId uint32, diskType types.DiskType) { + d := en.Disks[diskId] + if diskId == 0 || d == nil { + en.AddEcVolumeShards(vid, collection, []erasure_coding.ShardId{shardId}, diskType) + return + } + diskInfo := en.Info.DiskInfos[string(diskType)] + if diskInfo == nil { + diskInfo = &master_pb.DiskInfo{Type: string(diskType)} + en.Info.DiskInfos[string(diskType)] = diskInfo + } + var eci *master_pb.VolumeEcShardInformationMessage + for _, e := range diskInfo.EcShardInfos { + if needle.VolumeId(e.Id) == vid && e.DiskId == diskId { + eci = e + break + } + } + if eci == nil { + eci = &master_pb.VolumeEcShardInformationMessage{ + Id: uint32(vid), + Collection: collection, + DiskId: diskId, + DiskType: string(diskType), + } + diskInfo.EcShardInfos = append(diskInfo.EcShardInfos, eci) + } + si := erasure_coding.ShardsInfoFromVolumeEcShardInformationMessage(eci) + before := si.Count() + si.Set(erasure_coding.NewShardInfo(shardId, 0)) + eci.EcIndexBits = si.Bitmap() + eci.ShardSizes = si.SizesInt64() + added := si.Count() - before + + en.FreeEcSlot -= added + d.EcShardCount += added + d.FreeEcSlots -= added + if d.EcShards == nil { + d.EcShards = make(map[needle.VolumeId]*erasure_coding.ShardsInfo) + } + if d.EcShards[vid] == nil { + d.EcShards[vid] = erasure_coding.NewShardsInfo() + } + d.EcShards[vid].Set(erasure_coding.NewShardInfo(shardId, 0)) +} + +// mergeRejectedTargets folds the destinations this run saw refuse into the +// no-receive sets and returns the newly added ids; an empty result means no +// progress is possible. +func (ecb *ecBalancer) mergeRejectedTargets(noReceiveNodes map[string]bool, noReceiveDisks map[string]map[uint32]bool) (added []string) { + ecb.mu.Lock() + defer ecb.mu.Unlock() + for id := range ecb.rejectedTargets { + if !noReceiveNodes[id] { + noReceiveNodes[id] = true + added = append(added, id) + } + } + for id, disks := range ecb.rejectedDisks { + if noReceiveNodes[id] { + continue + } + if noReceiveDisks[id] == nil { + noReceiveDisks[id] = make(map[uint32]bool) + } + for diskId := range disks { + if !noReceiveDisks[id][diskId] { + noReceiveDisks[id][diskId] = true + added = append(added, fmt.Sprintf("%s:disk%d", id, diskId)) + } + } + } + sort.Strings(added) + return +} + +func (ecb *ecBalancer) hasOtherError() bool { + ecb.mu.Lock() + defer ecb.mu.Unlock() + return ecb.sawOtherErr +} + +// recordMoveResult journals an apply-mode move outcome: a success is replayed +// onto the next attempt's snapshot, a "no space left" marks the target +// no-receive, and any other failure makes the run non-retryable. +func (ecb *ecBalancer) recordMoveResult(m ecbalancer.Move, err error) { + ecb.mu.Lock() + defer ecb.mu.Unlock() + switch { + case err == nil: + ecb.completedMoves = append(ecb.completedMoves, m) + case m.TargetNode != "" && volume_move.IsNoSpaceLeftError(err): + if m.TargetDisk > 0 { + // The refusal convicts only the named disk; the node's other + // disks may still have room. + if ecb.rejectedDisks == nil { + ecb.rejectedDisks = make(map[string]map[uint32]bool) + } + if ecb.rejectedDisks[m.TargetNode] == nil { + ecb.rejectedDisks[m.TargetNode] = make(map[uint32]bool) + } + ecb.rejectedDisks[m.TargetNode][m.TargetDisk] = true + } else { + // The server auto-picked and found no usable disk: the whole + // node is out. + if ecb.rejectedTargets == nil { + ecb.rejectedTargets = make(map[string]bool) + } + ecb.rejectedTargets[m.TargetNode] = true + } + default: + ecb.sawOtherErr = true } - return ecb.balance(collections) } // defaultECRatio resolves a collection's EC data/parity counts, defaulting to @@ -852,7 +1102,7 @@ func defaultECRatio(_ string) (int, int) { // balance plans EC shard moves with the shared planner and executes them. When // collections is empty all collections present are balanced. func (ecb *ecBalancer) balance(collections []string) error { - topo, volumeRatio, selected := toBalancerTopology(ecb.ecNodes, collections, ecb.diskType, ecb.volumeIds, ecb.migratingVolumeIds) + topo, volumeRatio, selected := toBalancerTopology(ecb.ecNodes, collections, ecb.diskType, ecb.volumeIds, ecb.migratingVolumeIds, ecb.noReceiveTargets, ecb.noReceiveDisks) if len(ecb.volumeIds) > 0 { requested := make([]uint32, 0, len(ecb.volumeIds)) for vid := range ecb.volumeIds { @@ -896,6 +1146,7 @@ func (ecb *ecBalancer) balance(collections []string) error { } fmt.Printf("planned %d ec shard move(s) and %d ec shard deletion(s)\n", len(moves)-deletions, deletions) } + ecb.plannedMoves = len(moves) return ecb.executeMoves(moves) } @@ -907,7 +1158,7 @@ func (ecb *ecBalancer) balance(collections []string) error { // (0,0 when unreported, e.g. always in OSS), which Plan prefers over the // collection ratio for mixed-ratio clusters, and the set of volume ids that made // it into the topology. -func toBalancerTopology(ecNodes []*EcNode, collections []string, diskType types.DiskType, volumeIds map[uint32]bool, migratingVolumeIds map[uint32]bool) (*ecbalancer.Topology, func(collection string, vid uint32) (int, int), map[uint32]bool) { +func toBalancerTopology(ecNodes []*EcNode, collections []string, diskType types.DiskType, volumeIds map[uint32]bool, migratingVolumeIds map[uint32]bool, noReceiveTargets map[string]bool, noReceiveDisks map[string]map[uint32]bool) (*ecbalancer.Topology, func(collection string, vid uint32) (int, int), map[uint32]bool) { allowed := make(map[string]bool, len(collections)) for _, c := range collections { allowed[c] = true @@ -924,11 +1175,17 @@ func toBalancerTopology(ecNodes []*EcNode, collections []string, diskType types. for _, en := range ecNodes { rackKey := string(en.DC) + ":" + string(en.Rack) node := topo.AddNode(en.Info.Id, string(en.DC), rackKey, en.FreeEcSlot) + if noReceiveTargets[en.Info.Id] { + node.SetNoReceive() + } // Group by physical machine (host) so shards spread across machines, not just // nodes; the id stays the node identity used for moves. node.SetHost(pb.NewServerAddressFromDataNode(en.Info).ToHost()) for diskId, d := range en.Disks { node.AddDisk(diskId, d.DiskType, d.FreeEcSlots, d.EcShardCount) + if noReceiveDisks[en.Info.Id][diskId] { + node.SetDiskNoReceive(diskId) + } } for diskTypeKey, diskInfo := range en.Info.DiskInfos { if diskInfo == nil { @@ -1069,14 +1326,16 @@ func (ecb *ecBalancer) executeMove(byID map[string]*EcNode, m ecbalancer.Move) e // thing making this safe -- and the plan saying so is not evidence. A // topology entry can name a location holding nothing, and deleting on // that basis removes the last copy. Confirm the keep node has it. - if err := verifyEcShardOnKeepNode(grpcDialOption, m.Collection, vid, m.KeepNode, shardId); err != nil { - return err - } + err := verifyEcShardOnKeepNode(grpcDialOption, m.Collection, vid, m.KeepNode, shardId) addr := pb.NewServerAddressFromDataNode(src.Info) - if err := UnmountEcShards(grpcDialOption, vid, addr, shardIds); err != nil { - return err + if err == nil { + err = UnmountEcShards(grpcDialOption, vid, addr, shardIds) } - return SourceServerDeleteEcShards(grpcDialOption, m.Collection, vid, addr, shardIds) + if err == nil { + err = SourceServerDeleteEcShards(grpcDialOption, m.Collection, vid, addr, shardIds) + } + ecb.recordMoveResult(m, err) + return err } dst := byID[m.TargetNode] @@ -1092,7 +1351,9 @@ func (ecb *ecBalancer) executeMove(byID map[string]*EcNode, m ecbalancer.Move) e // Dry-run: update the in-memory model only. return MoveMountedShardToEcNode(ecb.env, src, m.Collection, vid, shardId, dst, m.TargetDisk, false, ecb.diskType) } - return ecb.applyShardMoveRPC(src, dst, m.Collection, vid, shardId, m.TargetDisk) + err := ecb.applyShardMoveRPC(src, dst, m.Collection, vid, shardId, m.TargetDisk) + ecb.recordMoveResult(m, err) + return err } // applyShardMoveRPC copies a shard to the destination disk, verifies the diff --git a/weed/operation/volume_move/ec_move.go b/weed/operation/volume_move/ec_move.go index bd06d55a3..7ad0d1197 100644 --- a/weed/operation/volume_move/ec_move.go +++ b/weed/operation/volume_move/ec_move.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "io" + "strings" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" @@ -202,6 +203,14 @@ func (m *Mover) DeleteEcShards(ctx context.Context, volumeId needle.VolumeId, co }) } +// IsNoSpaceLeftError matches the volume server's refusal when no local disk +// can take the copied shard ("no space left" from FindEcShardTargetLocation, +// an ENOSPC writing to a caller-pinned disk, and the Rust server's same +// reply), however the move chain wraps it. +func IsNoSpaceLeftError(err error) bool { + return err != nil && strings.Contains(err.Error(), "no space left") +} + // RemoveEcShards unmounts then deletes shards in place — the dedup path for a // shard that already has a copy elsewhere. func (m *Mover) RemoveEcShards(ctx context.Context, volumeId needle.VolumeId, collection string, server pb.ServerAddress, shardIds []erasure_coding.ShardId) error { diff --git a/weed/server/volume_grpc_erasure_coding.go b/weed/server/volume_grpc_erasure_coding.go index 5405fe1e0..91d4bd9f4 100644 --- a/weed/server/volume_grpc_erasure_coding.go +++ b/weed/server/volume_grpc_erasure_coding.go @@ -394,7 +394,7 @@ func (vs *VolumeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_serv } location = vs.store.FindEcShardTargetLocation(req.Collection, needle.VolumeId(req.VolumeId), erasure_coding.DataShardsCount, shardIds...) if location == nil { - return nil, fmt.Errorf("no space left") + return nil, status.Error(codes.ResourceExhausted, "no space left") } } diff --git a/weed/storage/erasure_coding/ecbalancer/balancer.go b/weed/storage/erasure_coding/ecbalancer/balancer.go index 47569e0e9..0c8541fea 100644 --- a/weed/storage/erasure_coding/ecbalancer/balancer.go +++ b/weed/storage/erasure_coding/ecbalancer/balancer.go @@ -38,6 +38,10 @@ type Node struct { rack string // composite rack key (e.g. "dc1:rack1") freeSlots int + // noReceive marks a node that refused shards (e.g. answered "no space + // left"): it stays usable as a move source but is never picked as a target, + // regardless of what its freeSlots say. + noReceive bool disks map[uint32]*disk shards map[volKey]*volumeShards } @@ -47,7 +51,8 @@ type disk struct { diskType string tags []string // placement tags, for preferred-tag tiering freeSlots int - shardCount int // total EC shards on this disk across all volumes + shardCount int // total EC shards on this disk across all volumes + noReceive bool // this disk refused a shard; siblings may still receive } type volumeShards struct { @@ -156,6 +161,39 @@ func (n *Node) SetHost(host string) { } } +// SetNoReceive marks the node ineligible as a move destination. Use when the +// node itself refused shards (e.g. answered "no space left") so the heartbeat's +// free-slot count cannot be trusted; the node can still shed shards as a source. +func (n *Node) SetNoReceive() { + n.noReceive = true +} + +// SetDiskNoReceive marks one disk on the node ineligible as a move destination. +// An explicit-disk copy can be refused while the node's other disks still have +// room; a disk_id 0 refusal means the server searched every disk and is a +// whole-node refusal (SetNoReceive) instead. +func (n *Node) SetDiskNoReceive(diskID uint32) { + if d, ok := n.disks[diskID]; ok { + d.noReceive = true + } +} + +// receiveSlots is the node's receiving capacity: its aggregate free slots with +// refused disks subtracted, and 0 for a refused node. Slots a disk just +// refused are phantom capacity whatever the snapshot reported. +func (n *Node) receiveSlots() int { + if n.noReceive { + return 0 + } + free := n.freeSlots + for _, d := range n.disks { + if d.noReceive { + free -= d.freeSlots + } + } + return max(free, 0) +} + // AddDisk registers a physical disk. shardCount is the disk's total EC shard // count across all volumes (used for disk scoring); freeSlots is the per-disk // free EC shard slots. @@ -296,7 +334,7 @@ func buildRacks(nodes map[string]*Node) map[string]*rack { racks[n.rack] = r } r.nodes[n.id] = n - r.freeSlots += n.freeSlots + r.freeSlots += n.receiveSlots() } return racks } @@ -440,7 +478,7 @@ func planRackTotalCap(vk volKey, racks map[string]*rack, rackShardCount map[stri continue } for _, n := range r.nodes { - room[rk] += min(max(n.freeSlots, 0), max(rp.SameRackCount-volumeShardCount(n, vk), 0)) + room[rk] += min(n.receiveSlots(), max(rp.SameRackCount-volumeShardCount(n, vk), 0)) } } return rackTotalCap(sortedKeys(racks), nil, room, totalShards) @@ -575,12 +613,15 @@ func pickBestNodeForVolumeOnDisk(nodes []*Node, vk volKey, rp *super_block.Repli // metadata is authoritative when present; without it the node's aggregate // stands in so the volume server's own disk selection can place the shard. func freeSlotsForType(n *Node, diskType string) int { + if n.noReceive { + return 0 + } if len(n.disks) == 0 { return n.freeSlots } free := 0 for _, d := range n.disks { - if d.freeSlots > 0 && (diskType == "" || d.diskType == diskType) { + if d.freeSlots > 0 && !d.noReceive && (diskType == "" || d.diskType == diskType) { free += d.freeSlots } } @@ -604,7 +645,7 @@ func pickBestNodeForVolume(nodes []*Node, vk volKey, rp *super_block.ReplicaPlac bestFree := -1 bestScore := -1 for _, node := range nodes { - if node.freeSlots <= 0 { + if node.receiveSlots() <= 0 { continue } count := volumeShardCount(node, vk) @@ -909,7 +950,7 @@ func detectGlobalImbalance(nodes map[string]*Node, racks map[string]*rack, diskT continue } util := float64(count) / float64(capacity) - if util < minUtil && node.freeSlots > 0 { + if util < minUtil && node.receiveSlots() > 0 { minUtil, minCount, minNode = util, count, node } if util > maxUtil { @@ -1091,7 +1132,7 @@ func pickBestDiskOnNode(node *Node, vk volKey, diskType string, shardID, dataSha if diskType != "" && d.diskType != diskType { continue } - if d.freeSlots <= 0 { + if d.freeSlots <= 0 || d.noReceive { continue } diff --git a/weed/storage/erasure_coding/ecbalancer/balancer_test.go b/weed/storage/erasure_coding/ecbalancer/balancer_test.go index 0748452b3..0bb2ddc95 100644 --- a/weed/storage/erasure_coding/ecbalancer/balancer_test.go +++ b/weed/storage/erasure_coding/ecbalancer/balancer_test.go @@ -926,3 +926,67 @@ func TestGlobalPrefersVolumeAbsentFromDestinationMachine(t *testing.T) { t.Errorf("first global move is volume %d, want 200 (vol100 already on the destination machine)", moves[0].volumeID) } } + +// TestNoReceiveNodeNeverTargeted: a node that refused a copy ("no space left") +// must not be picked as a destination no matter how many free slots it reports, +// but it still sheds its own over-cap shards as a source. +func TestNoReceiveNodeNeverTargeted(t *testing.T) { + const vid = uint32(1) + topo := NewTopology() + src := topo.AddNode("n1", "dc1", "dc1:r1", 50) + src.AddShards(vid, "c", 0, bits(0, 1, 2, 3, 4, 5, 6, 7, 8, 9)) + full := topo.AddNode("n2", "dc1", "dc1:r1", 50) + full.AddShards(vid, "c", 0, bits(10, 11, 12, 13)) // all parity, over the 2/node cap + full.SetNoReceive() + topo.AddNode("n3", "dc1", "dc1:r1", 50) + + moves := Plan(topo, Options{Ratio: ratio(10, 4)}) + if len(moves) == 0 { + t.Fatal("expected moves off the over-cap nodes") + } + shed := false + for _, m := range moves { + if m.TargetNode == "n2" { + t.Fatalf("move targets the refusing node: %+v", m) + } + if m.SourceNode == "n2" { + shed = true + } + } + if !shed { + t.Error("the refusing node should still shed its over-cap shards as a source") + } +} + +// TestNoReceiveDiskLeavesSiblingsEligible: an explicit-disk refusal marks only +// that disk, not the node — the planner may still target the node's other +// disks, but never the refused one. +func TestNoReceiveDiskLeavesSiblingsEligible(t *testing.T) { + const vid = uint32(1) + topo := NewTopology() + src := topo.AddNode("n1", "dc1", "dc1:r1", 50) + src.AddShards(vid, "c", 0, bits(0, 1, 2, 3, 4, 5, 6, 7, 8, 9)) + n2 := topo.AddNode("n2", "dc1", "dc1:r1", 50) + // The refused disk scores better (empty); without the noReceive check the + // picker would choose it again. + n2.AddDisk(1, "hdd", 50, 0) + n2.AddDisk(2, "hdd", 50, 50) + n2.AddShards(vid, "c", 2, bits(10, 11, 12, 13)) + n2.SetDiskNoReceive(1) + topo.AddNode("n3", "dc1", "dc1:r1", 50) + + moves := Plan(topo, Options{Ratio: ratio(10, 4)}) + targeted := false + for _, m := range moves { + if m.TargetNode != "n2" { + continue + } + targeted = true + if m.TargetDisk == 1 { + t.Fatalf("move targets the refused disk: %+v", m) + } + } + if !targeted { + t.Fatal("a node with one refused disk should still receive on its other disks") + } +} diff --git a/weed/storage/erasure_coding/ecbalancer/place.go b/weed/storage/erasure_coding/ecbalancer/place.go index 136d1028c..c9d8e438e 100644 --- a/weed/storage/erasure_coding/ecbalancer/place.go +++ b/weed/storage/erasure_coding/ecbalancer/place.go @@ -430,6 +430,9 @@ func rackTotalCap(rackKeys []string, held, room map[string]int, totalShards int) // its free slots, bounded by the per-disk durability cap (maxPerDisk shards of // one volume per disk; <= 0 disables it). func diskShardRoom(n *Node, d *disk, vk volKey, maxPerDisk int) int { + if d.noReceive { + return 0 + } room := d.freeSlots if maxPerDisk > 0 { held := 0 @@ -444,13 +447,16 @@ func diskShardRoom(n *Node, d *disk, vk volKey, maxPerDisk int) int { // nodeShardRoom returns how many more shards of the volume the node's eligible // disks can take, bounded by the node's free slots. func nodeShardRoom(n *Node, vk volKey, eligible func(*disk) bool, maxPerDisk int) int { + if n.noReceive { + return 0 + } room := 0 for _, d := range n.disks { if eligible(d) { room += diskShardRoom(n, d, vk, maxPerDisk) } } - return max(min(room, n.freeSlots), 0) + return max(min(room, n.receiveSlots()), 0) } // rackShardRoom returns how many more shards of the volume the rack can take. @@ -532,7 +538,7 @@ func pickBestDiskEligible(node *Node, vk volKey, eligible func(*disk) bool, pref bestPreferred := false for _, diskID := range sortedDiskKeys(node.disks) { d := node.disks[diskID] - if !eligible(d) || d.freeSlots <= 0 { + if !eligible(d) || d.freeSlots <= 0 || d.noReceive { continue } existingShards := 0 diff --git a/weed/worker/tasks/ec_balance/ec_balance_task.go b/weed/worker/tasks/ec_balance/ec_balance_task.go index b7d4222b8..206c412a0 100644 --- a/weed/worker/tasks/ec_balance/ec_balance_task.go +++ b/weed/worker/tasks/ec_balance/ec_balance_task.go @@ -116,17 +116,29 @@ func (t *ECBalanceTask) Execute(ctx context.Context, params *worker_pb.TaskParam return nil } - err := mover.MoveEcShards(ctx, volume_move.EcShardMove{ + move := volume_move.EcShardMove{ VolumeId: needle.VolumeId(params.VolumeId), Collection: params.Collection, ShardIds: shardIds, Source: sourceAddr, Target: targetAddr, TargetDisk: target.DiskId, - }, volume_move.EcMoveOptions{ + } + opts := volume_move.EcMoveOptions{ IoBytePerSecond: ecParams.GetIoBytePerSecond(), Progress: t.reportProgress, - }) + } + err := mover.MoveEcShards(ctx, move, opts) + // A "no space left" refusal to a pinned disk convicts only that disk; + // retry once letting the server pick any disk on the same node. The + // planner's node choice stands; only the intra-node disk defers to the + // server's live view. When the node is truly full, the retry fails with + // the same error. + if err != nil && target.DiskId > 0 && volume_move.IsNoSpaceLeftError(err) { + glog.Infof("EC balance: %s disk %d has no space left; retrying with server-picked disk", target.Node, target.DiskId) + move.TargetDisk = 0 + err = mover.MoveEcShards(ctx, move, opts) + } if err != nil { return err } diff --git a/weed/worker/tasks/ec_balance/ec_balance_task_test.go b/weed/worker/tasks/ec_balance/ec_balance_task_test.go index 66b3e62ad..00caa3316 100644 --- a/weed/worker/tasks/ec_balance/ec_balance_task_test.go +++ b/weed/worker/tasks/ec_balance/ec_balance_task_test.go @@ -6,6 +6,8 @@ import ( "testing" "github.com/seaweedfs/seaweedfs/weed/pb/worker_pb" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" ) func TestECBalanceTaskRejectsOutOfRangeShardIds(t *testing.T) { @@ -25,3 +27,32 @@ func TestECBalanceTaskRejectsOutOfRangeShardIds(t *testing.T) { t.Fatalf("Execute: expected out-of-range rejection, got: %v", err) } } + +// A refusal against the pinned disk convicts only that disk, not the node: +// the task retries once letting the server pick a disk, and the move completes. +func TestECBalanceTaskRetriesServerPickOnPinnedDiskRefusal(t *testing.T) { + source := startFakeEcVolumeServer(t, 7, 5) + target := startFakeEcVolumeServer(t, 7) + target.refuseCopyOnDisk = 2 + + task := NewECBalanceTask("t2", 7, "c1", + grpc.WithTransportCredentials(insecure.NewCredentials())) + params := &worker_pb.TaskParams{ + VolumeId: 7, + Sources: []*worker_pb.TaskSource{{Node: source.address(), ShardIds: []uint32{5}}}, + Targets: []*worker_pb.TaskTarget{{Node: target.address(), DiskId: 2, ShardIds: []uint32{5}}}, + } + + if err := task.Execute(context.Background(), params); err != nil { + t.Fatalf("Execute: %v", err) + } + if got := target.copyDiskIds; len(got) != 2 || got[0] != 2 || got[1] != 0 { + t.Fatalf("expected copies to disk 2 then server-picked 0, got %v", got) + } + if source.has(7, 5) { + t.Fatalf("source still holds shard 5") + } + if !target.has(7, 5) { + t.Fatalf("target never received shard 5") + } +} diff --git a/weed/worker/tasks/ec_balance/fake_ec_volume_server_test.go b/weed/worker/tasks/ec_balance/fake_ec_volume_server_test.go index efc375209..b3fddd887 100644 --- a/weed/worker/tasks/ec_balance/fake_ec_volume_server_test.go +++ b/weed/worker/tasks/ec_balance/fake_ec_volume_server_test.go @@ -32,6 +32,12 @@ type fakeEcVolumeServer struct { // the request even when it was a no-op against the inventory. deleted []string unmounted []string + // refuseCopyOnDisk, when nonzero, makes VolumeEcShardsCopy fail requests + // pinning that disk id with "no space left" — a full disk; a request with + // DiskId 0 (server picks) still succeeds. + refuseCopyOnDisk uint32 + // copyDiskIds records the DiskId each copy request carried. + copyDiskIds []uint32 // grpcAddr is host:port; the code under test dials by address string. grpcAddr string stop func() @@ -46,11 +52,31 @@ func (f *fakeEcVolumeServer) VolumeEcShardsInfo(ctx context.Context, req *volume VolumeId: req.VolumeId, ShardId: sid, Collection: f.collection, + Size: 1, }) } return resp, nil } +// VolumeEcShardsCopy records the request's DiskId and adds the shards to this +// server's inventory, unless the pinned disk is the configured refusing disk. +func (f *fakeEcVolumeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_server_pb.VolumeEcShardsCopyRequest) (*volume_server_pb.VolumeEcShardsCopyResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.copyDiskIds = append(f.copyDiskIds, req.DiskId) + if f.refuseCopyOnDisk != 0 && req.DiskId == f.refuseCopyOnDisk { + return nil, fmt.Errorf("disk %d: no space left", req.DiskId) + } + for _, sid := range req.ShardIds { + f.shards[req.VolumeId][sid] = true + } + return &volume_server_pb.VolumeEcShardsCopyResponse{}, nil +} + +func (f *fakeEcVolumeServer) VolumeEcShardsMount(ctx context.Context, req *volume_server_pb.VolumeEcShardsMountRequest) (*volume_server_pb.VolumeEcShardsMountResponse, error) { + return &volume_server_pb.VolumeEcShardsMountResponse{}, nil +} + func (f *fakeEcVolumeServer) VolumeEcShardsDelete(ctx context.Context, req *volume_server_pb.VolumeEcShardsDeleteRequest) (*volume_server_pb.VolumeEcShardsDeleteResponse, error) { f.mu.Lock() defer f.mu.Unlock()