From 602746f51d65fe168e3fbb68f7453cf35a17d1f0 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 14 Aug 2026 17:26:54 -0700 Subject: [PATCH] test: EC lifecycle chaos harness, with four fixes it found (#10763) * ec: let the encode's balance see a migrating volume's shards across disk-type buckets Shard generation writes beside the source .dat, so a cross-tier encode (source on hdd, -diskType=ssd) leaves the fresh shards in the source disk-type bucket. The encode's internal balance ingested only the target bucket, saw no shards, and planned no moves; the spread guard then correctly aborted the encode (and before that guard existed, the shards silently stayed clumped on the generation host in the wrong tier). EcBalance now takes the encode batch as migratingVolumeIds and ingests those volumes' shards from every bucket, while everything else keeps the bucket filter so a plain ec.balance never drags deliberately tiered shards onto another disk type. The in-memory model delete also becomes bucket-agnostic: a node holds a given shard in exactly one bucket, and a bucket-scoped delete missed cross-bucket moves in the dry-run model. * volume: decode reads shard 0 from its resolved path, not the EC volume's base dir On a multi-disk server a volume's shards can sit on several disks; the store registers each shard with its own path and CollectEcShards resolves them, but FindDatFileSize derived the .ec00 path from the EcVolume's base directory. When shard 0 lived on a sibling disk, VolumeEcShardsToVolume failed with 'open ...ec00: no such file or directory' and ec.decode aborted. * ec: decode re-copies shards the topology claims but the target does not hold An interrupted earlier decode or balance can leave the master believing the decode target holds a shard whose file never landed: the mount registered but the partial copy was cleaned, or the file was swept. The collect step took the topology's word for it, excluded the shard from the copy set, and the decode failed with 'missing shard'. Probe the target's live inventory (VolumeEcShardsInfo) and treat anything it cannot serve as still-to-copy. * ec: decode discovers shards across disk-type buckets Shards sit wherever encode generation and balance left them: a cross-tier encode leaves them in the source disk-type bucket, a partial migration straddles buckets. ec.decode scoped its shard discovery to the -diskType bucket and reported a decodable volume as having no shards at all. Union across buckets, the way the encode's shard verification already does. * test: EC chaos lifecycle harness Randomized, seeded sequences of the EC lifecycle against a live cluster in the production-shaped layout: multiple data disks per server, a separate -dir.idx directory so .ecx/.ecj sidecars are shared across disks, and a tagged ssd tier. Operations cover encode (hdd and ssd targets), balance, shard damage plus rebuild, decode, re-encode, deletes, scrub, tier moves, crash-restarts, sidecar fault injections (a data-dir .vif pushed into the shared idx dir; a stale-generation shard planted beside a newer encode), and interruptions: a real weed shell subprocess killed mid-encode, mid-decode, and mid-balance, with the recovery re-run required to converge. One invariant holds after every step: every stored byte reads back identical and every deleted needle stays deleted. EC_CHAOS_SEED and EC_CHAOS_STEPS make runs reproducible and scalable. A known gap is tolerated and logged rather than fixed here: a shard mounted on two disks of one node (orphan adoption after an interrupted copy) is invisible to ec.balance's dedup and unaddressable by ec.shard.unmount's shard@address form, so no cleanup path exists yet. * test: fail payload-corruption checks on the test goroutine t.Fatalf inside require.Eventually's condition runs on the poller's goroutine, where Goexit kills only that goroutine and the corruption message can be lost behind a generic timeout. Record the mismatch, end the polling, and fail on the test goroutine. Also assert the full shard count in the cross-bucket decode-discovery test. --- test/erasure_coding/chaos_lifecycle_test.go | 1175 +++++++++++++++++ weed/ec/ec_balance_migrate_test.go | 63 + weed/ec/ec_common.go | 77 +- weed/ec/ec_decode.go | 38 +- weed/ec/ec_encode.go | 2 +- weed/ec/ec_encode_test.go | 13 +- weed/server/volume_grpc_erasure_coding.go | 8 +- weed/shell/command_ec_common.go | 2 +- weed/storage/erasure_coding/ec_decoder.go | 18 +- .../ec_decoder_multidisk_test.go | 54 + weed/storage/erasure_coding/verification.go | 26 +- 11 files changed, 1423 insertions(+), 53 deletions(-) create mode 100644 test/erasure_coding/chaos_lifecycle_test.go create mode 100644 weed/ec/ec_balance_migrate_test.go create mode 100644 weed/storage/erasure_coding/ec_decoder_multidisk_test.go diff --git a/test/erasure_coding/chaos_lifecycle_test.go b/test/erasure_coding/chaos_lifecycle_test.go new file mode 100644 index 000000000..b64fcbc1d --- /dev/null +++ b/test/erasure_coding/chaos_lifecycle_test.go @@ -0,0 +1,1175 @@ +package erasure_coding + +import ( + "bytes" + "context" + "crypto/rand" + "encoding/json" + "fmt" + "io" + mrand "math/rand" + "net" + "net/http" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + + "github.com/seaweedfs/seaweedfs/weed/operation" + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/shell" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" +) + +// TestECChaosLifecycle drives randomized sequences of the EC lifecycle — +// encode (hdd and ssd targets), balance, shard damage + rebuild, decode, +// re-encode (a new generation), deletes, scrub, tier moves, crash-restarts, +// and sidecar fault injections — against a live cluster running the +// production-shaped layout: multiple data disks per server with a separate +// -dir.idx directory, so the .ecx/.ecj sidecars are shared across disks. +// +// One invariant is checked after every step: every byte a client stored comes +// back identical, and every deleted needle stays deleted. Shard counting alone +// cannot tell a healthy volume from one serving a stale generation or a +// mis-rebuilt shard; reading the payloads back can. +// +// The sequence is seeded (EC_CHAOS_SEED) and reproducible; EC_CHAOS_STEPS +// scales the random portion. The fault scenarios that motivated the test — +// losing a data-dir .vif (forcing the shared idx-dir fallback), and planting a +// stale-generation shard file next to a newer encode — are always exercised +// once, regardless of what the random schedule picks. +const ( + chaosMasterAddr = "127.0.0.1:9338" + chaosMasterPort = "9338" + chaosCollection = "chaos" + chaosServerCount = 3 + chaosDisksPerNode = 3 // disk0, disk1 default type; disk2 tagged ssd +) + +func chaosVolumePort(i int) string { return fmt.Sprintf("811%d", i) } + +func TestECChaosLifecycle(t *testing.T) { + if testing.Short() { + t.Skip("Skipping EC chaos lifecycle test in short mode") + } + + seed := int64(1) + if s := os.Getenv("EC_CHAOS_SEED"); s != "" { + v, err := strconv.ParseInt(s, 10, 64) + require.NoError(t, err, "EC_CHAOS_SEED must be an integer") + seed = v + } + steps := 8 + if s := os.Getenv("EC_CHAOS_STEPS"); s != "" { + v, err := strconv.Atoi(s) + require.NoError(t, err, "EC_CHAOS_STEPS must be an integer") + steps = v + } + t.Logf("chaos seed=%d steps=%d (override with EC_CHAOS_SEED / EC_CHAOS_STEPS)", seed, steps) + + testDir := t.TempDir() + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Minute) + defer cancel() + + cluster, err := startChaosCluster(ctx, testDir) + require.NoError(t, err) + defer cluster.Stop() + + require.NoError(t, waitForServer(chaosMasterAddr, 30*time.Second)) + for i := 0; i < chaosServerCount; i++ { + require.NoError(t, waitForServer("127.0.0.1:"+chaosVolumePort(i), 30*time.Second)) + } + time.Sleep(8 * time.Second) + + commandEnv := shell.NewCommandEnv(&shell.ShellOptions{ + Masters: stringPtr(chaosMasterAddr), + GrpcDialOption: grpc.WithInsecure(), + FilerGroup: stringPtr("default"), + }) + connectToMasterAndSync(ctx, t, commandEnv) + + r := &chaosRun{ + t: t, + ctx: ctx, + cluster: cluster, + env: commandEnv, + rng: mrand.New(mrand.NewSource(seed)), + testDir: testDir, + payloads: map[string][]byte{}, + deleted: map[string]bool{}, + fidVol: map[string]uint32{}, + volumes: map[uint32]*chaosVolumeState{}, + } + r.relock() + defer r.unlockIfHeld() + + // Seed the cluster with payloads across several volumes. The first upload + // also creates the collection, which the volume.grow spread below needs. + for i := 0; i < 24; i++ { + r.uploadOne() + } + require.GreaterOrEqual(t, len(r.volumes), 2, "seeding should produce at least two volumes") + time.Sleep(3 * time.Second) + + // Spread volumes onto every disk of every node: the master only enumerates + // disks that already hold data, and shards only spread across enumerated + // disks (see TestMultiDiskECBalanceNoShardLoss). Without this a balance can + // find no eligible targets and the encode's clump guard aborts. + require.Eventually(t, func() bool { + spread := nodeVolumeDiskCounts(t, commandEnv) + if len(spread) == chaosServerCount && allAtLeast(spread, 2) { + return true + } + for i := 0; i < chaosServerCount; i++ { + server := "127.0.0.1:" + chaosVolumePort(i) + if spread[server] < 2 { + out, gerr := captureCommandOutput(t, shell.Commands[findCommandIndex("volume.grow")], + []string{"-collection", chaosCollection, "-dataNode", server, "-count", "4"}, commandEnv) + t.Logf("volume.grow on %s: err=%v output:\n%s", server, gerr, out) + } + } + return false + }, 90*time.Second, 2*time.Second, "volumes never spread across >=2 disks on all %d nodes", chaosServerCount) + + r.verify("seeding") + + // Random schedule. Every op re-verifies the full payload set. + ops := []struct { + name string + weight int + run func() bool + }{ + {"encode", 4, r.opEncode}, + {"decode", 2, r.opDecode}, + {"balance", 2, r.opBalance}, + {"damage+rebuild", 2, r.opDamageAndRebuild}, + {"delete", 2, r.opDelete}, + {"upload", 2, r.opUpload}, + {"scrub", 2, r.opScrub}, + {"crash-restart", 1, r.opCrashRestart}, + {"tier-move", 1, r.opTierMove}, + {"vif-fallback", 1, r.opVifFallback}, + {"stale-generation", 1, r.opPlantStaleGeneration}, + {"interrupted-encode", 2, r.opInterruptedEncode}, + {"interrupted-decode", 1, r.opInterruptedDecode}, + {"interrupted-balance", 1, r.opInterruptedBalance}, + } + total := 0 + for _, op := range ops { + total += op.weight + } + ran := map[string]bool{} + for step := 1; step <= steps; step++ { + n := r.rng.Intn(total) + for _, op := range ops { + if n -= op.weight; n < 0 { + t.Logf("── chaos step %d/%d: %s ──", step, steps, op.name) + if op.run() { + ran[op.name] = true + r.verify(op.name) + } else { + t.Logf("step %d: %s not applicable, skipped", step, op.name) + } + break + } + } + } + + // Deterministic tail: the scenarios this test exists for always run once. + for _, must := range []struct { + name string + run func() bool + }{ + {"encode", r.opEncode}, + {"damage+rebuild", r.opDamageAndRebuild}, + {"vif-fallback", r.opVifFallback}, + {"stale-generation", r.opPlantStaleGeneration}, + {"interrupted-encode", r.opInterruptedEncode}, + {"interrupted-decode", r.opInterruptedDecode}, + {"interrupted-balance", r.opInterruptedBalance}, + {"decode", r.opDecode}, + } { + if ran[must.name] { + continue + } + t.Logf("── chaos tail: %s ──", must.name) + if must.run() { + r.verify(must.name) + } else { + t.Logf("tail: %s not applicable, skipped", must.name) + } + } + + r.verify("final") + t.Logf("chaos done: %d payloads live, %d deleted, %d volumes tracked", len(r.payloads), len(r.deleted), len(r.volumes)) +} + +type chaosVolumeState struct { + encoded bool +} + +type chaosRun struct { + t *testing.T + ctx context.Context + cluster *chaosCluster + env *shell.CommandEnv + rng *mrand.Rand + testDir string + + payloads map[string][]byte // live fid -> expected bytes + deleted map[string]bool // fids that must stay deleted + fidVol map[string]uint32 + volumes map[uint32]*chaosVolumeState + + unlock func() +} + +// ── invariants ────────────────────────────────────────────────────────────── + +// verify is the single invariant of the whole test: after any operation, every +// live payload reads back byte-identical from the current cluster state, and +// every deleted needle stays unreadable. Retries absorb heartbeat and mount +// propagation delays; content mismatches fail immediately — waiting cannot fix +// wrong bytes, and the first wrong read is the most useful state to stop in. +func (r *chaosRun) verify(afterStep string) { + r.t.Helper() + for fid, want := range r.payloads { + fid, want := fid, want + // The condition runs on Eventually's own goroutine, where t.Fatalf + // would only kill that goroutine; record a corruption and fail on the + // test goroutine instead. A wrong read still ends the polling at once — + // waiting cannot fix wrong bytes, and the first wrong read is the most + // useful state to stop in. + var corrupted string + require.Eventuallyf(r.t, func() bool { + got, err := chaosReadFid(fid, r.fidVol[fid]) + if err != nil { + return false + } + if !bytes.Equal(got, want) { + corrupted = fmt.Sprintf("payload %s corrupted after %s: got %d bytes, want %d bytes", fid, afterStep, len(got), len(want)) + } + return true + }, 90*time.Second, time.Second, "payload %s unreadable after %s", fid, afterStep) + require.Empty(r.t, corrupted, "%s", corrupted) + } + for fid := range r.deleted { + fid := fid + require.Eventuallyf(r.t, func() bool { + got, err := chaosReadFid(fid, r.fidVol[fid]) + return err != nil || len(got) == 0 + }, 30*time.Second, time.Second, "deleted payload %s came back after %s", fid, afterStep) + } + r.t.Logf("verified %d live + %d deleted payloads after %s", len(r.payloads), len(r.deleted), afterStep) +} + +// ── operations ────────────────────────────────────────────────────────────── + +func (r *chaosRun) opEncode() bool { + vid, ok := r.pickVolume(false) + if !ok { + return false + } + args := []string{"-volumeId", fmt.Sprintf("%d", vid), "-collection", chaosCollection, "-force"} + if r.rng.Intn(2) == 0 { + args = append(args, "-diskType", "ssd") + } + r.t.Logf("ec.encode args: %v", args) + out, err := r.shellCommand("ec.encode", args...) + r.t.Logf("ec.encode v%d output:\n%s", vid, out) + if err != nil { + vl, _ := r.shellCommand("volume.list") + r.t.Logf("volume.list at encode failure:\n%s", vl) + } + require.NoError(r.t, err, "ec.encode volume %d", vid) + r.volumes[vid].encoded = true + r.requireSingleGeneration(vid, "ec.encode") + return true +} + +func (r *chaosRun) opDecode() bool { + vid, ok := r.pickVolume(true) + if !ok { + return false + } + // -checkMinFreeSpace=false: this intentionally tiny cluster would otherwise + // refuse the decode for lack of headroom, which is not what is under test. + out, err := r.shellCommand("ec.decode", + "-volumeId", fmt.Sprintf("%d", vid), "-collection", chaosCollection, "-checkMinFreeSpace=false") + r.t.Logf("ec.decode v%d output:\n%s", vid, out) + require.NoError(r.t, err, "ec.decode volume %d", vid) + r.volumes[vid].encoded = false + return true +} + +func (r *chaosRun) opBalance() bool { + out, err := r.shellCommand("ec.balance", "-collection", chaosCollection, "-apply") + r.t.Logf("ec.balance output:\n%s", out) + require.NoError(r.t, err, "ec.balance") + return true +} + +// opDamageAndRebuild removes up to two shard files of an encoded volume +// straight off the disks, restarts the servers so the master relearns disk +// truth, and repairs with ec.rebuild — the flow of a real shard-loss incident. +func (r *chaosRun) opDamageAndRebuild() bool { + vid, ok := r.pickVolume(true) + if !ok { + return false + } + before := len(collectDistinctShardIDs(r.testDir, vid)) + if before < erasureShardCount { + r.t.Logf("damage+rebuild: volume %d has %d/%d distinct shards on disk, skipping", vid, before, erasureShardCount) + return false // a prior fault is still outstanding; skip rather than stack damage + } + removed := removeTwoShardFiles(r.t, r.testDir, vid) + r.t.Logf("removed shard files for shards %v of volume %d", removed, vid) + + // After the restart the master must agree with disk truth before any repair + // runs. The exact count is not fixed: a stale shard file planted by an + // earlier fault can legitimately re-register from disk and cover one of the + // removed ids, so the invariant is agreement, not a specific number. + r.restartAllVolumeServers() + require.Eventually(r.t, func() bool { + registered := masterEcShardIds(r.env, vid) + onDisk := collectDistinctShardIDs(r.testDir, vid) + if len(registered) != len(onDisk) { + return false + } + for id := range onDisk { + if !registered[id] { + return false + } + } + return true + }, 90*time.Second, 2*time.Second, "master never agreed with disk truth for volume %d (master=%v disk=%v)", + vid, sortedKeysOf(masterEcShardIds(r.env, vid)), sortedKeysOf(collectDistinctShardIDs(r.testDir, vid))) + + out, err := r.shellCommand("ec.rebuild", "-collection", chaosCollection, "-apply") + r.t.Logf("ec.rebuild output:\n%s", out) + require.NoError(r.t, err, "ec.rebuild") + require.Eventually(r.t, func() bool { + return len(collectDistinctShardIDs(r.testDir, vid)) == erasureShardCount + }, 90*time.Second, time.Second, "ec.rebuild did not restore all shards of volume %d", vid) + return true +} + +func (r *chaosRun) opDelete() bool { + // Keep at least one live payload per volume so no volume ever empties out + // completely (an all-deleted volume decodes into nothing, which is its own + // test, not this one). + liveByVol := map[uint32]int{} + for fid := range r.payloads { + liveByVol[r.fidVol[fid]]++ + } + var candidates []string + for fid := range r.payloads { + if liveByVol[r.fidVol[fid]] > 1 { + candidates = append(candidates, fid) + liveByVol[r.fidVol[fid]]-- + } + if len(candidates) == 2 { + break + } + } + if len(candidates) == 0 { + return false + } + for _, fid := range candidates { + require.NoError(r.t, chaosDeleteFid(fid, r.fidVol[fid]), "delete %s", fid) + delete(r.payloads, fid) + r.deleted[fid] = true + r.t.Logf("deleted %s (volume %d)", fid, r.fidVol[fid]) + } + return true +} + +func (r *chaosRun) opUpload() bool { + for i := 0; i < 3; i++ { + r.uploadOne() + } + return true +} + +func (r *chaosRun) opScrub() bool { + out, err := r.shellCommand("ec.scrub", "-mode", "local") + r.t.Logf("ec.scrub output:\n%s", out) + require.NoError(r.t, err, "ec.scrub") + require.NotContains(r.t, out, "scrub failures", "ec.scrub reported broken EC volumes") + return true +} + +func (r *chaosRun) opCrashRestart() bool { + r.restartAllVolumeServers() + return true +} + +func (r *chaosRun) opTierMove() bool { + // Best effort: with -fullPercent=0 every quiet regular volume qualifies. + // Zero moved volumes is fine — the invariant read-back is the point. + out, err := r.shellCommand("volume.tier.move", + "-fromDiskType", "hdd", "-toDiskType", "ssd", + "-collectionPattern", "^"+chaosCollection+"$", + "-fullPercent", "0", "-quietFor", "1s", "-apply") + r.t.Logf("volume.tier.move output:\n%s", out) + require.NoError(r.t, err, "volume.tier.move") + return true +} + +// opVifFallback simulates the split-sidecar layout: the data-dir .vif of one +// disk's shards moves into the server's shared -dir.idx directory, and the +// server restarts. Loading must fall back to the idx-dir copy (issue #9212 +// layout) and reads must stay correct. +func (r *chaosRun) opVifFallback() bool { + vid, ok := r.pickVolume(true) + if !ok { + return false + } + moved := false + for server := 0; server < chaosServerCount && !moved; server++ { + for disk := 0; disk < chaosDisksPerNode; disk++ { + dataVif := filepath.Join(r.testDir, fmt.Sprintf("server%d_disk%d", server, disk), + fmt.Sprintf("%s_%d.vif", chaosCollection, vid)) + if _, err := os.Stat(dataVif); err != nil { + continue + } + idxVif := filepath.Join(r.testDir, fmt.Sprintf("server%d_idx", server), + fmt.Sprintf("%s_%d.vif", chaosCollection, vid)) + require.NoError(r.t, os.Rename(dataVif, idxVif), "move .vif to idx dir") + r.t.Logf("moved %s -> %s", dataVif, idxVif) + r.restartOneVolumeServer(server) + moved = true + break + } + } + return moved +} + +// opPlantStaleGeneration reproduces the orphaned-generation hazard: it stashes +// an encoded volume's shard files, decodes and re-encodes the volume (a new +// generation with a new .vif stamp), then plants one stale shard file from the +// old generation onto a disk of a server that holds new-generation shards, and +// restarts that server. Whatever the server decides to do with the orphan — +// delete it, quarantine it, or register it — reads must never serve its bytes. +func (r *chaosRun) opPlantStaleGeneration() bool { + vid, ok := r.pickVolume(true) + if !ok { + return false + } + if n := len(collectDistinctShardIDs(r.testDir, vid)); n < erasureShardCount { + r.t.Logf("stale-generation: volume %d has %d/%d distinct shards on disk, skipping", vid, n, erasureShardCount) + return false + } + + // Stash one old-generation shard file. + stale := r.stashOneShardFile(vid) + if stale == "" { + return false + } + + out, err := r.shellCommand("ec.decode", + "-volumeId", fmt.Sprintf("%d", vid), "-collection", chaosCollection, "-checkMinFreeSpace=false") + r.t.Logf("ec.decode v%d output:\n%s", vid, out) + require.NoError(r.t, err, "ec.decode volume %d (stale-generation scenario)", vid) + r.verify("decode before re-encode") + + out, err = r.shellCommand("ec.encode", + "-volumeId", fmt.Sprintf("%d", vid), "-collection", chaosCollection, "-force") + r.t.Logf("ec.encode v%d output:\n%s", vid, out) + require.NoError(r.t, err, "re-encode volume %d (stale-generation scenario)", vid) + r.volumes[vid].encoded = true + + // Plant the old-generation shard on a server that holds new shards, on a + // disk of that server that does not currently hold this volume — the + // cross-disk mixing case, where the shared idx-dir sidecars are the only + // local generation authority for the planted file. + planted := false + for server := 0; server < chaosServerCount && !planted; server++ { + serverHasNew, diskWithout := false, -1 + for disk := 0; disk < chaosDisksPerNode; disk++ { + pattern := filepath.Join(r.testDir, fmt.Sprintf("server%d_disk%d", server, disk), + fmt.Sprintf("%s_%d.ec*", chaosCollection, vid)) + if m, _ := filepath.Glob(pattern); len(m) > 0 { + serverHasNew = true + } else if diskWithout == -1 { + diskWithout = disk + } + } + if serverHasNew && diskWithout >= 0 { + dst := filepath.Join(r.testDir, fmt.Sprintf("server%d_disk%d", server, diskWithout), filepath.Base(stale)) + require.NoError(r.t, copyFileContents(stale, dst), "plant stale shard") + r.t.Logf("planted stale generation shard %s on server %d disk %d", filepath.Base(stale), server, diskWithout) + r.restartOneVolumeServer(server) + planted = true + } + } + if !planted { + r.t.Logf("no server had both new shards and a free disk; stale plant skipped") + } + return true +} + +// ── interruption ops ──────────────────────────────────────────────────────── +// +// These kill a real `weed shell` subprocess mid-operation — the operator's +// shell dying — and then prove the cluster recovers: whatever half-finished +// state the kill left (readonly sources, partial or unmounted shards, an +// undeleted original, a half-collected decode), the next run of the same +// command must converge to a clean state, and reads must stay correct +// throughout. This is the restart-not-resume recovery model: an interrupted +// run is never resumed, the retry starts clean via the pre-encode sweep. + +// runInterruptedShell feeds "lock" plus the command to a weed shell +// subprocess and kills the process after killAfter. Stdin stays open so the +// shell never exits gracefully — the lock releases only through the master +// noticing the dead connection, which the follow-up relock must survive. +func (r *chaosRun) runInterruptedShell(command string, killAfter time.Duration) string { + weedBinary := findWeedBinary() + require.NotEmpty(r.t, weedBinary, "weed binary not found") + cmd := exec.CommandContext(r.ctx, weedBinary, "shell", "-master="+chaosMasterAddr) + var out bytes.Buffer + cmd.Stdout, cmd.Stderr = &out, &out + stdin, err := cmd.StdinPipe() + require.NoError(r.t, err) + require.NoError(r.t, cmd.Start()) + fmt.Fprintf(stdin, "lock\n%s\n", command) + time.Sleep(killAfter) + cmd.Process.Kill() + cmd.Wait() + return out.String() +} + +func (r *chaosRun) volumeHasRegularReplica(vid uint32) bool { + found := false + for _, dn := range chaosDataNodes(r.env) { + for _, di := range dn.GetDiskInfos() { + for _, vi := range di.GetVolumeInfos() { + if vi.GetId() == vid { + found = true + } + } + } + } + return found +} + +func (r *chaosRun) opInterruptedEncode() bool { + vid, ok := r.pickVolume(false) + if !ok { + return false + } + killAfter := time.Duration(1+r.rng.Intn(8)) * time.Second + r.unlockIfHeld() + out := r.runInterruptedShell( + fmt.Sprintf("ec.encode -volumeId %d -collection %s -force", vid, chaosCollection), killAfter) + r.t.Logf("killed ec.encode v%d after %v; output so far:\n%s", vid, killAfter, out) + r.relock() + + // Recovery. If the kill came after the originals were deleted, the encode + // had effectively completed and the volume is EC now; any earlier kill + // leaves the regular volume in place (possibly readonly, possibly beside + // partial shards), and a re-run must sweep the leftovers and finish. + if !r.volumeHasRegularReplica(vid) { + r.t.Logf("interrupted encode of volume %d had already completed", vid) + r.volumes[vid].encoded = true + r.requireSingleGeneration(vid, "interrupted ec.encode (completed)") + return true + } + out2, err := r.shellCommand("ec.encode", + "-volumeId", fmt.Sprintf("%d", vid), "-collection", chaosCollection, "-force") + r.t.Logf("recovery ec.encode v%d output:\n%s", vid, out2) + if err != nil && !r.volumeHasRegularReplica(vid) { + // The killed run's original-deletion outran the topology snapshot the + // re-run planned from; the encode had in fact completed. + r.t.Logf("interrupted encode of volume %d had already completed (original deletion outran the topology)", vid) + err = nil + } + if err != nil { + vl, _ := r.shellCommand("volume.list") + r.t.Logf("volume.list at recovery-encode failure:\n%s", vl) + } + require.NoError(r.t, err, "recovery ec.encode volume %d after interruption", vid) + r.volumes[vid].encoded = true + r.requireSingleGeneration(vid, "recovery ec.encode") + return true +} + +func (r *chaosRun) opInterruptedDecode() bool { + vid, ok := r.pickVolume(true) + if !ok { + return false + } + killAfter := time.Duration(1+r.rng.Intn(6)) * time.Second + r.unlockIfHeld() + out := r.runInterruptedShell( + fmt.Sprintf("ec.decode -volumeId %d -collection %s -checkMinFreeSpace=false", vid, chaosCollection), killAfter) + r.t.Logf("killed ec.decode v%d after %v; output so far:\n%s", vid, killAfter, out) + r.relock() + + // Recovery: while any shards remain the decode is unfinished (the kill may + // have left a hybrid: regenerated volume plus undeleted shards); a re-run + // must complete it. No shards left means the decode had finished — and the + // master's view can lag the killed run's final deletions, so a re-run that + // finds no shards is also completion, not a failure. + if len(masterEcShardIds(r.env, vid)) > 0 { + out2, err := r.shellCommand("ec.decode", + "-volumeId", fmt.Sprintf("%d", vid), "-collection", chaosCollection, "-checkMinFreeSpace=false") + r.t.Logf("recovery ec.decode v%d output:\n%s", vid, out2) + if err != nil && strings.Contains(err.Error(), "no EC shards found") { + r.t.Logf("interrupted decode of volume %d had already completed (shard deletions outran the topology)", vid) + } else { + require.NoError(r.t, err, "recovery ec.decode volume %d after interruption", vid) + } + } else { + r.t.Logf("interrupted decode of volume %d had already completed", vid) + } + r.volumes[vid].encoded = false + return true +} + +func (r *chaosRun) opInterruptedBalance() bool { + anyEncoded := false + for _, st := range r.volumes { + if st.encoded { + anyEncoded = true + } + } + if !anyEncoded { + return false + } + killAfter := time.Duration(1+r.rng.Intn(5)) * time.Second + r.unlockIfHeld() + out := r.runInterruptedShell( + fmt.Sprintf("ec.balance -collection %s -apply", chaosCollection), killAfter) + r.t.Logf("killed ec.balance after %v; output so far:\n%s", killAfter, out) + r.relock() + + // Recovery: an interrupted move leaves a shard copied but not yet deleted + // at the source. Re-running the balance must converge — its dedup phase + // removes the extra copies — until the replication check is clean. The + // master's cleanup of the killed shell's lock can invalidate the lock this + // harness re-acquired right after the kill, so a lock error inside the + // loop is answered the way an operator would: run lock again and retry. + require.Eventually(r.t, func() bool { + relockOn := func(err error) bool { + if err != nil && strings.Contains(err.Error(), "lock") { + r.relock() + } + return err != nil + } + if _, err := r.shellCommand("ec.balance", "-collection", chaosCollection, "-apply"); relockOn(err) { + r.t.Logf("recovery ec.balance: %v", err) + return false + } + report, err := r.shellCommand("ec.check.replication", "-details") + if relockOn(err) { + r.t.Logf("ec.check.replication: %v", err) + return false + } + if strings.Contains(report, "under-replicated") { + r.t.Logf("replication not clean yet:\n%s", report) + return false + } + if crossNode, sameNode := classifyOverReplication(report); crossNode { + r.t.Logf("replication not clean yet:\n%s", report) + return false + } else if sameNode { + // KNOWN GAP: a shard mounted on two disks of ONE node (e.g. an + // orphan adopted after an interrupted copy) is invisible to + // ec.balance's dedup, and ec.shard.unmount's shard@address form + // cannot disambiguate two copies behind one address. Nothing can + // clean this state today; reads stay correct, so tolerate it here + // and keep it visible in the log. + r.t.Logf("tolerating same-node duplicate shards (no cleanup path exists):\n%s", report) + } + return true + }, 120*time.Second, 3*time.Second, "cluster never converged to clean replication after interrupted balance") + return true +} + +// classifyOverReplication parses an ec.check.replication -details report and +// says whether any shard is duplicated across distinct nodes (crossNode) or +// only within one node (sameNode, the two-disks-one-node adoption case). +func classifyOverReplication(report string) (crossNode, sameNode bool) { + for _, line := range strings.Split(report, "\n") { + open := strings.Index(line, "=> [") + if open < 0 { + continue + } + addrs := strings.Fields(strings.Trim(line[open+len("=> ["):], "[] \r")) + if len(addrs) < 2 { + continue + } + distinct := map[string]bool{} + for _, a := range addrs { + distinct[a] = true + } + if len(distinct) > 1 { + crossNode = true + } else { + sameNode = true + } + } + return crossNode, sameNode +} + +// ── helpers ───────────────────────────────────────────────────────────────── + +func (r *chaosRun) pickVolume(encoded bool) (uint32, bool) { + var candidates []uint32 + for vid, st := range r.volumes { + if st.encoded == encoded { + candidates = append(candidates, vid) + } + } + if len(candidates) == 0 { + return 0, false + } + // Deterministic pick under one seed: order by volume id. + min := candidates[0] + for _, v := range candidates { + if v < min { + min = v + } + } + return min, true +} + +func (r *chaosRun) uploadOne() { + data := make([]byte, 2048+r.rng.Intn(14*1024)) + _, err := rand.Read(data) + require.NoError(r.t, err) + var vid needle.VolumeId + var fid string + for retry := 0; retry < 5; retry++ { + vid, fid, err = chaosUploadPayload(data) + if err == nil { + break + } + time.Sleep(2 * time.Second) + } + require.NoError(r.t, err, "upload payload") + r.payloads[fid] = data + r.fidVol[fid] = uint32(vid) + if _, ok := r.volumes[uint32(vid)]; !ok { + r.volumes[uint32(vid)] = &chaosVolumeState{} + } +} + +func (r *chaosRun) shellCommand(name string, args ...string) (string, error) { + return captureCommandOutput(r.t, shell.Commands[findCommandIndex(name)], args, r.env) +} + +func (r *chaosRun) relock() { + locked, unlock := tryLockWithTimeout(r.t, r.env, 45*time.Second) + require.True(r.t, locked, "could not acquire shell lock") + r.unlock = unlock +} + +func (r *chaosRun) unlockIfHeld() { + if r.unlock != nil { + r.unlock() + r.unlock = nil + } +} + +func (r *chaosRun) restartAllVolumeServers() { + require.NoError(r.t, r.cluster.RestartVolumeServers(r.ctx)) + for i := 0; i < chaosServerCount; i++ { + require.NoError(r.t, waitForServer("127.0.0.1:"+chaosVolumePort(i), 30*time.Second)) + } + time.Sleep(3 * time.Second) + r.relock() // the restart's master disconnect drops the shell lock +} + +func (r *chaosRun) restartOneVolumeServer(i int) { + require.NoError(r.t, r.cluster.RestartVolumeServer(r.ctx, i)) + require.NoError(r.t, waitForServer("127.0.0.1:"+chaosVolumePort(i), 30*time.Second)) + time.Sleep(3 * time.Second) + r.relock() +} + +// requireSingleGeneration asserts that, at a quiescent point, every EC shard +// entry the master reports for the volume carries the same encode generation +// stamp — the state the encode pipeline is supposed to leave behind. +func (r *chaosRun) requireSingleGeneration(vid uint32, step string) { + r.t.Helper() + require.Eventually(r.t, func() bool { + generations := masterEcGenerations(r.env, vid) + return len(generations) == 1 + }, 60*time.Second, 2*time.Second, + "volume %d reports mixed encode generations after %s: %v", vid, step, masterEcGenerations(r.env, vid)) +} + +// stashOneShardFile copies one current shard file of the volume into a stash +// dir and returns the stash path ("" when none found). +func (r *chaosRun) stashOneShardFile(vid uint32) string { + stashDir := filepath.Join(r.testDir, "stale_stash") + _ = os.MkdirAll(stashDir, 0o755) + for server := 0; server < chaosServerCount; server++ { + for disk := 0; disk < chaosDisksPerNode; disk++ { + pattern := filepath.Join(r.testDir, fmt.Sprintf("server%d_disk%d", server, disk), + fmt.Sprintf("%s_%d.ec*", chaosCollection, vid)) + matches, _ := filepath.Glob(pattern) + for _, m := range matches { + if strings.HasSuffix(m, ".ecx") || strings.HasSuffix(m, ".ecj") || strings.HasSuffix(m, ".ecsum") { + continue + } + dst := filepath.Join(stashDir, filepath.Base(m)) + if err := copyFileContents(m, dst); err != nil { + continue + } + return dst + } + } + } + return "" +} + +func copyFileContents(src, dst string) error { + in, err := os.Open(src) + if err != nil { + return err + } + defer in.Close() + out, err := os.Create(dst) + if err != nil { + return err + } + defer out.Close() + if _, err := io.Copy(out, in); err != nil { + return err + } + return out.Sync() +} + +// chaosDataNodes lists the data nodes from a fresh master topology snapshot. +func chaosDataNodes(commandEnv *shell.CommandEnv) []*master_pb.DataNodeInfo { + var resp *master_pb.VolumeListResponse + err := commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + var e error + resp, e = client.VolumeList(context.Background(), &master_pb.VolumeListRequest{}) + return e + }) + if err != nil || resp.GetTopologyInfo() == nil { + return nil + } + var nodes []*master_pb.DataNodeInfo + for _, dc := range resp.GetTopologyInfo().GetDataCenterInfos() { + for _, rack := range dc.GetRackInfos() { + nodes = append(nodes, rack.GetDataNodeInfos()...) + } + } + return nodes +} + +// masterEcGenerations returns the distinct encode generation stamps the master +// currently reports for a volume's EC shards. +func masterEcGenerations(commandEnv *shell.CommandEnv, volumeId uint32) map[int64]bool { + generations := map[int64]bool{} + var resp *master_pb.VolumeListResponse + err := commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { + var e error + resp, e = client.VolumeList(context.Background(), &master_pb.VolumeListRequest{}) + return e + }) + if err != nil || resp.GetTopologyInfo() == nil { + return generations + } + for _, dc := range resp.GetTopologyInfo().GetDataCenterInfos() { + for _, rack := range dc.GetRackInfos() { + for _, dn := range rack.GetDataNodeInfos() { + for _, di := range dn.GetDiskInfos() { + for _, eci := range di.GetEcShardInfos() { + if eci.GetId() == volumeId { + generations[eci.GetEncodeTsNs()] = true + } + } + } + } + } + } + return generations +} + +// ── payload plumbing against the chaos master ─────────────────────────────── + +func chaosUploadPayload(data []byte) (needle.VolumeId, string, error) { + assignResult, err := operation.Assign(context.Background(), func(ctx context.Context) pb.ServerAddress { + return pb.ServerAddress(chaosMasterAddr) + }, grpc.WithInsecure(), &operation.VolumeAssignRequest{ + Count: 1, + Collection: chaosCollection, + Replication: "000", + }) + if err != nil { + return 0, "", err + } + uploader, err := operation.NewUploader() + if err != nil { + return 0, "", err + } + uploadResult, err, _ := uploader.Upload(context.Background(), bytes.NewReader(data), &operation.UploadOption{ + UploadUrl: "http://" + assignResult.Url + "/" + assignResult.Fid, + Filename: "chaos.bin", + MimeType: "application/octet-stream", + }) + if err != nil { + return 0, "", err + } + if uploadResult.Error != "" { + return 0, "", fmt.Errorf("upload error: %s", uploadResult.Error) + } + fidObj, err := needle.ParseFileIdFromString(assignResult.Fid) + if err != nil { + return 0, "", err + } + return fidObj.VolumeId, assignResult.Fid, nil +} + +func chaosLookupLocations(volumeId uint32) ([]string, error) { + resp, err := http.Get(fmt.Sprintf("http://%s/dir/lookup?volumeId=%d", chaosMasterAddr, volumeId)) + if err != nil { + return nil, err + } + defer resp.Body.Close() + var lookup struct { + Locations []struct { + Url string `json:"url"` + } `json:"locations"` + } + if err := json.NewDecoder(resp.Body).Decode(&lookup); err != nil { + return nil, err + } + var urls []string + for _, l := range lookup.Locations { + urls = append(urls, l.Url) + } + if len(urls) == 0 { + return nil, fmt.Errorf("no locations for volume %d", volumeId) + } + return urls, nil +} + +func chaosReadFid(fid string, volumeId uint32) ([]byte, error) { + urls, err := chaosLookupLocations(volumeId) + if err != nil { + return nil, err + } + var lastErr error + for _, url := range urls { + get, err := http.Get(fmt.Sprintf("http://%s/%s", url, fid)) + if err != nil { + lastErr = err + continue + } + body, err := io.ReadAll(get.Body) + get.Body.Close() + if err != nil { + lastErr = err + continue + } + if get.StatusCode != http.StatusOK { + lastErr = fmt.Errorf("GET %s from %s: %d", fid, url, get.StatusCode) + continue + } + return body, nil + } + return nil, lastErr +} + +func chaosDeleteFid(fid string, volumeId uint32) error { + urls, err := chaosLookupLocations(volumeId) + if err != nil { + return err + } + var lastErr error + for _, url := range urls { + req, err := http.NewRequest(http.MethodDelete, fmt.Sprintf("http://%s/%s", url, fid), nil) + if err != nil { + return err + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + lastErr = err + continue + } + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + if resp.StatusCode >= 200 && resp.StatusCode < 300 { + return nil + } + lastErr = fmt.Errorf("DELETE %s from %s: %d", fid, url, resp.StatusCode) + } + return lastErr +} + +// ── cluster ───────────────────────────────────────────────────────────────── + +// chaosCluster is a master plus three volume servers, each with three data +// disks (the third tagged ssd) and a separate -dir.idx directory, so the +// .ecx/.ecj sidecars are shared across the server's disks — the layout whose +// edge cases this test exists to exercise. Individual servers can be killed +// and restarted over their existing directories. +type chaosCluster struct { + masterCmd *exec.Cmd + volumeServers []*exec.Cmd + testDir string + logFiles []*os.File +} + +func (c *chaosCluster) Stop() { + for _, cmd := range c.volumeServers { + if cmd != nil && cmd.Process != nil { + cmd.Process.Kill() + cmd.Wait() + } + } + if c.masterCmd != nil && c.masterCmd.Process != nil { + c.masterCmd.Process.Kill() + c.masterCmd.Wait() + } + for _, f := range c.logFiles { + if f != nil { + f.Close() + } + } +} + +func (c *chaosCluster) RestartVolumeServers(ctx context.Context) error { + for i := range c.volumeServers { + if err := c.RestartVolumeServer(ctx, i); err != nil { + return err + } + } + return nil +} + +func (c *chaosCluster) RestartVolumeServer(ctx context.Context, i int) error { + if cmd := c.volumeServers[i]; cmd != nil && cmd.Process != nil { + cmd.Process.Kill() + cmd.Wait() + } + time.Sleep(time.Second) + cmd, err := c.startVolumeServer(ctx, i, "volume-restart.log") + if err != nil { + return err + } + c.volumeServers[i] = cmd + time.Sleep(2 * time.Second) + return nil +} + +func (c *chaosCluster) startVolumeServer(ctx context.Context, i int, logName string) (*exec.Cmd, error) { + weedBinary := findWeedBinary() + if weedBinary == "" { + return nil, fmt.Errorf("weed binary not found") + } + var diskDirs, maxVolumes, diskTypes []string + for d := 0; d < chaosDisksPerNode; d++ { + dir := filepath.Join(c.testDir, fmt.Sprintf("server%d_disk%d", i, d)) + if err := os.MkdirAll(dir, 0o755); err != nil { + return nil, err + } + diskDirs = append(diskDirs, dir) + maxVolumes = append(maxVolumes, "4") + if d == chaosDisksPerNode-1 { + diskTypes = append(diskTypes, "ssd") + } else { + diskTypes = append(diskTypes, "hdd") + } + } + idxDir := filepath.Join(c.testDir, fmt.Sprintf("server%d_idx", i)) + if err := os.MkdirAll(idxDir, 0o755); err != nil { + return nil, err + } + cmd := exec.CommandContext(ctx, weedBinary, "volume", + "-port", chaosVolumePort(i), + "-dir", strings.Join(diskDirs, ","), + "-dir.idx", idxDir, + "-disk", strings.Join(diskTypes, ","), + "-max", strings.Join(maxVolumes, ","), + "-master", chaosMasterAddr, + "-ip", "127.0.0.1", + "-dataCenter", "dc1", + "-rack", fmt.Sprintf("rack%d", i), + ) + logDir := filepath.Join(c.testDir, fmt.Sprintf("server%d_logs", i)) + if err := os.MkdirAll(logDir, 0o755); err != nil { + return nil, err + } + logFile, err := os.OpenFile(filepath.Join(logDir, logName), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644) + if err != nil { + return nil, err + } + c.logFiles = append(c.logFiles, logFile) + cmd.Stdout = logFile + cmd.Stderr = logFile + if err := cmd.Start(); err != nil { + return nil, err + } + return cmd, nil +} + +func startChaosCluster(ctx context.Context, dataDir string) (*chaosCluster, error) { + weedBinary := findWeedBinary() + if weedBinary == "" { + return nil, fmt.Errorf("weed binary not found") + } + // A leaked cluster from an earlier run would silently absorb this run's + // traffic (same fixed ports) and make every on-disk assertion meaningless. + // Refuse to start over occupied ports. + ports := []string{chaosMasterAddr} + for i := 0; i < chaosServerCount; i++ { + ports = append(ports, "127.0.0.1:"+chaosVolumePort(i)) + } + for _, addr := range ports { + if conn, err := net.DialTimeout("tcp", addr, 300*time.Millisecond); err == nil { + conn.Close() + return nil, fmt.Errorf("port %s is already in use (stale cluster from an earlier run?)", addr) + } + } + cluster := &chaosCluster{testDir: dataDir} + + masterDir := filepath.Join(dataDir, "master") + if err := os.MkdirAll(masterDir, 0o755); err != nil { + return nil, err + } + masterCmd := exec.CommandContext(ctx, weedBinary, "master", + "-port", chaosMasterPort, + "-mdir", masterDir, + "-volumeSizeLimitMB", "10", + "-ip", "127.0.0.1", + "-peers", "none", + ) + masterLog, err := os.Create(filepath.Join(masterDir, "master.log")) + if err != nil { + return nil, err + } + cluster.logFiles = append(cluster.logFiles, masterLog) + masterCmd.Stdout = masterLog + masterCmd.Stderr = masterLog + if err := masterCmd.Start(); err != nil { + return nil, err + } + cluster.masterCmd = masterCmd + time.Sleep(2 * time.Second) + + for i := 0; i < chaosServerCount; i++ { + cmd, err := cluster.startVolumeServer(ctx, i, "volume.log") + if err != nil { + cluster.Stop() + return nil, fmt.Errorf("start volume server %d: %w", i, err) + } + cluster.volumeServers = append(cluster.volumeServers, cmd) + } + time.Sleep(8 * time.Second) + return cluster, nil +} diff --git a/weed/ec/ec_balance_migrate_test.go b/weed/ec/ec_balance_migrate_test.go new file mode 100644 index 000000000..aeeb3835b --- /dev/null +++ b/weed/ec/ec_balance_migrate_test.go @@ -0,0 +1,63 @@ +package ec + +import ( + "testing" + + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" + "github.com/seaweedfs/seaweedfs/weed/storage/types" +) + +// TestEcBalanceMigratesCrossDiskTypeShards: a cross-tier encode generates its +// shards beside the source .dat — in the SOURCE disk-type bucket — while the +// encode's balance targets -diskType. The balance must still see and spread +// those shards when the volume is named as migrating; without that, the +// planner finds nothing in the target bucket, plans no moves, and the encode's +// spread guard aborts a perfectly good encode. +func TestEcBalanceMigratesCrossDiskTypeShards(t *testing.T) { + allShards := []erasure_coding.ShardId{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13} + + build := func() []*EcNode { + // Volume 1's fresh shards sit in the default (hdd, "") bucket of dn1; + // the balance runs with -diskType=ssd. + return []*EcNode{ + newEcNode("dc1", "rack1", "dn1", 100).addEcVolumeAndShardsForTest(1, "c1", allShards), + newEcNode("dc1", "rack2", "dn2", 100), + newEcNode("dc1", "rack3", "dn3", 100), + } + } + + // Without the migrating hint the target-bucket filter hides the shards: + // nothing moves. This is the standalone ec.balance semantic — shards of + // other tiers stay where they are. + ecb := &ecBalancer{ + ecNodes: build(), + applyBalancing: false, + diskType: types.SsdType, + } + if err := ecb.balance([]string{"c1"}); err != nil { + t.Fatalf("balance without migrating hint: %v", err) + } + if got := ecb.ecNodes[0].LocalShardIdCount(1); got != len(allShards) { + t.Fatalf("without migrating hint, cross-type shards must stay put; dn1 has %d", got) + } + + // With the volume named as migrating, the balance ingests the shards from + // the source bucket and spreads them. + ecb = &ecBalancer{ + ecNodes: build(), + applyBalancing: false, + diskType: types.SsdType, + migratingVolumeIds: map[uint32]bool{1: true}, + } + if err := ecb.balance([]string{"c1"}); err != nil { + t.Fatalf("balance with migrating hint: %v", err) + } + onDn1 := ecb.ecNodes[0].LocalShardIdCount(1) + spread := ecb.ecNodes[1].LocalShardIdCount(1) + ecb.ecNodes[2].LocalShardIdCount(1) + if onDn1 == len(allShards) || spread == 0 { + t.Fatalf("migrating volume's shards did not spread: dn1=%d others=%d", onDn1, spread) + } + if onDn1+spread != len(allShards) { + t.Fatalf("shards lost or duplicated during dry-run spread: dn1=%d others=%d", onDn1, spread) + } +} diff --git a/weed/ec/ec_common.go b/weed/ec/ec_common.go index 1fda10a09..c8ab8a90c 100644 --- a/weed/ec/ec_common.go +++ b/weed/ec/ec_common.go @@ -298,7 +298,7 @@ func MoveMountedShardToEcNode(env *Env, existingLocation *EcNode, collection str } destinationEcNode.AddEcVolumeShards(vid, collection, copiedShardIds, diskType) - existingLocation.DeleteEcVolumeShards(vid, copiedShardIds, diskType) + existingLocation.DeleteEcVolumeShards(vid, copiedShardIds) return nil @@ -634,9 +634,18 @@ func (ecNode *EcNode) AddEcVolumeShards(vid needle.VolumeId, collection string, return ecNode } -func (ecNode *EcNode) DeleteEcVolumeShards(vid needle.VolumeId, shardIds []erasure_coding.ShardId, diskType types.DiskType) *EcNode { +// DeleteEcVolumeShards removes the shards from the node model wherever they +// sit. A node holds a given shard in exactly one disk-type bucket, but which +// bucket is not the caller's to know: a mid-migration (cross-tier encode) +// volume keeps its fresh shards in the SOURCE disk's bucket while the balance +// runs against the target type, so a bucket-scoped delete would miss them and +// the dry-run model would count a moved shard twice. +func (ecNode *EcNode) DeleteEcVolumeShards(vid needle.VolumeId, shardIds []erasure_coding.ShardId) *EcNode { - if diskInfo, found := ecNode.Info.DiskInfos[string(diskType)]; found { + for _, diskInfo := range ecNode.Info.DiskInfos { + if diskInfo == nil { + continue + } for _, eci := range diskInfo.EcShardInfos { if needle.VolumeId(eci.Id) == vid { si := erasure_coding.ShardsInfoFromVolumeEcShardInformationMessage(eci) @@ -750,6 +759,9 @@ type ecBalancer struct { // volumeIds narrows the plan to these ec volume ids; nil balances every volume // of the selected collections. volumeIds map[uint32]bool + // migratingVolumeIds are mid-encode volumes whose shards are ingested from + // every disk-type bucket; see EcBalance. + migratingVolumeIds map[uint32]bool } // excludeNodes is a set of server addresses kept out of the balance as copy/move @@ -760,7 +772,16 @@ type ecBalancer struct { // // volumeIds, when non-empty, restricts the plan to those ec volume ids; empty // balances every volume of the given collections. -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) (err error) { +// +// migratingVolumeIds names volumes whose shards are ingested from EVERY +// disk-type bucket, not just the diskType one. ec.encode passes its batch: +// shard generation writes beside the source .dat, so a cross-tier encode +// (source on hdd, -diskType=ssd) leaves the fresh shards in the source bucket, +// where a target-bucket-only balance cannot see them — it plans no moves and +// the encode's spread guard aborts. Everything else keeps the bucket filter, +// 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 { @@ -795,6 +816,13 @@ func EcBalance(env *Env, collections []string, dc string, ecReplicaPlacement *su volumeIdFilter[uint32(vid)] = true } } + var migrating map[uint32]bool + if len(migratingVolumeIds) > 0 { + migrating = make(map[uint32]bool, len(migratingVolumeIds)) + for _, vid := range migratingVolumeIds { + migrating[uint32(vid)] = true + } + } ecb := &ecBalancer{ env: env, @@ -805,6 +833,7 @@ func EcBalance(env *Env, collections []string, dc string, ecReplicaPlacement *su ioBytePerSecond: ioBytePerSecond, diskType: diskType, volumeIds: volumeIdFilter, + migratingVolumeIds: migrating, } if len(collections) == 0 { @@ -823,7 +852,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) + topo, volumeRatio, selected := toBalancerTopology(ecb.ecNodes, collections, ecb.diskType, ecb.volumeIds, ecb.migratingVolumeIds) if len(ecb.volumeIds) > 0 { requested := make([]uint32, 0, len(ecb.volumeIds)) for vid := range ecb.volumeIds { @@ -875,7 +904,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) (*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) (*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 @@ -898,21 +927,29 @@ func toBalancerTopology(ecNodes []*EcNode, collections []string, diskType types. for diskId, d := range en.Disks { node.AddDisk(diskId, d.DiskType, d.FreeEcSlots, d.EcShardCount) } - diskInfo, found := en.Info.DiskInfos[string(diskType)] - if !found { - continue - } - for _, eci := range diskInfo.EcShardInfos { - if len(allowed) > 0 && !allowed[eci.Collection] { + for diskTypeKey, diskInfo := range en.Info.DiskInfos { + if diskInfo == nil { continue } - if volumeIds != nil && !volumeIds[eci.Id] { - continue - } - selected[eci.Id] = true - node.AddShards(eci.Id, eci.Collection, eci.DiskId, erasure_coding.ShardBits(eci.EcIndexBits)) - if d, p := ecbalancer.VolumeShardRatio(eci); d > 0 || p > 0 { - volRatios[volRatioKey{eci.Collection, eci.Id}] = [2]int{d, p} + for _, eci := range diskInfo.EcShardInfos { + // A migrating (mid-encode) volume's fresh shards sit beside the + // source .dat, in whatever bucket that disk belongs to; ingest + // them regardless so the balance can move them onto the target + // disk type. All other volumes keep the bucket filter. + if diskTypeKey != string(diskType) && !migratingVolumeIds[eci.Id] { + continue + } + if len(allowed) > 0 && !allowed[eci.Collection] { + continue + } + if volumeIds != nil && !volumeIds[eci.Id] { + continue + } + selected[eci.Id] = true + node.AddShards(eci.Id, eci.Collection, eci.DiskId, erasure_coding.ShardBits(eci.EcIndexBits)) + if d, p := ecbalancer.VolumeShardRatio(eci); d > 0 || p > 0 { + volRatios[volRatioKey{eci.Collection, eci.Id}] = [2]int{d, p} + } } } } @@ -1021,7 +1058,7 @@ func (ecb *ecBalancer) executeMove(byID map[string]*EcNode, m ecbalancer.Move) e if m.Phase == "dedup" { fmt.Printf("dedup: delete ec shard %d.%d on %s\n", vid, shardId, m.SourceNode) if !ecb.applyBalancing { - src.DeleteEcVolumeShards(vid, shardIds, ecb.diskType) + src.DeleteEcVolumeShards(vid, shardIds) return nil } grpcDialOption := ecb.env.GrpcDialOption diff --git a/weed/ec/ec_decode.go b/weed/ec/ec_decode.go index 57665614e..c236b35d8 100644 --- a/weed/ec/ec_decode.go +++ b/weed/ec/ec_decode.go @@ -27,12 +27,12 @@ func DoEcDecode(env *Env, topoInfo *master_pb.TopologyInfo, collection string, v } // find volume location - nodeToEcShardsInfo, dataShards := collectEcNodeShardsInfo(topoInfo, vid, diskType) + nodeToEcShardsInfo, dataShards := collectEcNodeShardsInfo(topoInfo, vid) fmt.Printf("ec volume %d shard locations: %+v\n", vid, nodeToEcShardsInfo) if len(nodeToEcShardsInfo) == 0 { - return fmt.Errorf("no EC shards found for volume %d (diskType %s)", vid, diskType.ReadableString()) + return fmt.Errorf("no EC shards found for volume %d", vid) } var originalShardCounts map[pb.ServerAddress]int @@ -214,6 +214,23 @@ func collectEcShards(env *Env, nodeToShardsInfo map[pb.ServerAddress]*erasure_co return "", fmt.Errorf("no eligible target datanodes available to decode volume %d", vid) } + // Trust only shards the target can actually serve: an interrupted earlier + // decode or balance can leave the master believing the target holds a + // shard whose file never landed. A phantom entry here would exclude the + // shard from the copy set and the decode would then fail with "missing + // shard"; probing the target's live inventory makes the re-run re-copy it. + if present, probeErr := erasure_coding.CollectShardsOnServer(context.Background(), collection, uint32(vid), string(targetNodeLocation), env.GrpcDialOption); probeErr == nil { + confirmed := erasure_coding.NewShardsInfo() + for _, sid := range existingShardsInfo.Ids() { + if present.Has(sid) { + confirmed.Set(erasure_coding.NewShardInfo(sid, 0)) + } + } + existingShardsInfo = confirmed + } else { + fmt.Printf("collectEcShards: probe %s inventory for volume %d: %v (keeping the topology's view)\n", targetNodeLocation, vid, probeErr) + } + fmt.Printf("collectEcShards: ec volume %d collect shards to %s from: %+v\n", vid, targetNodeLocation, nodeToShardsInfo) copiedShardsInfo := erasure_coding.NewShardsInfo() @@ -294,14 +311,19 @@ func CollectEcShardIds(topoInfo *master_pb.TopologyInfo, collectionRegex *regexp return } -func collectEcNodeShardsInfo(topoInfo *master_pb.TopologyInfo, vid needle.VolumeId, diskType types.DiskType) (map[pb.ServerAddress]*erasure_coding.ShardsInfo, int) { +func collectEcNodeShardsInfo(topoInfo *master_pb.TopologyInfo, vid needle.VolumeId) (map[pb.ServerAddress]*erasure_coding.ShardsInfo, int) { res := make(map[pb.ServerAddress]*erasure_coding.ShardsInfo) EachDataNode(topoInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) { - if diskInfo, found := dn.DiskInfos[string(diskType)]; found { - // A node may report several EcShardInfos for one volume — one per - // physical disk holding shards of it (multi-disk nodes). Union them - // rather than overwriting, or only the last disk's shards survive and - // the node looks like it is missing shards it actually has. + // Union across ALL disk-type buckets and, within a node, across its + // physical disks. Shards sit wherever encode generation and balance + // left them — a cross-tier encode leaves them in the source bucket, a + // partial migration straddles buckets — and a decode that only looks + // at one bucket reports a decodable volume as having no shards at all. + // (Same rationale as the encode's shard verification.) + for _, diskInfo := range dn.DiskInfos { + if diskInfo == nil { + continue + } for _, v := range diskInfo.EcShardInfos { if v.Id == uint32(vid) { addr := pb.NewServerAddressFromDataNode(dn) diff --git a/weed/ec/ec_encode.go b/weed/ec/ec_encode.go index 3994a6ad9..8dfe9beb8 100644 --- a/weed/ec/ec_encode.go +++ b/weed/ec/ec_encode.go @@ -117,7 +117,7 @@ func ProcessEcEncodeBatch(env *Env, writer io.Writer, volumeIds []needle.VolumeI // safely verified and deleted without waiting for all batches to finish. // skippedNodes are excluded so a recovered node's stale orphan is never // paired with a new-generation shard. - if err := EcBalance(env, balanceCollections, "", rp, diskType, maxParallelization, 0, applyBalancing, skippedNodes, nil); err != nil { + if err := EcBalance(env, balanceCollections, "", rp, diskType, maxParallelization, 0, applyBalancing, skippedNodes, nil, volumeIds); err != nil { return fmt.Errorf("re-balance ec shards for collection(s) %v: %w", balanceCollections, err) } if err := verifyEcShardsBeforeDelete(env, volumeIds, diskType, applyBalancing); err != nil { diff --git a/weed/ec/ec_encode_test.go b/weed/ec/ec_encode_test.go index ce9b70e52..d57408f5d 100644 --- a/weed/ec/ec_encode_test.go +++ b/weed/ec/ec_encode_test.go @@ -449,10 +449,15 @@ func TestEcShardCountIgnoresDiskTypeOfTheShards(t *testing.T) { assert.NoError(t, err, "a complete shard set must not read as unrecoverable") assert.False(t, degraded) - // The disk-scoped view is what the check used to consult, and it is blind - // to this volume entirely. - scoped, _ := collectEcNodeShardsInfo(topo, needle.VolumeId(1), types.ToDiskType("")) - assert.Empty(t, scoped, "the hdd-scoped view cannot see ssd shards") + // The decode's node view unions across buckets for the same reason: a + // decode scoped to the hdd bucket would report this decodable volume as + // having no shards at all. + byNode, _ := collectEcNodeShardsInfo(topo, needle.VolumeId(1)) + assert.Len(t, byNode, 1, "decode discovery must see shards on a non-default medium") + for _, si := range byNode { + assert.Equal(t, erasure_coding.TotalShardsCount, si.Count(), + "decode discovery must include every shard on the non-default medium") + } } // The message an aborted deletion leaves behind is all the operator has to go diff --git a/weed/server/volume_grpc_erasure_coding.go b/weed/server/volume_grpc_erasure_coding.go index 9004bf964..45cd2e6fe 100644 --- a/weed/server/volume_grpc_erasure_coding.go +++ b/weed/server/volume_grpc_erasure_coding.go @@ -990,10 +990,12 @@ func (vs *VolumeServer) VolumeEcShardsToVolume(ctx context.Context, req *volume_ return nil, status.Errorf(codes.FailedPrecondition, "ec volume %d %s", req.VolumeId, erasure_coding.EcNoLiveEntriesSubstring) } - // calculate .dat file size - datFileSize, err := erasure_coding.FindDatFileSize(dataBaseFileName, indexBaseFileName) + // calculate .dat file size. Pass shard 0's resolved path: on a multi-disk + // server the volume's shards can sit on several disks, so the .ec00 is not + // necessarily beside the EcVolume's own base path. + datFileSize, err := erasure_coding.FindDatFileSize(shardFileNames[0], indexBaseFileName) if err != nil { - return nil, fmt.Errorf("FindDatFileSize %s: %v", dataBaseFileName, err) + return nil, fmt.Errorf("FindDatFileSize %s: %v", shardFileNames[0], err) } // The shard block layout was fixed by the .dat size at encode time (recorded diff --git a/weed/shell/command_ec_common.go b/weed/shell/command_ec_common.go index 5165baa2b..0af548945 100644 --- a/weed/shell/command_ec_common.go +++ b/weed/shell/command_ec_common.go @@ -138,7 +138,7 @@ func moveMountedShardToEcNode(commandEnv *CommandEnv, existingLocation *EcNode, // EcBalance balances EC shards across the cluster; see ec.EcBalance for the // excludeNodes and volumeIds semantics. func EcBalance(commandEnv *CommandEnv, 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) (err error) { - return ec.EcBalance(commandEnv.ecEnv(), collections, dc, ecReplicaPlacement, diskType, maxParallelization, ioBytePerSecond, applyBalancing, excludeNodes, volumeIds) + return ec.EcBalance(commandEnv.ecEnv(), collections, dc, ecReplicaPlacement, diskType, maxParallelization, ioBytePerSecond, applyBalancing, excludeNodes, volumeIds, nil) } // compileCollectionPattern compiles a regex pattern for collection matching. diff --git a/weed/storage/erasure_coding/ec_decoder.go b/weed/storage/erasure_coding/ec_decoder.go index 6d7c9d9b6..5b08a232f 100644 --- a/weed/storage/erasure_coding/ec_decoder.go +++ b/weed/storage/erasure_coding/ec_decoder.go @@ -93,11 +93,15 @@ func WriteIdxFileFromEcIndex(baseFileName string) (err error) { // FindDatFileSize calculate .dat file size from max offset entry // there may be extra deletions after that entry // but they are deletions anyway -func FindDatFileSize(dataBaseFileName, indexBaseFileName string) (datSize int64, err error) { +// shard0FileName is the actual path of the .ec00 shard file, which on a +// multi-disk server may sit on a different disk than the EcVolume's own base +// path — the store registers shards per disk, so the caller must pass the +// path CollectEcShards resolved rather than deriving it from a base name. +func FindDatFileSize(shard0FileName, indexBaseFileName string) (datSize int64, err error) { - version, err := readEcVolumeVersion(dataBaseFileName) + version, err := readEcVolumeVersion(shard0FileName) if err != nil { - return 0, fmt.Errorf("read ec volume %s version: %v", dataBaseFileName, err) + return 0, fmt.Errorf("read ec volume %s version: %v", shard0FileName, err) } // Safety: ensure datSize is at least SuperBlockSize. While the caller typically @@ -122,19 +126,19 @@ func FindDatFileSize(dataBaseFileName, indexBaseFileName string) (datSize int64, return } -func readEcVolumeVersion(baseFileName string) (version needle.Version, err error) { +func readEcVolumeVersion(shard0FileName string) (version needle.Version, err error) { // find volume version - datFile, err := os.OpenFile(baseFileName+".ec00", os.O_RDONLY, 0644) + datFile, err := os.OpenFile(shard0FileName, os.O_RDONLY, 0644) if err != nil { - return 0, fmt.Errorf("open ec volume %s superblock: %v", baseFileName, err) + return 0, fmt.Errorf("open ec volume %s superblock: %v", shard0FileName, err) } datBackend := backend.NewDiskFile(datFile) superBlock, err := super_block.ReadSuperBlock(datBackend) datBackend.Close() if err != nil { - return 0, fmt.Errorf("read ec volume %s superblock: %v", baseFileName, err) + return 0, fmt.Errorf("read ec volume %s superblock: %v", shard0FileName, err) } return superBlock.Version, nil diff --git a/weed/storage/erasure_coding/ec_decoder_multidisk_test.go b/weed/storage/erasure_coding/ec_decoder_multidisk_test.go new file mode 100644 index 000000000..6a3631eef --- /dev/null +++ b/weed/storage/erasure_coding/ec_decoder_multidisk_test.go @@ -0,0 +1,54 @@ +package erasure_coding + +import ( + "os" + "path/filepath" + "testing" +) + +// TestFindDatFileSizeWithRelocatedShard0: on a multi-disk server a volume's +// shards can sit on several disks, so the .ec00 is not necessarily beside the +// EcVolume's base path. FindDatFileSize takes the resolved shard-0 path; it +// must work when that path points to a different directory than the index. +func TestFindDatFileSizeWithRelocatedShard0(t *testing.T) { + diskA := t.TempDir() + diskB := t.TempDir() + + for _, ext := range []string{".dat", ".idx"} { + data, err := os.ReadFile("1" + ext) + if err != nil { + t.Fatalf("read fixture 1%s: %v", ext, err) + } + if err := os.WriteFile(filepath.Join(diskA, "1"+ext), data, 0o644); err != nil { + t.Fatal(err) + } + } + + base := filepath.Join(diskA, "1") + ctx := NewDefaultECContext("", 0) + if _, err := generateEcFiles(base, 50, largeBlockSize, smallBlockSize, ctx); err != nil { + t.Fatalf("generateEcFiles: %v", err) + } + if err := WriteSortedFileFromIdx(base, ".ecx"); err != nil { + t.Fatalf("WriteSortedFileFromIdx: %v", err) + } + + wantSize, err := FindDatFileSize(base+".ec00", base) + if err != nil { + t.Fatalf("FindDatFileSize with co-located shard 0: %v", err) + } + + // Relocate shard 0 to a sibling disk, as a balance or copy can leave it. + moved := filepath.Join(diskB, "1.ec00") + if err := os.Rename(base+".ec00", moved); err != nil { + t.Fatal(err) + } + + gotSize, err := FindDatFileSize(moved, base) + if err != nil { + t.Fatalf("FindDatFileSize with relocated shard 0: %v", err) + } + if gotSize != wantSize { + t.Fatalf("dat size changed with relocated shard 0: got %d want %d", gotSize, wantSize) + } +} diff --git a/weed/storage/erasure_coding/verification.go b/weed/storage/erasure_coding/verification.go index 390e33a93..ac6fa6ecb 100644 --- a/weed/storage/erasure_coding/verification.go +++ b/weed/storage/erasure_coding/verification.go @@ -150,15 +150,12 @@ func SummarizeShardInventory(perServer map[string]ServerShardInventory) string { // A server that cannot be queried is unknown, not confirmed, and returns an // error: treating an unreachable peer as proof of a surviving copy is how a // network blip becomes data loss. -func VerifyShardsOnServer(ctx context.Context, collection string, volumeID uint32, - server string, shardIDs []uint32, dialOption grpc.DialOption) error { - - if server == "" { - return fmt.Errorf("no server given to verify volume %d shard(s) %v", volumeID, shardIDs) - } - - var present ShardBits - callErr := operation.WithVolumeServerClient(false, pb.ServerAddress(server), dialOption, +// CollectShardsOnServer asks one volume server which shards of the volume it +// actually serves right now — its live inventory, as opposed to what the +// master's possibly stale topology claims for it. +func CollectShardsOnServer(ctx context.Context, collection string, volumeID uint32, + server string, dialOption grpc.DialOption) (present ShardBits, err error) { + err = operation.WithVolumeServerClient(false, pb.ServerAddress(server), dialOption, func(client volume_server_pb.VolumeServerClient) error { resp, e := client.VolumeEcShardsInfo(ctx, &volume_server_pb.VolumeEcShardsInfoRequest{ VolumeId: volumeID, @@ -174,6 +171,17 @@ func VerifyShardsOnServer(ctx context.Context, collection string, volumeID uint3 } return nil }) + return present, err +} + +func VerifyShardsOnServer(ctx context.Context, collection string, volumeID uint32, + server string, shardIDs []uint32, dialOption grpc.DialOption) error { + + if server == "" { + return fmt.Errorf("no server given to verify volume %d shard(s) %v", volumeID, shardIDs) + } + + present, callErr := CollectShardsOnServer(ctx, collection, volumeID, server, dialOption) if callErr != nil { return fmt.Errorf("verify volume %d shard(s) %v on %s: %w", volumeID, shardIDs, server, callErr) }