diff --git a/weed/ec/ec_move_runner.go b/weed/ec/ec_move_runner.go new file mode 100644 index 0000000000..912cb6d688 --- /dev/null +++ b/weed/ec/ec_move_runner.go @@ -0,0 +1,141 @@ +package ec + +import ( + "context" + "errors" + "fmt" + "io" + "sync" + "sync/atomic" + + "github.com/seaweedfs/seaweedfs/weed/operation/volume_move" + "github.com/seaweedfs/seaweedfs/weed/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" +) + +// ErrLockLost stops every move that has not started, whatever stopOnError says, +// and fails the whole run. +var ErrLockLost = errors.New("lock is lost") + +// errShardOnTarget marks a failed move whose target registered the shard: only +// the source cleanup failed, so the shard takes a slot on the target. +var errShardOnTarget = errors.New("the target has the shard") + +// EcShardMovesResult counts what RunEcShardMoves did with the planned moves. +type EcShardMovesResult struct { + Moved int + Failed int + NotStarted int + // FailedMoves are the failed moves whose target did not register the shard; + // their shards are still on the source. + FailedMoves []ecbalancer.Move + // LandedMoves are the failed moves whose target has the shard. + LandedMoves []ecbalancer.Move +} + +// RunEcShardMoves applies planned shard moves, up to maxParallelization at once, +// scheduled like ec.balance moves (runMovesSpreadingTargets): never two moves of +// one volume at a time, and the free workers go to the least busy targets. Each +// move copies the shard, verifies the target registered it and only then deletes +// it on the source (volume_move.Mover.MoveEcShards). The in-memory model is not +// touched, so the caller updates it while planning. +// +// With stopOnError no new move starts after the first failure (moves already +// running finish). Otherwise every other move still runs, including the other +// shards of the failed move's volume: unlike ec.balance moves, evacuation moves +// all leave one server for other ones, so no move waits for a slot another frees. +// Losing the admin lock stops the moves that have not started in either case. +// The errors of the failed moves are joined. +func RunEcShardMoves(env *Env, nodes []*EcNode, moves []ecbalancer.Move, maxParallelization int, ioBytePerSecond int64, stopOnError bool, writer io.Writer) (EcShardMovesResult, error) { + byID := make(map[string]*EcNode, len(nodes)) + for _, en := range nodes { + byID[en.Info.Id] = en + } + out := &lockedWriter{w: writer} + mover := volume_move.NewMover(env.GrpcDialOption) + return runEcShardMoves(moves, maxParallelization, stopOnError, func(m ecbalancer.Move) error { + if !env.isLocked() { + return ErrLockLost + } + src, dst := byID[m.SourceNode], byID[m.TargetNode] + if src == nil || dst == nil { + return fmt.Errorf("move ec shard %d.%d %s => %s: unknown volume server", m.VolumeID, m.ShardID, m.SourceNode, m.TargetNode) + } + move := volume_move.EcShardMove{ + VolumeId: needle.VolumeId(m.VolumeID), + Collection: m.Collection, + ShardIds: []erasure_coding.ShardId{erasure_coding.ShardId(m.ShardID)}, + Source: pb.NewServerAddressFromDataNode(src.Info), + Target: pb.NewServerAddressFromDataNode(dst.Info), + TargetDisk: m.TargetDisk, + } + err := mover.MoveEcShards(context.Background(), move, volume_move.EcMoveOptions{IoBytePerSecond: ioBytePerSecond, Writer: out}) + if err != nil { + if mover.VerifyEcShards(context.Background(), move.VolumeId, move.Target, move.ShardIds) == nil { + err = fmt.Errorf("%w: %w", errShardOnTarget, err) + } + return fmt.Errorf("move ec shard %d.%d %s => %s: %w", m.VolumeID, m.ShardID, m.SourceNode, m.TargetNode, err) + } + fmt.Fprintf(out, "moved ec shard %d.%d %s => %s\n", m.VolumeID, m.ShardID, m.SourceNode, m.TargetNode) + return nil + }) +} + +// runEcShardMoves is RunEcShardMoves with the move itself passed in. +func runEcShardMoves(moves []ecbalancer.Move, maxParallelization int, stopOnError bool, move func(ecbalancer.Move) error) (EcShardMovesResult, error) { + var stopped atomic.Bool + var moved, failed atomic.Int64 + var mu sync.Mutex + var errs []error + var failedMoves, landedMoves []ecbalancer.Move + err := runMovesSpreadingTargets(maxParallelization, moves, func(m ecbalancer.Move) error { + if stopped.Load() { + return nil + } + if err := move(m); err != nil { + failed.Add(1) + mu.Lock() + if errors.Is(err, errShardOnTarget) { + landedMoves = append(landedMoves, m) + } else { + failedMoves = append(failedMoves, m) + } + mu.Unlock() + if stopOnError || errors.Is(err, ErrLockLost) { + stopped.Store(true) + return err + } + // Not returned, so the runner keeps the rest of this volume's moves. + mu.Lock() + errs = append(errs, err) + mu.Unlock() + return nil + } + moved.Add(1) + return nil + }) + result := EcShardMovesResult{ + Moved: int(moved.Load()), + Failed: int(failed.Load()), + FailedMoves: failedMoves, + LandedMoves: landedMoves, + } + // Moves skipped after a stop, and moves dropped together with a failed move + // of their volume, did not run. + result.NotStarted = len(moves) - result.Moved - result.Failed + return result, errors.Join(append(errs, err)...) +} + +// lockedWriter serializes the progress lines of concurrent moves. +type lockedWriter struct { + mu sync.Mutex + w io.Writer +} + +func (lw *lockedWriter) Write(p []byte) (int, error) { + lw.mu.Lock() + defer lw.mu.Unlock() + return lw.w.Write(p) +} diff --git a/weed/ec/ec_move_runner_test.go b/weed/ec/ec_move_runner_test.go new file mode 100644 index 0000000000..0b54744ac7 --- /dev/null +++ b/weed/ec/ec_move_runner_test.go @@ -0,0 +1,159 @@ +package ec + +import ( + "errors" + "fmt" + "sync" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding/ecbalancer" +) + +// evacuationMoves is what volumeServer.evacuate plans: every shard of each +// volume leaves the same source node. +func evacuationMoves(volumes, shards int) []ecbalancer.Move { + var moves []ecbalancer.Move + for v := 0; v < volumes; v++ { + for s := 0; s < shards; s++ { + moves = append(moves, ecbalancer.Move{ + VolumeID: uint32(100 + v), + ShardID: s, + SourceNode: "source", + TargetNode: []string{"node-a", "node-b", "node-c"}[(v+s)%3], + }) + } + } + return moves +} + +func TestRunEcShardMovesStopsOnFirstError(t *testing.T) { + moves := evacuationMoves(3, 2) + var ran []ecbalancer.Move + result, err := runEcShardMoves(moves, 1, true, func(m ecbalancer.Move) error { + ran = append(ran, m) + if len(ran) == 2 { + return errors.New("input/output error") + } + return nil + }) + if err == nil { + t.Fatal("expected the failed move's error") + } + if len(ran) != 2 { + t.Fatalf("expected no move to start after the failure, %d ran", len(ran)) + } + if result.Moved != 1 || result.Failed != 1 || result.NotStarted != len(moves)-2 { + t.Fatalf("unexpected result %+v", result) + } +} + +func TestRunEcShardMovesGoesOnAfterFailure(t *testing.T) { + moves := evacuationMoves(3, 3) + broken := moves[0] + ran := map[uint32]int{} + result, err := runEcShardMoves(moves, 1, false, func(m ecbalancer.Move) error { + ran[m.VolumeID]++ + if m.VolumeID == broken.VolumeID && m.ShardID == broken.ShardID { + return errors.New("input/output error") + } + return nil + }) + if err == nil { + t.Fatal("expected the failed move's error") + } + if ran[broken.VolumeID] != 3 { + t.Fatalf("expected the other shards of volume %d to move after its failed shard, %d moves ran", broken.VolumeID, ran[broken.VolumeID]) + } + if result.Moved != len(moves)-1 || result.Failed != 1 || result.NotStarted != 0 { + t.Fatalf("unexpected result %+v", result) + } + if len(result.FailedMoves) != 1 || result.FailedMoves[0] != broken { + t.Fatalf("expected the broken move to be reported, got %+v", result.FailedMoves) + } +} + +func TestRunEcShardMovesInParallel(t *testing.T) { + const parallelization = 4 + moves := evacuationMoves(20, 4) + started := make(chan struct{}, len(moves)) + release := make(chan struct{}) + var mu sync.Mutex + running, most := 0, 0 + var result EcShardMovesResult + var err error + done := make(chan struct{}) + go func() { + defer close(done) + result, err = runEcShardMoves(moves, parallelization, true, func(m ecbalancer.Move) error { + mu.Lock() + running++ + most = max(most, running) + mu.Unlock() + started <- struct{}{} + <-release + mu.Lock() + running-- + mu.Unlock() + return nil + }) + }() + // The first moves are held until parallelization of them have started. + for i := 0; i < parallelization; i++ { + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatalf("only %d moves started at once, expected %d", i, parallelization) + } + } + close(release) + <-done + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if result.Moved != len(moves) || result.Failed != 0 || result.NotStarted != 0 { + t.Fatalf("unexpected result %+v", result) + } + if most != parallelization { + t.Fatalf("%d moves ran at once, expected %d", most, parallelization) + } +} + +func TestRunEcShardMovesStopsWhenLockIsLost(t *testing.T) { + moves := evacuationMoves(3, 2) + ran := 0 + result, err := runEcShardMoves(moves, 1, false, func(m ecbalancer.Move) error { + ran++ + if ran == 2 { + return ErrLockLost + } + return nil + }) + if !errors.Is(err, ErrLockLost) { + t.Fatalf("expected the lost lock to be reported, got %v", err) + } + if ran != 2 || result.Moved != 1 || result.Failed != 1 || result.NotStarted != len(moves)-2 { + t.Fatalf("expected no move after the lock was lost even with skipping on: ran %d, %+v", ran, result) + } +} + +// A move that failed after its target registered the shard counts as failed but +// is not reported for planning again: the target's slot is taken. +func TestRunEcShardMovesKeepsShardsOnTheTargetOutOfFailedMoves(t *testing.T) { + moves := evacuationMoves(2, 1) + result, err := runEcShardMoves(moves, 1, false, func(m ecbalancer.Move) error { + if m.VolumeID == moves[0].VolumeID { + return fmt.Errorf("move ec shard: %w: delete: input/output error", errShardOnTarget) + } + return errors.New("copy: input/output error") + }) + if err == nil { + t.Fatal("expected the failed moves' errors") + } + if result.Failed != 2 || len(result.FailedMoves) != 1 || result.FailedMoves[0] != moves[1] { + t.Fatalf("expected only the move without a target copy in FailedMoves, got %+v", result) + } + if len(result.LandedMoves) != 1 || result.LandedMoves[0] != moves[0] { + t.Fatalf("expected the move with a target copy in LandedMoves, got %+v", result) + } +} diff --git a/weed/shell/command_volume_server_evacuate.go b/weed/shell/command_volume_server_evacuate.go index 287fa10346..464d4b3111 100644 --- a/weed/shell/command_volume_server_evacuate.go +++ b/weed/shell/command_volume_server_evacuate.go @@ -9,8 +9,10 @@ import ( "slices" + "github.com/seaweedfs/seaweedfs/weed/ec" "github.com/seaweedfs/seaweedfs/weed/pb/master_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/super_block" "github.com/seaweedfs/seaweedfs/weed/storage/types" @@ -21,9 +23,15 @@ func init() { } type commandVolumeServerEvacuate struct { - topologyInfo *master_pb.TopologyInfo - targetServer *string - volumeRack *string + topologyInfo *master_pb.TopologyInfo + targetServer *string + volumeRack *string + maxParallelization int + movesPerExec int + movedCount int + limitReported bool + // moveEcShards replaces the RPC moves in tests. + moveEcShards func(moves []ecbalancer.Move, stopOnError bool) (ec.EcShardMovesResult, error) } func (c *commandVolumeServerEvacuate) Name() string { @@ -33,7 +41,7 @@ func (c *commandVolumeServerEvacuate) Name() string { func (c *commandVolumeServerEvacuate) Help() string { return `move out all data on a volume server - volumeServer.evacuate -node + volumeServer.evacuate -node [-skipNonMoveable] [-maxParallelization 1] [-movesPerExec 0] [-noLock] [-apply] This command moves all data away from the volume server. The volumes on the volume servers will be redistributed. @@ -44,6 +52,15 @@ func (c *commandVolumeServerEvacuate) Help() string { good destination to meet the replication requirement. E.g. a volume replication 001 in a cluster with 2 volume servers can not be moved. You can use "-skipNonMoveable" to move the rest volumes. + It also skips EC shards that fail to move (a read error on a failing disk, for + example) and goes on with the rest, the other shards of the same volume included. + + EC shards are moved up to -maxParallelization at a time (default 1), never two + shards of one volume at once. Normal volumes are moved one at a time. + + -movesPerExec limits how many volumes and EC shards one run moves (default 0 for + unlimited), so a big server can be emptied in several runs. + You can use "-noLock" to run without cluster-wide lock. ` @@ -65,12 +82,24 @@ func (c *commandVolumeServerEvacuate) Do(args []string, commandEnv *CommandEnv, applyChangeAlias := vsEvacuateCommand.Bool("force", false, "actually apply the changes (alias for -apply)") retryCount := vsEvacuateCommand.Int("retry", 0, "how many times to retry") noLock := vsEvacuateCommand.Bool("noLock", false, "do not lock the admin shell at one's own risk") + maxParallelization := vsEvacuateCommand.Int("maxParallelization", 1, "run up to X EC shard moves in parallel") + movesPerExec := vsEvacuateCommand.Int("movesPerExec", 0, "how many volumes and EC shards to move in one run (default is 0 for unlimited)") if err = vsEvacuateCommand.Parse(args); err != nil { if errors.Is(err, flag.ErrHelp) { return nil } return err } + if *maxParallelization < 1 { + return fmt.Errorf("maxParallelization must be >= 1") + } + if *movesPerExec < 0 { + return fmt.Errorf("movesPerExec must be >= 0") + } + c.maxParallelization = *maxParallelization + c.movesPerExec = *movesPerExec + c.movedCount = 0 + c.limitReported = false handleDeprecatedForceFlag(writer, vsEvacuateCommand, applyChangeAlias, applyChange) infoAboutSimulationMode(writer, *applyChange, "-apply") @@ -149,10 +178,16 @@ func (c *commandVolumeServerEvacuate) evacuateNormalVolumes(commandEnv *CommandE // this disk's VolumeInfos in place. Iterate over a snapshot so removals // don't leave nil holes in the slice we are ranging over. for _, vol := range slices.Clone(diskInfo.VolumeInfos) { + if c.limitReached(writer) { + return nil + } hasMoved, err := moveAwayOneNormalVolume(commandEnv, volumeReplicas, vol, thisNode, otherNodes, applyChange) if err != nil { fmt.Fprintf(writer, "move away volume %d from %s: %v\n", vol.Id, volumeServer, err) } + if hasMoved { + c.movedCount++ + } if !hasMoved { if skipNonMoveable { replicaPlacement, _ := super_block.NewReplicaPlacementFromByte(byte(vol.ReplicaPlacement)) @@ -167,6 +202,32 @@ func (c *commandVolumeServerEvacuate) evacuateNormalVolumes(commandEnv *CommandE return nil } +// ecShardOnServer names one EC shard on one source server. +type ecShardOnServer struct { + server string + vid uint32 + shardId int +} + +// plannedEcMove is a shard move together with what the plan changed in the +// in-memory model, so a failed move can be taken back out of it. +type plannedEcMove struct { + move ecbalancer.Move + src, dst *EcNode + diskType types.DiskType + dstHadShard bool + // dstSlots are the free slots the plan took from dst. + dstSlots int +} + +// ecEvacuationGroup holds the EC nodes of one disk type, collected once so the +// model keeps its updates across planning rounds. +type ecEvacuationGroup struct { + diskType types.DiskType + thisNodes []*EcNode + otherNodes []*EcNode +} + func (c *commandVolumeServerEvacuate) evacuateEcVolumes(commandEnv *CommandEnv, volumeServer string, skipNonMoveable, applyChange bool, writer io.Writer) error { // Evacuate EC volumes for all disk types discovered from topology // Disk types are free-form tags (e.g., "", "hdd", "ssd", "nvme", etc.) @@ -174,43 +235,146 @@ func (c *commandVolumeServerEvacuate) evacuateEcVolumes(commandEnv *CommandEnv, // We collect topology once at the start and track capacity changes ourselves // (via freeEcSlot decrement after each move) rather than repeatedly refreshing, // which would give a false sense of correctness since topology could be stale. + // + // Moves are planned first, updating the in-memory model as the old + // one-move-at-a-time loop did, then applied up to -maxParallelization at a time. + // A failed move gives its target back to the model; shards that found no + // target in the plan are then planned again, as the old loop would have + // placed them on the room the failed move did not use. diskTypes := collectVolumeDiskTypes(c.topologyInfo) + var groups []ecEvacuationGroup + var allNodes []*EcNode for _, diskType := range diskTypes { ecNodes, _ := collectEcVolumeServersByDc(c.topologyInfo, "", diskType) + allNodes = append(allNodes, ecNodes...) thisNodes, otherNodes := c.ecNodesOtherThan(ecNodes, volumeServer) if len(thisNodes) == 0 { // This server doesn't have EC shards for this disk type, skip continue } + groups = append(groups, ecEvacuationGroup{diskType: diskType, thisNodes: thisNodes, otherNodes: otherNodes}) + } - // move away ec volumes for this disk type - for _, thisNode := range thisNodes { - diskInfo, found := thisNode.Info.DiskInfos[string(diskType)] + source := volumeServer + if source == "" { + source = "rack " + *c.volumeRack + } + failed := make(map[ecShardOnServer]bool) + // Each round moves or fails every shard it plans, and a failed shard is not + // planned again, so the rounds end. + for { + planned, unplanned, planErr := c.planEcEvacuation(groups, volumeServer, skipNonMoveable, failed, writer) + reportUnplanned := func() { + for _, vid := range unplanned { + fmt.Fprintf(writer, "failed to move away ec volume %d from %s\n", vid, source) + } + } + if !applyChange { + c.movedCount += len(planned) + reportUnplanned() + c.limitReached(writer) + return planErr + } + if len(planned) == 0 { + reportUnplanned() + return planErr + } + + moves := make([]ecbalancer.Move, len(planned)) + byShard := make(map[ecShardOnServer]plannedEcMove, len(planned)) + for i, p := range planned { + moves[i] = p.move + byShard[ecShardOnServer{p.move.SourceNode, p.move.VolumeID, p.move.ShardID}] = p + } + fmt.Fprintf(writer, "moving %d ec shard(s) from %s, up to %d at a time\n", len(moves), source, c.maxParallelization) + result, err := c.runEcShardMoves(commandEnv, allNodes, moves, !skipNonMoveable, writer) + c.movedCount += result.Moved + fmt.Fprintf(writer, "ec shards from %s: %d moved, %d failed, %d not started\n", source, result.Moved, result.Failed, result.NotStarted) + if err != nil { + // A lost lock stopped the moves that had not started: fail the run, + // even with -skipNonMoveable. + if !skipNonMoveable || errors.Is(err, ec.ErrLockLost) { + return err + } + fmt.Fprintf(writer, "skipped ec shards that failed to move: %v\n", err) + } + // The shard of a failed move is still on its source, and its target did + // not receive it. A target that had the shard before reports it after + // any failure, and took no new slot for it. + undo := append([]ecbalancer.Move(nil), result.FailedMoves...) + for _, m := range result.LandedMoves { + if byShard[ecShardOnServer{m.SourceNode, m.VolumeID, m.ShardID}].dstHadShard { + undo = append(undo, m) + } + } + for _, m := range undo { + key := ecShardOnServer{m.SourceNode, m.VolumeID, m.ShardID} + if p, found := byShard[key]; found { + unplanEcMove(p) + failed[key] = true + } + } + if planErr != nil { + return planErr + } + limitReached := c.limitReached(writer) + if len(undo) == 0 || len(unplanned) == 0 || limitReached { + reportUnplanned() + return nil + } + fmt.Fprintf(writer, "planning again %d ec volume(s) that found no destination, now that failed moves left room\n", len(unplanned)) + } +} + +// planEcEvacuation plans the moves of the EC shards still on the evacuated +// servers, leaving out the shards whose move failed before. unplanned lists the +// volumes with a shard that found no destination; without -skipNonMoveable the +// first one ends the plan with planErr, keeping the moves planned before it. +func (c *commandVolumeServerEvacuate) planEcEvacuation(groups []ecEvacuationGroup, volumeServer string, skipNonMoveable bool, failed map[ecShardOnServer]bool, writer io.Writer) (planned []plannedEcMove, unplanned []uint32, planErr error) { + for _, group := range groups { + for _, thisNode := range group.thisNodes { + diskInfo, found := thisNode.Info.DiskInfos[string(group.diskType)] if !found { continue } for _, ecShardInfo := range diskInfo.EcShardInfos { - hasMoved, err := c.moveAwayOneEcVolume(commandEnv, ecShardInfo, thisNode, otherNodes, applyChange, diskType, writer) - if err != nil { - fmt.Fprintf(writer, "move away volume %d from %s: %v\n", ecShardInfo.Id, volumeServer, err) - } - if !hasMoved { - if skipNonMoveable { - fmt.Fprintf(writer, "failed to move away ec volume %d from %s\n", ecShardInfo.Id, volumeServer) - } else { - return fmt.Errorf("failed to move away ec volume %d from %s", ecShardInfo.Id, volumeServer) + limit := 0 + if c.movesPerExec > 0 { + limit = c.movesPerExec - c.movedCount - len(planned) + if limit <= 0 { + break } } + volumeMoves, ok := c.planMoveAwayOneEcVolume(ecShardInfo, thisNode, group.otherNodes, group.diskType, limit, failed, writer) + planned = append(planned, volumeMoves...) + if !ok { + if !skipNonMoveable { + // Like the one-at-a-time loop did, still move what comes before it. + return planned, nil, fmt.Errorf("failed to move away ec volume %d from %s", ecShardInfo.Id, volumeServer) + } + unplanned = append(unplanned, ecShardInfo.Id) + } } } } - return nil + return planned, unplanned, nil } -func (c *commandVolumeServerEvacuate) moveAwayOneEcVolume(commandEnv *CommandEnv, ecShardInfo *master_pb.VolumeEcShardInformationMessage, thisNode *EcNode, otherNodes []*EcNode, applyChange bool, diskType types.DiskType, writer io.Writer) (hasMoved bool, err error) { +// planMoveAwayOneEcVolume picks a destination for each shard of this EC volume on +// thisNode, at most limit shards when limit > 0, and records the moves in the +// in-memory model. Shards whose move failed before are left where they are. ok +// is false when a shard has no destination; the shards planned before it are +// still returned. +func (c *commandVolumeServerEvacuate) planMoveAwayOneEcVolume(ecShardInfo *master_pb.VolumeEcShardInformationMessage, thisNode *EcNode, otherNodes []*EcNode, diskType types.DiskType, limit int, failed map[ecShardOnServer]bool, writer io.Writer) (planned []plannedEcMove, ok bool) { si := erasure_coding.ShardsInfoFromVolumeEcShardInformationMessage(ecShardInfo) for _, shardId := range si.Ids() { + if limit > 0 && len(planned) >= limit { + break + } + if failed[ecShardOnServer{thisNode.Info.Id, ecShardInfo.Id, int(shardId)}] { + continue + } // Sort by: 1) fewest shards of this volume, 2) most free EC slots // This ensures we prefer nodes with capacity and balanced shard distribution slices.SortFunc(otherNodes, func(a, b *EcNode) int { @@ -222,7 +386,7 @@ func (c *commandVolumeServerEvacuate) moveAwayOneEcVolume(commandEnv *CommandEnv return b.FreeEcSlot - a.FreeEcSlot // Then prefer more free slots }) - shardMoved := false + shardPlanned := false skippedNodes := 0 for i := 0; i < len(otherNodes); i++ { emptyNode := otherNodes[i] @@ -246,30 +410,90 @@ func (c *commandVolumeServerEvacuate) moveAwayOneEcVolume(commandEnv *CommandEnv } else { fmt.Fprintf(writer, "moving ec volume %s%d.%d %s => %s\n", collectionPrefix, ecShardInfo.Id, shardId, thisNode.Info.Id, emptyNode.Info.Id) } - err = moveMountedShardToEcNode(commandEnv, thisNode, ecShardInfo.Collection, vid, shardId, emptyNode, destDiskId, applyChange, diskType) - if err != nil { - hasMoved = false - return - } else { - hasMoved = true - shardMoved = true - // Update the node's free slot count after successful move - emptyNode.FreeEcSlot-- - break - } + shardIds := []erasure_coding.ShardId{shardId} + dstHadShard := ecNodeHasShard(emptyNode, ecShardInfo.Id, shardId) + dstFree := emptyNode.FreeEcSlot + emptyNode.AddEcVolumeShards(vid, ecShardInfo.Collection, shardIds, diskType) + thisNode.DeleteEcVolumeShards(vid, shardIds) + // Update the node's free slot count after the planned move + emptyNode.FreeEcSlot-- + planned = append(planned, plannedEcMove{ + move: ecbalancer.Move{ + VolumeID: ecShardInfo.Id, + ShardID: int(shardId), + Collection: ecShardInfo.Collection, + SourceNode: thisNode.Info.Id, + TargetNode: emptyNode.Info.Id, + TargetDisk: destDiskId, + }, + src: thisNode, + dst: emptyNode, + diskType: diskType, + dstHadShard: dstHadShard, + dstSlots: dstFree - emptyNode.FreeEcSlot, + }) + shardPlanned = true + break } - if !shardMoved { + if !shardPlanned { if skippedNodes > 0 { fmt.Fprintf(writer, "no available destination for ec shard %d.%d: %d nodes have no free slots\n", ecShardInfo.Id, shardId, skippedNodes) } // Ensure partial moves are reported as failures to prevent data loss - hasMoved = false - return + return planned, false } } - return + return planned, true +} + +// unplanEcMove takes a failed move back out of the in-memory model: the shard is +// on its source again and the target's slot is free. +func unplanEcMove(p plannedEcMove) { + vid := needle.VolumeId(p.move.VolumeID) + shardIds := []erasure_coding.ShardId{erasure_coding.ShardId(p.move.ShardID)} + free := p.dst.FreeEcSlot + p.dstSlots + if !p.dstHadShard { + p.dst.DeleteEcVolumeShards(vid, shardIds) + } + p.dst.FreeEcSlot = free + p.src.AddEcVolumeShards(vid, p.move.Collection, shardIds, p.diskType) +} + +// ecNodeHasShard reports whether the in-memory model has the shard on the node. +func ecNodeHasShard(node *EcNode, vid uint32, shardId erasure_coding.ShardId) bool { + for _, diskInfo := range node.Info.DiskInfos { + if diskInfo == nil { + continue + } + for _, eci := range diskInfo.EcShardInfos { + if eci.Id == vid && erasure_coding.ShardsInfoFromVolumeEcShardInformationMessage(eci).Has(shardId) { + return true + } + } + } + return false +} + +// runEcShardMoves applies the planned moves; tests replace it through moveEcShards. +func (c *commandVolumeServerEvacuate) runEcShardMoves(commandEnv *CommandEnv, nodes []*EcNode, moves []ecbalancer.Move, stopOnError bool, writer io.Writer) (ec.EcShardMovesResult, error) { + if c.moveEcShards != nil { + return c.moveEcShards(moves, stopOnError) + } + return ec.RunEcShardMoves(commandEnv.ecEnv(), nodes, moves, c.maxParallelization, 0, stopOnError, writer) +} + +// limitReached reports whether -movesPerExec moves have been made, and says so once. +func (c *commandVolumeServerEvacuate) limitReached(writer io.Writer) bool { + if c.movesPerExec == 0 || c.movedCount < c.movesPerExec { + return false + } + if !c.limitReported { + c.limitReported = true + fmt.Fprintf(writer, "reached -movesPerExec=%d, stopping; run the command again to move the rest\n", c.movesPerExec) + } + return true } func moveAwayOneNormalVolume(commandEnv *CommandEnv, volumeReplicas map[uint32][]*VolumeReplica, vol *master_pb.VolumeInformationMessage, thisNode *Node, otherNodes []*Node, applyChange bool) (hasMoved bool, err error) { diff --git a/weed/shell/command_volume_server_evacuate_test.go b/weed/shell/command_volume_server_evacuate_test.go index 7621cf14a2..bc048efaab 100644 --- a/weed/shell/command_volume_server_evacuate_test.go +++ b/weed/shell/command_volume_server_evacuate_test.go @@ -1,12 +1,18 @@ package shell import ( + "bytes" + "errors" "io" "os" "strings" "testing" "github.com/seaweedfs/seaweedfs/weed/cluster" + "github.com/seaweedfs/seaweedfs/weed/ec" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding/ecbalancer" "github.com/seaweedfs/seaweedfs/weed/wdclient/exclusive_locks" ) @@ -87,3 +93,376 @@ func TestVolumeServerEvacuateHelpIsNotAnError(t *testing.T) { t.Fatalf("-h should print usage without failing, got: %v", err) } } + +func newEvacuateCommandForTest(movesPerExec int) *commandVolumeServerEvacuate { + targetServer := "" + volumeRack := "" + c := &commandVolumeServerEvacuate{ + targetServer: &targetServer, + volumeRack: &volumeRack, + maxParallelization: 1, + movesPerExec: movesPerExec, + } + c.topologyInfo = parseOutput(topoDataEc) + return c +} + +// ecShardCount counts the EC shards a volume server holds in the topology. +func ecShardCount(topo *master_pb.TopologyInfo, volumeServer string) (shards int) { + for _, dc := range topo.DataCenterInfos { + for _, rack := range dc.RackInfos { + for _, dn := range rack.DataNodeInfos { + if dn.Id != volumeServer { + continue + } + for _, disk := range dn.DiskInfos { + for _, eci := range disk.EcShardInfos { + shards += erasure_coding.ShardsInfoFromVolumeEcShardInformationMessage(eci).Count() + } + } + } + } + } + return shards +} + +func TestVolumeServerEvacuateEcPlansEveryShard(t *testing.T) { + c := newEvacuateCommandForTest(0) + volumeServer := "172.19.0.10:8702" + shards := ecShardCount(c.topologyInfo, volumeServer) + var out bytes.Buffer + if err := c.evacuateEcVolumes(nil, volumeServer, false, false, &out); err != nil { + t.Fatalf("evacuate: %v", err) + } + planned := strings.Count(out.String(), "moving ec volume") + if shards == 0 || planned != shards || c.movedCount != shards { + t.Fatalf("%s holds %d ec shards, planned %d moves, counted %d", volumeServer, shards, planned, c.movedCount) + } + if left := ecShardCount(c.topologyInfo, volumeServer); left != 0 { + t.Fatalf("the model still has %d ec shards on %s after the plan", left, volumeServer) + } + for _, line := range strings.Split(strings.TrimSpace(out.String()), "\n") { + if strings.HasPrefix(line, "moving ec volume") && strings.Fields(line)[6] == volumeServer { + t.Fatalf("a shard is planned onto the evacuated server: %s", line) + } + } +} + +func TestVolumeServerEvacuateEcMovesPerExec(t *testing.T) { + all := newEvacuateCommandForTest(0) + volumeServer := "172.19.0.10:8702" + if err := all.evacuateEcVolumes(nil, volumeServer, false, false, io.Discard); err != nil { + t.Fatalf("evacuate: %v", err) + } + if all.movedCount < 3 { + t.Fatalf("the test needs a server with at least 3 shards, it has %d", all.movedCount) + } + + c := newEvacuateCommandForTest(2) + var out bytes.Buffer + if err := c.evacuateEcVolumes(nil, volumeServer, false, false, &out); err != nil { + t.Fatalf("evacuate: %v", err) + } + if planned := strings.Count(out.String(), "moving ec volume"); planned != 2 || c.movedCount != 2 { + t.Fatalf("expected 2 planned shard moves with -movesPerExec=2, got %d (counted %d)", planned, c.movedCount) + } + if !strings.Contains(out.String(), "reached -movesPerExec=2") { + t.Fatalf("expected the limit to be reported, got:\n%s", out.String()) + } +} + +// withEcShards gives an EC volume on a volume server the listed shards. The +// volume.list parser keeps only the first shard of a "shards:[5 11 12]" entry. +func withEcShards(topo *master_pb.TopologyInfo, volumeServer string, vid uint32, shardIds ...erasure_coding.ShardId) bool { + for _, dc := range topo.DataCenterInfos { + for _, rack := range dc.RackInfos { + for _, dn := range rack.DataNodeInfos { + if dn.Id != volumeServer { + continue + } + for _, disk := range dn.DiskInfos { + for _, eci := range disk.EcShardInfos { + if eci.Id != vid { + continue + } + si := erasure_coding.NewShardsInfo() + for _, id := range shardIds { + si.Set(erasure_coding.NewShardInfo(id, 0)) + } + eci.EcIndexBits = si.Bitmap() + eci.ShardSizes = si.SizesInt64() + return true + } + } + } + } + } + return false +} + +// Every limit, including the ones that end in the middle of a volume's shards, +// plans exactly that many moves. +func TestVolumeServerEvacuateEcMovesPerExecCutsInsideVolume(t *testing.T) { + volumeServer := "172.19.0.20:8706" + newCommand := func(movesPerExec int) *commandVolumeServerEvacuate { + c := newEvacuateCommandForTest(movesPerExec) + if !withEcShards(c.topologyInfo, volumeServer, 14322, 5, 11, 12) { + t.Fatalf("ec volume 14322 is not on %s", volumeServer) + } + return c + } + shards := ecShardCount(newCommand(0).topologyInfo, volumeServer) + if shards != 6 { + t.Fatalf("expected 6 ec shards on %s, found %d", volumeServer, shards) + } + for limit := 1; limit <= shards; limit++ { + c := newCommand(limit) + var out bytes.Buffer + if err := c.evacuateEcVolumes(nil, volumeServer, false, false, &out); err != nil { + t.Fatalf("limit %d: %v", limit, err) + } + if planned := strings.Count(out.String(), "moving ec volume"); planned != limit || c.movedCount != limit { + t.Fatalf("limit %d: planned %d moves, counted %d", limit, planned, c.movedCount) + } + } + c := newCommand(0) + if err := c.evacuateEcVolumes(nil, volumeServer, false, false, io.Discard); err != nil { + t.Fatalf("no limit: %v", err) + } + if c.movedCount != shards || ecShardCount(c.topologyInfo, volumeServer) != 0 { + t.Fatalf("no limit: planned %d of %d shards", c.movedCount, shards) + } +} + +// tightTargetTopology has a source holding shard 0 of volumes 1 and 2, and one +// target with room for a single planned shard: two free slots, and the plan +// takes two per shard. +func tightTargetTopology() *master_pb.TopologyInfo { + ecShard := func(vid uint32, shardIds ...erasure_coding.ShardId) *master_pb.VolumeEcShardInformationMessage { + si := erasure_coding.NewShardsInfo() + for _, id := range shardIds { + si.Set(erasure_coding.NewShardInfo(id, 0)) + } + return &master_pb.VolumeEcShardInformationMessage{Id: vid, Collection: "c", EcIndexBits: si.Bitmap(), ShardSizes: si.SizesInt64()} + } + node := func(id string, maxVolumes int64, shards ...*master_pb.VolumeEcShardInformationMessage) *master_pb.DataNodeInfo { + return &master_pb.DataNodeInfo{Id: id, DiskInfos: map[string]*master_pb.DiskInfo{ + "": {MaxVolumeCount: maxVolumes, EcShardInfos: shards}, + }} + } + source := node("source:8080", 10, ecShard(1, 0), ecShard(2, 0)) + // 1 volume slot = 10 shard slots, 8 of them taken by volume 99. + target := node("target:8080", 1, ecShard(99, 0, 1, 2, 3, 4, 5, 6, 7)) + return &master_pb.TopologyInfo{DataCenterInfos: []*master_pb.DataCenterInfo{{ + Id: "dc1", + RackInfos: []*master_pb.RackInfo{ + {Id: "rack1", DataNodeInfos: []*master_pb.DataNodeInfo{source}}, + {Id: "rack2", DataNodeInfos: []*master_pb.DataNodeInfo{target}}, + }, + }}} +} + +// A shard that fails to move must not keep the room it was planned on: the +// readable shard that found no destination is planned again onto it. +func TestVolumeServerEvacuateEcFailedMoveFreesItsTarget(t *testing.T) { + c := newEvacuateCommandForTest(0) + c.topologyInfo = tightTargetTopology() + var ran []ecbalancer.Move + c.moveEcShards = func(moves []ecbalancer.Move, stopOnError bool) (ec.EcShardMovesResult, error) { + var result ec.EcShardMovesResult + for _, m := range moves { + ran = append(ran, m) + if m.VolumeID == 1 { + result.Failed++ + result.FailedMoves = append(result.FailedMoves, m) + continue + } + result.Moved++ + } + if result.Failed > 0 { + return result, errors.New("input/output error") + } + return result, nil + } + var out bytes.Buffer + if err := c.evacuateEcVolumes(nil, "source:8080", true, true, &out); err != nil { + t.Fatalf("evacuate: %v", err) + } + if len(ran) != 2 || ran[0].VolumeID != 1 || ran[1].VolumeID != 2 { + t.Fatalf("expected volume 1 to fail and volume 2 to move after it, ran %+v\n%s", ran, out.String()) + } + if c.movedCount != 1 { + t.Fatalf("expected 1 moved shard, counted %d", c.movedCount) + } + if strings.Contains(out.String(), "failed to move away ec volume 2") { + t.Fatalf("volume 2 should have found a destination:\n%s", out.String()) + } +} + +// When every move fails, planning again ends once no shard is left to plan. +func TestVolumeServerEvacuateEcEndsWhenEveryMoveFails(t *testing.T) { + c := newEvacuateCommandForTest(0) + c.topologyInfo = tightTargetTopology() + rounds := 0 + c.moveEcShards = func(moves []ecbalancer.Move, stopOnError bool) (ec.EcShardMovesResult, error) { + rounds++ + return ec.EcShardMovesResult{Failed: len(moves), FailedMoves: moves}, errors.New("input/output error") + } + var out bytes.Buffer + if err := c.evacuateEcVolumes(nil, "source:8080", true, true, &out); err != nil { + t.Fatalf("evacuate: %v", err) + } + // Round 1 fails volume 1, round 2 fails volume 2; nothing is left to plan. + if rounds != 2 { + t.Fatalf("expected 2 rounds, got %d:\n%s", rounds, out.String()) + } + if c.movedCount != 0 { + t.Fatalf("expected nothing moved, counted %d", c.movedCount) + } +} + +// A lost lock fails the run even with -skipNonMoveable, so the caller does not +// take an unfinished evacuation for a finished one. +func TestVolumeServerEvacuateEcLostLockFailsTheRun(t *testing.T) { + c := newEvacuateCommandForTest(0) + c.topologyInfo = tightTargetTopology() + c.moveEcShards = func(moves []ecbalancer.Move, stopOnError bool) (ec.EcShardMovesResult, error) { + return ec.EcShardMovesResult{Failed: 1, NotStarted: len(moves) - 1, FailedMoves: moves[:1]}, errors.Join(ec.ErrLockLost) + } + err := c.evacuateEcVolumes(nil, "source:8080", true, true, io.Discard) + if !errors.Is(err, ec.ErrLockLost) { + t.Fatalf("expected the lost lock to fail the run, got %v", err) + } +} + +// An applied run that stops at -movesPerExec says so, so the operator runs it again. +func TestVolumeServerEvacuateEcReportsMovesPerExecWhenApplied(t *testing.T) { + c := newEvacuateCommandForTest(2) + c.moveEcShards = func(moves []ecbalancer.Move, stopOnError bool) (ec.EcShardMovesResult, error) { + return ec.EcShardMovesResult{Moved: len(moves)}, nil + } + var out bytes.Buffer + if err := c.evacuateEcVolumes(nil, "172.19.0.10:8702", false, true, &out); err != nil { + t.Fatalf("evacuate: %v", err) + } + if c.movedCount != 2 || !strings.Contains(out.String(), "reached -movesPerExec=2") { + t.Fatalf("expected 2 moves and the limit reported, counted %d:\n%s", c.movedCount, out.String()) + } +} + +// Every round moves or drops each planned shard, so planning again goes on for as +// long as failed moves leave room: a readable shard behind many unreadable ones +// still moves. +func TestVolumeServerEvacuateEcPlansAgainUntilNoRoomIsLeft(t *testing.T) { + c := newEvacuateCommandForTest(0) + c.topologyInfo = tightTargetTopology() + const unreadable = 12 + source := c.topologyInfo.DataCenterInfos[0].RackInfos[0].DataNodeInfos[0] + var shards []*master_pb.VolumeEcShardInformationMessage + for vid := uint32(1); vid <= unreadable+1; vid++ { + si := erasure_coding.NewShardsInfo() + si.Set(erasure_coding.NewShardInfo(0, 0)) + shards = append(shards, &master_pb.VolumeEcShardInformationMessage{Id: vid, Collection: "c", EcIndexBits: si.Bitmap(), ShardSizes: si.SizesInt64()}) + } + source.DiskInfos[""].EcShardInfos = shards + var moved []uint32 + c.moveEcShards = func(moves []ecbalancer.Move, stopOnError bool) (result ec.EcShardMovesResult, err error) { + for _, m := range moves { + if m.VolumeID <= unreadable { + result.Failed++ + result.FailedMoves = append(result.FailedMoves, m) + err = errors.New("input/output error") + continue + } + result.Moved++ + moved = append(moved, m.VolumeID) + } + return result, err + } + var out bytes.Buffer + if err := c.evacuateEcVolumes(nil, "source:8080", true, true, &out); err != nil { + t.Fatalf("evacuate: %v", err) + } + if len(moved) != 1 || moved[0] != unreadable+1 { + t.Fatalf("expected volume %d to move after %d failed ones, moved %v:\n%s", unreadable+1, unreadable, moved, out.String()) + } +} + +// A failed move gives back every slot the plan took from its target, also when +// the target holds the shard in another disk type, which the undo leaves alone. +func TestVolumeServerEvacuateEcFailedMoveGivesBackItsSlots(t *testing.T) { + c := newEvacuateCommandForTest(0) + c.topologyInfo = tightTargetTopology() + target := c.topologyInfo.DataCenterInfos[0].RackInfos[1].DataNodeInfos[0] + // One free slot: volume 99 takes 9 of the 10. + disk := target.DiskInfos[""] + si := erasure_coding.ShardsInfoFromVolumeEcShardInformationMessage(disk.EcShardInfos[0]) + si.Set(erasure_coding.NewShardInfo(8, 0)) + disk.EcShardInfos[0].EcIndexBits, disk.EcShardInfos[0].ShardSizes = si.Bitmap(), si.SizesInt64() + ssd := erasure_coding.NewShardsInfo() + ssd.Set(erasure_coding.NewShardInfo(0, 0)) + target.DiskInfos["ssd"] = &master_pb.DiskInfo{Type: "ssd", EcShardInfos: []*master_pb.VolumeEcShardInformationMessage{ + {Id: 1, Collection: "c", DiskType: "ssd", EcIndexBits: ssd.Bitmap(), ShardSizes: ssd.SizesInt64()}, + }} + var moved []uint32 + c.moveEcShards = func(moves []ecbalancer.Move, stopOnError bool) (result ec.EcShardMovesResult, err error) { + for _, m := range moves { + if m.VolumeID == 1 { + result.Failed++ + result.FailedMoves = append(result.FailedMoves, m) + err = errors.New("input/output error") + continue + } + result.Moved++ + moved = append(moved, m.VolumeID) + } + return result, err + } + var out bytes.Buffer + if err := c.evacuateEcVolumes(nil, "source:8080", true, true, &out); err != nil { + t.Fatalf("evacuate: %v", err) + } + if len(moved) != 1 || moved[0] != 2 { + t.Fatalf("expected volume 2 to take the slot volume 1 gave back, moved %v:\n%s", moved, out.String()) + } + if erasure_coding.ShardsInfoFromVolumeEcShardInformationMessage(target.DiskInfos["ssd"].EcShardInfos[0]).Count() != 1 { + t.Fatal("the undo removed the shard the target already had") + } +} + +// A failed move onto a target that already had the shard gives its slot back, +// although the target reports the shard after the failure. +func TestVolumeServerEvacuateEcFailedMoveOntoAnExistingShardGivesBackItsSlot(t *testing.T) { + c := newEvacuateCommandForTest(0) + c.topologyInfo = tightTargetTopology() + target := c.topologyInfo.DataCenterInfos[0].RackInfos[1].DataNodeInfos[0] + // One free slot: volume 99 takes 8 of the 10 and shard 1.0 the ninth. + disk := target.DiskInfos[""] + si := erasure_coding.NewShardsInfo() + si.Set(erasure_coding.NewShardInfo(0, 0)) + disk.EcShardInfos = append(disk.EcShardInfos, &master_pb.VolumeEcShardInformationMessage{Id: 1, Collection: "c", EcIndexBits: si.Bitmap(), ShardSizes: si.SizesInt64()}) + var moved []uint32 + c.moveEcShards = func(moves []ecbalancer.Move, stopOnError bool) (result ec.EcShardMovesResult, err error) { + for _, m := range moves { + if m.VolumeID == 1 { + result.Failed++ + result.LandedMoves = append(result.LandedMoves, m) + err = errors.New("input/output error") + continue + } + result.Moved++ + moved = append(moved, m.VolumeID) + } + return result, err + } + var out bytes.Buffer + if err := c.evacuateEcVolumes(nil, "source:8080", true, true, &out); err != nil { + t.Fatalf("evacuate: %v", err) + } + if len(moved) != 1 || moved[0] != 2 { + t.Fatalf("expected volume 2 to take the slot volume 1 gave back, moved %v:\n%s", moved, out.String()) + } + if erasure_coding.ShardsInfoFromVolumeEcShardInformationMessage(disk.EcShardInfos[1]).Count() != 1 { + t.Fatal("the undo removed the shard the target already had") + } +}