ec: re-plan balance moves when a target reports no space left (#11759)

* ecbalancer: add Node.SetNoReceive to exclude a node as a move target

A node that has refused a shard copy at runtime (e.g. 'no space left' on the actual disk, issue #11753) can no longer be trusted as a destination for the remainder of a balance run, but it may still hold shards that must move away. SetNoReceive excludes the node from target selection and from receiving-side capacity accounting, while leaving it eligible as a move source.

* ec: re-plan balance moves when a target reports no space left

Moves are planned from a heartbeat snapshot, but concurrent copies let the volume server's real disk state diverge from it, so a planned target can refuse the copy with 'no space left' mid-run and abort the whole balance (issue #11753). Track the refusing targets during execution, mark them no-receive, re-collect the topology, and plan the remaining moves again. The loop terminates because a marked node can never be picked as a target again.

* volume server: return ResourceExhausted when no disk can take an EC shard

A failed EcShardsCopy carried code=Unknown even though the cause is exhausted local capacity; ResourceExhausted describes the refusal and lets clients distinguish it from internal errors. The message text is kept so message-based matching still works (the Rust volume server reports the same text under Internal).

* ec: replay applied moves onto the re-collected snapshot and surface failures

Addresses review on the re-plan loop: a fresh topology read can still describe an applied move while heartbeats lag, so journal each applied move and replay it onto the next snapshot; retry only when every failure was a 'no space left' refusal (an unrelated failure aborts instead of being hidden); and exclude refused nodes' slots from the free-space precheck so a balance with no remaining receiver fails instead of reporting success for an empty plan.

* ec: carry per-disk state when replaying applied moves

A move that landed on an explicit TargetDisk loses that attribution when replayed onto a stale snapshot, leaving the disk's free slots and shard count unchanged; the next plan could pick the same disk, hit another refusal, and over-exclude the node. Replay the shard onto the target disk's entry and move the disk-level counters, and restore the source disk's counters on removal.

* ecbalancer: add Node.SetDiskNoReceive to exclude a single disk as a target

A copy with an explicit disk_id can be refused while the node's other disks still have room; a disk_id 0 refusal already means the server searched every disk, so the scopes differ. disk.noReceive removes that disk from receiving capacity and disk selection while leaving the node and its sibling disks eligible.

* ec: record explicit-disk refusals as disk-scoped no-receive

recordMoveResult now distinguishes the two refusal scopes: TargetDisk>0 marks only (node, disk), while TargetDisk==0 marks the whole node since the server's auto-pick found no usable disk. The free-slot precheck subtracts refused disk slots, and the replan log names refused destinations as node:diskN.

* ecbalancer: skip no-receive disks in pickBestDiskOnNode

The balance phases pick the destination disk through pickBestDiskOnNode, which ranked disks by score only; a refused disk with a low shard count would be re-picked, fail again, and abort the run since no new destination was rejected.

* volume_move: export IsNoSpaceLeftError for move callers

Both the ec.balance replanner and the worker ec_balance task need to recognize the volume server's no-space refusal; keep one matcher next to the move RPCs instead of a private copy in weed/ec.

* worker: retry pinned-disk EC moves with a server-picked disk on no space

ec_balance detection always pins TargetDisk, so a full disk fails the job even when a sibling disk has room. On a no-space refusal to the pinned disk, retry the move once with TargetDisk 0 so the volume server picks a disk on the same node; the planner's node choice stands.
This commit is contained in:
Chris Lu authored and GitHub committed 2026-10-15 18:47:07 -07:00
1 parent ee2d98f664
commit e6cf190596
10 files changed
+871 -62

No files matched your search

+359
View File
@@ -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)
}
}
+310 -49
View File
@@ -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
+9
View File
@@ -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 {
+1 -1
View File
@@ -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")
}
}
@@ -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
}
@@ -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")
}
}
@@ -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
@@ -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
}
@@ -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")
}
}
@@ -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()