Files
seaweedfs/weed/ec/ec_destructive_guards_test.go
Chris LuandGitHub 944d967502 refactor: extract EC orchestration into a shared weed/ec package (#10760)
* shell: move ErrorWaitGroup to weed/util

* shell: remove unused CandidateEcNode and EcRack types

* ec: extract EC orchestration logic from weed/shell into weed/ec

Move the EC node/topology model, balance engine, encode pipeline, decode
pipeline, and rebuild engine into a new weed/ec package so shell commands
and maintenance workers can share the logic. Shell commands keep flag
parsing and delegate through a small ec.Env (dial option, topology fetch,
volume locations, lock check). Tests move along with the code.

* shell: remove unused proportional-rebalance type stubs

* ec: move scrub, replication check, and shard unmount engines into weed/ec

* worker: share the EC generation-aware shard counter from weed/ec

* ec: gofmt

* shell: drop EC aliases with no remaining callers

* ec: guard a missing topology hook and nil disk entries in topology helpers

* ec: drop trailing newlines from decode error strings

* ec: re-check the shell lock before applying shard unmounts

* shell: trim -node entries in ec.scrub
2026-08-14 13:54:12 -07:00

232 lines
8.4 KiB
Go

package ec
import (
"bytes"
"strings"
"testing"
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
"github.com/seaweedfs/seaweedfs/weed/storage/types"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
)
// topoWith builds a one-node topology snapshot where regularVids appear as
// regular volumes (.dat) and ecVids appear as EC shards.
func topoWith(regularVids, ecVids []uint32) *master_pb.TopologyInfo {
di := &master_pb.DiskInfo{Type: "hdd"}
for _, v := range regularVids {
di.VolumeInfos = append(di.VolumeInfos, &master_pb.VolumeInformationMessage{Id: v})
}
for _, v := range ecVids {
di.EcShardInfos = append(di.EcShardInfos, &master_pb.VolumeEcShardInformationMessage{Id: v})
}
return &master_pb.TopologyInfo{
DataCenterInfos: []*master_pb.DataCenterInfo{{
RackInfos: []*master_pb.RackInfo{{
DataNodeInfos: []*master_pb.DataNodeInfo{{
Id: "n1",
DiskInfos: map[string]*master_pb.DiskInfo{"hdd": di},
}},
}},
}},
}
}
// TestAssertEncodableRegularVolumes: ec.encode must refuse an already-EC volume
// (no .dat — re-encoding would destroy its shards) and an unknown id, while
// allowing a regular volume and a failed-encode retry (regular .dat + orphans).
func TestAssertEncodableRegularVolumes(t *testing.T) {
cases := []struct {
name string
regular, ec []uint32
vids []needle.VolumeId
wantErrSubstr string
}{
{"regular volume", []uint32{1}, nil, []needle.VolumeId{1}, ""},
{"already EC", nil, []uint32{1}, []needle.VolumeId{1}, "already EC"},
{"regular plus stale orphan shards", []uint32{1}, []uint32{1}, []needle.VolumeId{1}, ""},
{"unknown id", nil, nil, []needle.VolumeId{1}, "not found"},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
err := AssertEncodableRegularVolumes(topoWith(tc.regular, tc.ec), tc.vids)
if tc.wantErrSubstr == "" {
if err != nil {
t.Fatalf("want nil error, got %v", err)
}
return
}
if err == nil || !strings.Contains(err.Error(), tc.wantErrSubstr) {
t.Fatalf("want error containing %q, got %v", tc.wantErrSubstr, err)
}
})
}
}
func newDryRunRebuilder(log *bytes.Buffer, nodes ...*EcNode) *ecRebuilder {
return &ecRebuilder{
env: &Env{},
ecNodes: nodes,
writer: log,
applyChanges: false,
}
}
// TestPrepareDataToRecover_DryRunRecoverableNoSideEffects: a dry-run on a
// recoverable degraded volume the rebuilder holds none of must still succeed
// and report its plan, while recording zero shards to delete (so the deferred
// cleanup issues no VolumeEcShardsDelete RPC).
func TestPrepareDataToRecover_DryRunRecoverableNoSideEffects(t *testing.T) {
var log bytes.Buffer
rebuilder := newEcNode("dc1", "rack1", "rebuilder", 100)
remote := newEcNode("dc1", "rack1", "remote", 100).
addEcVolumeAndShardsForTest(1, "c1", []erasure_coding.ShardId{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11})
erb := newDryRunRebuilder(&log, rebuilder, remote)
locations := make(EcShardLocations, erasure_coding.MaxShardCount)
for i := 0; i < 12; i++ {
locations[i] = []*EcNode{remote}
}
copied, local, err := erb.prepareDataToRecover(rebuilder, "c1", needle.VolumeId(1), locations)
if err != nil {
t.Fatalf("dry-run on a recoverable volume must not error: %v", err)
}
if len(copied) != 0 {
t.Errorf("dry-run must record no copied shards (deletion set), got %v", copied)
}
if len(local) != 0 {
t.Errorf("rebuilder holds no shards locally, got %v", local)
}
if !strings.Contains(log.String(), "would copy") {
t.Errorf("dry-run should report the would-copy plan, got:\n%s", log.String())
}
}
// TestPrepareDataToRecover_UnionsLocalShardsAcrossDisks: shards the rebuilder
// holds on more than one disk must all count as local (not copied), so the
// per-disk overwrite that made one disk's shards look remote (then self-copied
// with O_TRUNC and node-wide deleted) cannot happen.
func TestPrepareDataToRecover_UnionsLocalShardsAcrossDisks(t *testing.T) {
var log bytes.Buffer
rebuilder := newEcNode("dc1", "rack1", "rebuilder", 100).
addEcVolumeAndShardsForTest(1, "c1", []erasure_coding.ShardId{0, 1}).
AddEcVolumeShards(needle.VolumeId(1), "c1", []erasure_coding.ShardId{5}, types.SsdType)
remote := newEcNode("dc1", "rack1", "remote", 100)
erb := newDryRunRebuilder(&log, rebuilder)
locations := make(EcShardLocations, erasure_coding.MaxShardCount)
for i := 0; i < 12; i++ {
locations[i] = []*EcNode{remote}
}
locations[0] = []*EcNode{rebuilder}
locations[1] = []*EcNode{rebuilder}
locations[5] = []*EcNode{rebuilder}
copied, local, err := erb.prepareDataToRecover(rebuilder, "c1", needle.VolumeId(1), locations)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
missing := map[erasure_coding.ShardId]bool{0: true, 1: true, 5: true}
for _, s := range local {
delete(missing, s)
}
if len(missing) != 0 {
t.Errorf("shards %v should be local (union across disks); local=%v", missing, local)
}
for _, s := range copied {
if s == 0 || s == 1 || s == 5 {
t.Errorf("local shard %d must never be in the copy/delete set", s)
}
}
}
// TestCountLocalShards_UnionsAcrossDisks: slot accounting must union shards
// across the node's disks like prepareDataToRecover does, or slotsNeeded is
// overstated and selectAndReserveRebuilder can reject a valid node.
func TestCountLocalShards_UnionsAcrossDisks(t *testing.T) {
var log bytes.Buffer
rebuilder := newEcNode("dc1", "rack1", "rebuilder", 100).
addEcVolumeAndShardsForTest(1, "c1", []erasure_coding.ShardId{0, 1}).
AddEcVolumeShards(needle.VolumeId(1), "c1", []erasure_coding.ShardId{5}, types.SsdType)
erb := newDryRunRebuilder(&log, rebuilder)
if got := erb.countLocalShards(rebuilder, "c1", needle.VolumeId(1)); got != 3 {
t.Errorf("countLocalShards must union across disks: want 3, got %d", got)
}
}
// TestPrepareDataToRecover_SelfSourceNotCopied: if the rebuilder is the only
// listed holder of a shard it does not report locally, it must be treated as
// local rather than copied onto itself and then deleted.
func TestPrepareDataToRecover_SelfSourceNotCopied(t *testing.T) {
var log bytes.Buffer
rebuilder := newEcNode("dc1", "rack1", "rebuilder", 100).
addEcVolumeAndShardsForTest(1, "c1", []erasure_coding.ShardId{0, 1, 2, 3, 4, 6, 7, 8, 9, 10})
erb := newDryRunRebuilder(&log, rebuilder)
locations := make(EcShardLocations, erasure_coding.MaxShardCount)
for _, s := range []int{0, 1, 2, 3, 4, 6, 7, 8, 9, 10} {
locations[s] = []*EcNode{rebuilder}
}
locations[5] = []*EcNode{rebuilder} // sole holder is the rebuilder, not in its reported shards
copied, local, err := erb.prepareDataToRecover(rebuilder, "c1", needle.VolumeId(1), locations)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
for _, s := range copied {
if s == 5 {
t.Errorf("self-sourced shard 5 must not be copied/deleted")
}
}
found := false
for _, s := range local {
if s == 5 {
found = true
}
}
if !found {
t.Errorf("self-sourced shard 5 should be classified local, got local=%v", local)
}
}
func TestPrepareDataToRecover_ApplyCopyFailureDoesNotCountAsRecoverable(t *testing.T) {
var log bytes.Buffer
rebuilder := newEcNode("dc1", "rack1", "127.0.0.1:1", 100).
addEcVolumeAndShardsForTest(1, "c1", []erasure_coding.ShardId{0, 1, 2, 3, 4, 5, 6, 7, 8})
remote := newEcNode("dc1", "rack1", "127.0.0.1:2", 100).
addEcVolumeAndShardsForTest(1, "c1", []erasure_coding.ShardId{9})
erb := &ecRebuilder{
env: &Env{
GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
},
ecNodes: []*EcNode{rebuilder, remote},
writer: &log,
applyChanges: true,
}
locations := make(EcShardLocations, erasure_coding.MaxShardCount)
for i := 0; i < 9; i++ {
locations[i] = []*EcNode{rebuilder}
}
locations[9] = []*EcNode{remote}
copied, local, err := erb.prepareDataToRecover(rebuilder, "c1", needle.VolumeId(1), locations)
if err == nil {
t.Fatalf("copy failure must leave only 9 usable shards and fail recoverability gate; copied=%v local=%v log:\n%s", copied, local, log.String())
}
if !strings.Contains(err.Error(), "9 shards are not enough") {
t.Fatalf("expected recoverability error to count only successful/local shards, got %v; log:\n%s", err, log.String())
}
if len(copied) != 0 {
t.Fatalf("failed copy must not be recorded as copied/deletable, got %v", copied)
}
if !strings.Contains(log.String(), "failed to copy 1.9") {
t.Fatalf("expected failed copy to be logged, got:\n%s", log.String())
}
}