From 10cdaf381875492a2c752d1038797e96ff18208f Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 13 Jul 2026 17:18:42 -0700 Subject: [PATCH] Introduce weed shell command `ec.check.replication`. (#10328) * Introduce weed shell command `ec.check.replication`. This command performs a quick check of EC volume shard replication, reporting volumes whose shards are over- or under-replicated. Each volume is checked against its own data+parity ratio, obtained via erasure_coding.EcShardsVolume{Data,Parity}Shards, so builds that derive the EC ratio per volume report custom ratios correctly. The name follows the shell's convention (cluster.check, volume.check.disk); the closest normal-volume counterpart is volume.fix.replication. * shell: ec.check.replication reports mixed under+over-replication in both lists --- weed/shell/command_ec_check_replication.go | 283 +++++++++++++++++ .../command_ec_check_replication_test.go | 289 ++++++++++++++++++ weed/storage/erasure_coding/ec_shards_info.go | 17 ++ 3 files changed, 589 insertions(+) create mode 100644 weed/shell/command_ec_check_replication.go create mode 100644 weed/shell/command_ec_check_replication_test.go diff --git a/weed/shell/command_ec_check_replication.go b/weed/shell/command_ec_check_replication.go new file mode 100644 index 000000000..953650dea --- /dev/null +++ b/weed/shell/command_ec_check_replication.go @@ -0,0 +1,283 @@ +package shell + +import ( + "flag" + "fmt" + "io" + "slices" + "strconv" + "strings" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" +) + +func init() { + Commands = append(Commands, &commandEcCheckReplication{}) +} + +type commandEcCheckReplication struct { +} + +func (c *commandEcCheckReplication) Name() string { + return "ec.check.replication" +} + +func (c *commandEcCheckReplication) Help() string { + return `check EC volumes for under- or over-replicated shards + + ec.check.replication [-volumeIds=,...] [-details] + + Reports EC volumes whose shards are: + - under-replicated: at least one shard is missing from every node + - over-replicated: at least one shard has more than one copy, whether on + different nodes or on more than one disk of a node + + Over-replication is normal and transient while ec.balance or ec.encode is + running, since shards are copied before the redundant copies are deleted; + re-check once those operations finish before treating it as a problem. + + Each volume is checked against its own data+parity ratio rather than a fixed + shard count, so volumes with non-default erasure coding ratios are reported + correctly. + + Options: + -volumeIds: comma-separated EC volume IDs to check (default: all EC volumes) + -details: print the per-shard node placement for each flagged volume +` +} + +func (c *commandEcCheckReplication) HasTag(CommandTag) bool { + return false +} + +func (c *commandEcCheckReplication) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) { + checkReplicationCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError) + volumeIDsStr := checkReplicationCommand.String("volumeIds", "", "comma-separated EC volume IDs to process (optional)") + showDetails := checkReplicationCommand.Bool("details", false, "display result details, if available") + + if err = checkReplicationCommand.Parse(args); err != nil { + return err + } + if err = commandEnv.confirmIsLocked(args); err != nil { + return + } + + // Keep execution state in a per-run runner rather than on the long-lived + // command singleton, so concurrent or repeated invocations don't share state. + runner := &ecCheckReplicationRunner{ + writer: writer, + volumeIDMap: map[uint32]bool{}, + } + + if *volumeIDsStr != "" { + for _, vids := range strings.Split(*volumeIDsStr, ",") { + vids = strings.TrimSpace(vids) + if vids == "" { + continue + } + if vid, err := strconv.ParseUint(vids, 10, 32); err == nil { + runner.volumeIDMap[uint32(vid)] = true + } else { + return fmt.Errorf("invalid volume ID %q", vids) + } + } + } + + runner.dataNodes, err = collectDataNodes(commandEnv, 0) + if err != nil { + return err + } + + return runner.checkEcVolumes(*showDetails) +} + +// ecCheckReplicationRunner holds the state for a single ec.check.replication invocation. +type ecCheckReplicationRunner struct { + writer io.Writer + dataNodes []*master_pb.DataNodeInfo + volumeIDMap map[uint32]bool +} + +func (r *ecCheckReplicationRunner) write(format string, a ...any) { + fmt.Fprintf(r.writer, format, a...) +} + +func (r *ecCheckReplicationRunner) isVolumeIDValid(vid uint32) bool { + if len(r.volumeIDMap) == 0 { + return true + } + return r.volumeIDMap[vid] +} + +// ecVolumeShardReplication aggregates the observed shards for a single EC volume, +// together with the data+parity ratio the volume was encoded with. The ratio is +// taken per volume (via erasure_coding.EcShardsVolume*Shards) so custom EC +// ratios are checked against their own expected shard count. +type ecVolumeShardReplication struct { + dataShards int + parityShards int + // shardAddresses maps a shard id to the sorted node addresses hosting it. + shardAddresses map[erasure_coding.ShardId][]string +} + +func (h *ecVolumeShardReplication) totalShards() int { + return h.dataShards + h.parityShards +} + +// replicaCount is the total number of shard copies observed, counting each +// over-replicated shard once per hosting node. +func (h *ecVolumeShardReplication) replicaCount() int { + n := 0 + for _, addrs := range h.shardAddresses { + n += len(addrs) + } + return n +} + +// unexpectedShardIds returns, in ascending order, any observed shard ids that +// fall outside the volume's expected data+parity range. These are anomalies +// (e.g. a stray shard) and are treated as over-replication. +func (h *ecVolumeShardReplication) unexpectedShardIds() []erasure_coding.ShardId { + var ids []erasure_coding.ShardId + for sid := range h.shardAddresses { + if int(sid) >= h.totalShards() { + ids = append(ids, sid) + } + } + slices.Sort(ids) + return ids +} + +// TODO: check shard sizes? +func (r *ecCheckReplicationRunner) checkEcVolumes(showDetails bool) error { + // collect EC shard placement, keyed by volume id + volumes := map[uint32]*ecVolumeShardReplication{} + for _, dni := range r.dataNodes { + nodeAddress := dni.GetAddress() + for _, di := range dni.GetDiskInfos() { + for _, eci := range di.GetEcShardInfos() { + vid := eci.GetId() + if !r.isVolumeIDValid(vid) { + continue + } + + h, ok := volumes[vid] + if !ok { + // all shards of a volume share one ratio; take it from the + // first shard message seen for the volume. + h = &ecVolumeShardReplication{ + dataShards: erasure_coding.EcShardsVolumeDataShards(eci), + parityShards: erasure_coding.EcShardsVolumeParityShards(eci), + shardAddresses: map[erasure_coding.ShardId][]string{}, + } + volumes[vid] = h + } + + sinfo := erasure_coding.ShardsInfoFromVolumeEcShardInformationMessage(eci) + for _, sid := range sinfo.Ids() { + h.shardAddresses[sid] = append(h.shardAddresses[sid], nodeAddress) + } + } + } + } + + if len(volumes) == 0 { + // Asking about specific volume IDs that turn out not to be EC volumes is + // an error; an unfiltered run over a cluster with no EC volumes is a + // legitimate, healthy state rather than a failure. + if len(r.volumeIDMap) > 0 { + return fmt.Errorf("no EC volumes found") + } + r.write("No EC volumes found.\n") + return nil + } + + // keep the shard address lists nicely sorted, for the sake of readability. + for _, h := range volumes { + for _, addrs := range h.shardAddresses { + slices.Sort(addrs) + } + } + + // classify each volume against its own expected shard count + underreplicatedVolumeIDs := []uint32{} + overreplicatedVolumeIDs := []uint32{} + for vid, h := range volumes { + under := false + over := false + for sid := 0; sid < h.totalShards(); sid++ { + switch len(h.shardAddresses[erasure_coding.ShardId(sid)]) { + case 0: + under = true + case 1: + default: + over = true + } + } + // shard ids beyond the expected data+parity range are unexpected extras, + // i.e. redundant data the ratio doesn't call for. + if len(h.unexpectedShardIds()) > 0 { + over = true + } + + // under- and over-replication are independent problems (missing shards risk + // data loss, redundant shards waste space), so a volume exhibiting both is + // reported in both lists. + if under { + underreplicatedVolumeIDs = append(underreplicatedVolumeIDs, vid) + } + if over { + overreplicatedVolumeIDs = append(overreplicatedVolumeIDs, vid) + } + } + slices.Sort(underreplicatedVolumeIDs) + slices.Sort(overreplicatedVolumeIDs) + + // ...and display results + if len(underreplicatedVolumeIDs) == 0 && len(overreplicatedVolumeIDs) == 0 { + r.write("EC volumes are healthy.\n") + return nil + } + + if len(underreplicatedVolumeIDs) != 0 { + r.write("Found %d/%d under-replicated EC volumes: %v\n", len(underreplicatedVolumeIDs), len(volumes), underreplicatedVolumeIDs) + } + if len(overreplicatedVolumeIDs) != 0 { + r.write("Found %d/%d over-replicated EC volumes: %v\n", len(overreplicatedVolumeIDs), len(volumes), overreplicatedVolumeIDs) + } + + if showDetails { + if len(underreplicatedVolumeIDs) != 0 { + r.write("\n") + r.writeShardMaps("under-replicated", underreplicatedVolumeIDs, volumes) + } + if len(overreplicatedVolumeIDs) != 0 { + r.write("\n") + r.writeShardMaps("over-replicated", overreplicatedVolumeIDs, volumes) + } + } + + return nil +} + +func (r *ecCheckReplicationRunner) writeShardMaps(kind string, volumeIDs []uint32, volumes map[uint32]*ecVolumeShardReplication) { + for _, vid := range volumeIDs { + h := volumes[vid] + r.write("Shards map for %s EC volume %v (%d/%d shards):\n", kind, vid, h.replicaCount(), h.totalShards()) + for sid := 0; sid < h.totalShards(); sid++ { + shardTypeDesc := "" + if sid >= h.dataShards { + shardTypeDesc = " (parity)" + } + if addrs, ok := h.shardAddresses[erasure_coding.ShardId(sid)]; ok { + r.write("\t%02d%s => %v\n", sid, shardTypeDesc, addrs) + } else { + r.write("\t%02d%s is missing\n", sid, shardTypeDesc) + } + } + for _, sid := range h.unexpectedShardIds() { + r.write("\t%02d (unexpected) => %v\n", int(sid), h.shardAddresses[sid]) + } + } +} diff --git a/weed/shell/command_ec_check_replication_test.go b/weed/shell/command_ec_check_replication_test.go new file mode 100644 index 000000000..f7621f76d --- /dev/null +++ b/weed/shell/command_ec_check_replication_test.go @@ -0,0 +1,289 @@ +package shell + +import ( + "bytes" + "fmt" + "reflect" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" +) + +// TestEcCheckReplication exercises the ec.check.replication classification and reporting over a +// fake topology. EC volume 1 is fully replicated, EC volume 2 is +// under-replicated, and EC volume 3 is over-replicated. +func TestEcCheckReplication(t *testing.T) { + nodes := []*master_pb.DataNodeInfo{ + { + Address: "localhost:1111", + DiskInfos: map[string]*master_pb.DiskInfo{ + "d1": { + EcShardInfos: []*master_pb.VolumeEcShardInformationMessage{ + { + Id: 1, + EcIndexBits: 0b0000000011111, + }, + { + Id: 2, + EcIndexBits: 0b0000000011111, + }, + { + Id: 3, + EcIndexBits: 0b11111111111111, + }, + }, + }, + }, + }, + { + Address: "localhost:2222", + DiskInfos: map[string]*master_pb.DiskInfo{ + "d2": { + EcShardInfos: []*master_pb.VolumeEcShardInformationMessage{ + { + Id: 1, + EcIndexBits: 0b11111111100000, + }, + { + Id: 2, + EcIndexBits: 0b11111000000000, + }, + { + Id: 3, + EcIndexBits: 0b00000000001111, + }, + }, + }, + }, + }, + } + + testCases := []struct { + name string + showDetails bool + volumeIDMap map[uint32]bool + want string + wantErr error + }{ + { + name: "default", + showDetails: false, + want: `Found 1/3 under-replicated EC volumes: [2] +Found 1/3 over-replicated EC volumes: [3] +`, + }, + { + name: "single healthy volume", + showDetails: false, + volumeIDMap: map[uint32]bool{1: true}, + want: `EC volumes are healthy. +`, + }, + { + name: "filtered by volume ID", + showDetails: false, + volumeIDMap: map[uint32]bool{1: true, 2: true, 5: true}, + want: `Found 1/2 under-replicated EC volumes: [2] +`, + }, + { + name: "no valid volume IDs", + showDetails: false, + volumeIDMap: map[uint32]bool{5: true, 6: true}, + want: "", + wantErr: fmt.Errorf("no EC volumes found"), + }, + { + name: "with details", + showDetails: true, + want: `Found 1/3 under-replicated EC volumes: [2] +Found 1/3 over-replicated EC volumes: [3] + +Shards map for under-replicated EC volume 2 (10/14 shards): + 00 => [localhost:1111] + 01 => [localhost:1111] + 02 => [localhost:1111] + 03 => [localhost:1111] + 04 => [localhost:1111] + 05 is missing + 06 is missing + 07 is missing + 08 is missing + 09 => [localhost:2222] + 10 (parity) => [localhost:2222] + 11 (parity) => [localhost:2222] + 12 (parity) => [localhost:2222] + 13 (parity) => [localhost:2222] + +Shards map for over-replicated EC volume 3 (18/14 shards): + 00 => [localhost:1111 localhost:2222] + 01 => [localhost:1111 localhost:2222] + 02 => [localhost:1111 localhost:2222] + 03 => [localhost:1111 localhost:2222] + 04 => [localhost:1111] + 05 => [localhost:1111] + 06 => [localhost:1111] + 07 => [localhost:1111] + 08 => [localhost:1111] + 09 => [localhost:1111] + 10 (parity) => [localhost:1111] + 11 (parity) => [localhost:1111] + 12 (parity) => [localhost:1111] + 13 (parity) => [localhost:1111] +`, + }, + } + + for _, tc := range testCases { + var buf bytes.Buffer + + runner := &ecCheckReplicationRunner{ + writer: &buf, + dataNodes: nodes, + volumeIDMap: tc.volumeIDMap, + } + gotErr := runner.checkEcVolumes(tc.showDetails) + got := buf.String() + + if got != tc.want { + t.Errorf("%s: got %v, want %v", tc.name, got, tc.want) + } + if !reflect.DeepEqual(gotErr, tc.wantErr) { + t.Errorf("%s: got error %v, want %v", tc.name, gotErr, tc.wantErr) + } + } +} + +// TestEcCheckReplicationNoVolumes checks the empty-topology behavior: an unfiltered +// run over a cluster with no EC volumes is healthy (informational message, no +// error), while a run filtered to volume IDs that are not EC volumes errors. +func TestEcCheckReplicationNoVolumes(t *testing.T) { + // a data node reporting a normal volume but no EC shards + nodes := []*master_pb.DataNodeInfo{ + { + Address: "localhost:1111", + DiskInfos: map[string]*master_pb.DiskInfo{"d1": {}}, + }, + } + + t.Run("unfiltered", func(t *testing.T) { + var buf bytes.Buffer + runner := &ecCheckReplicationRunner{writer: &buf, dataNodes: nodes} + if err := runner.checkEcVolumes(false); err != nil { + t.Errorf("got error %v, want nil", err) + } + if got, want := buf.String(), "No EC volumes found.\n"; got != want { + t.Errorf("got %q, want %q", got, want) + } + }) + + t.Run("filtered no match", func(t *testing.T) { + var buf bytes.Buffer + runner := &ecCheckReplicationRunner{writer: &buf, dataNodes: nodes, volumeIDMap: map[uint32]bool{5: true}} + if err := runner.checkEcVolumes(false); err == nil { + t.Errorf("got nil error, want %q", "no EC volumes found") + } + if got := buf.String(); got != "" { + t.Errorf("got output %q, want none", got) + } + }) +} + +// TestEcCheckReplicationUnexpectedShard verifies that a shard id beyond a volume's +// expected data+parity range is flagged as over-replication and surfaced in the +// details map, rather than being silently ignored. Volume 20 is a healthy 10+4 +// (shards 0..13 on nodeA) with a stray shard 14 on nodeB. +func TestEcCheckReplicationUnexpectedShard(t *testing.T) { + nodes := []*master_pb.DataNodeInfo{ + { + Address: "nodeA", + DiskInfos: map[string]*master_pb.DiskInfo{ + "d1": { + EcShardInfos: []*master_pb.VolumeEcShardInformationMessage{ + {Id: 20, EcIndexBits: 0b11111111111111}, // shards 0..13 + }, + }, + }, + }, + { + Address: "nodeB", + DiskInfos: map[string]*master_pb.DiskInfo{ + "d2": { + EcShardInfos: []*master_pb.VolumeEcShardInformationMessage{ + {Id: 20, EcIndexBits: 0b100000000000000}, // stray shard 14 + }, + }, + }, + }, + } + + want := `Found 1/1 over-replicated EC volumes: [20] + +Shards map for over-replicated EC volume 20 (15/14 shards): + 00 => [nodeA] + 01 => [nodeA] + 02 => [nodeA] + 03 => [nodeA] + 04 => [nodeA] + 05 => [nodeA] + 06 => [nodeA] + 07 => [nodeA] + 08 => [nodeA] + 09 => [nodeA] + 10 (parity) => [nodeA] + 11 (parity) => [nodeA] + 12 (parity) => [nodeA] + 13 (parity) => [nodeA] + 14 (unexpected) => [nodeB] +` + + var buf bytes.Buffer + runner := &ecCheckReplicationRunner{writer: &buf, dataNodes: nodes} + if err := runner.checkEcVolumes(true); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got := buf.String(); got != want { + t.Errorf("unexpected shard:\n got: %q\nwant: %q", got, want) + } +} + +// TestEcCheckReplicationMixedAnomalies verifies that a volume with both a missing +// shard and a duplicated shard is reported in both lists. Volume 30 has shard 5 +// missing everywhere (under-replicated) and shard 0 on two nodes (over-replicated). +func TestEcCheckReplicationMixedAnomalies(t *testing.T) { + nodes := []*master_pb.DataNodeInfo{ + { + Address: "nodeA", + DiskInfos: map[string]*master_pb.DiskInfo{ + "d1": { + EcShardInfos: []*master_pb.VolumeEcShardInformationMessage{ + {Id: 30, EcIndexBits: 0b11111111011111}, // shards 0..13 except 5 + }, + }, + }, + }, + { + Address: "nodeB", + DiskInfos: map[string]*master_pb.DiskInfo{ + "d2": { + EcShardInfos: []*master_pb.VolumeEcShardInformationMessage{ + {Id: 30, EcIndexBits: 0b00000000000001}, // duplicate of shard 0 + }, + }, + }, + }, + } + + want := `Found 1/1 under-replicated EC volumes: [30] +Found 1/1 over-replicated EC volumes: [30] +` + + var buf bytes.Buffer + runner := &ecCheckReplicationRunner{writer: &buf, dataNodes: nodes} + if err := runner.checkEcVolumes(false); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got := buf.String(); got != want { + t.Errorf("mixed anomalies:\n got: %q\nwant: %q", got, want) + } +} diff --git a/weed/storage/erasure_coding/ec_shards_info.go b/weed/storage/erasure_coding/ec_shards_info.go index 8d6496c76..dc8775dbb 100644 --- a/weed/storage/erasure_coding/ec_shards_info.go +++ b/weed/storage/erasure_coding/ec_shards_info.go @@ -119,6 +119,23 @@ func EcShardsTotalSize(vi *master_pb.VolumeEcShardInformationMessage) int64 { return total } +// EcShardsVolumeDataShards returns the number of data shards for the EC volume +// described by vi. Open-source SeaweedFS always uses the fixed 10+4 layout, so +// this returns DataShardsCount. It is defined as a per-volume accessor (rather +// than referencing DataShardsCount directly at every call site) so callers such +// as ec.check.replication stay correct when a build derives the ratio per volume instead +// of from the global constant. +func EcShardsVolumeDataShards(vi *master_pb.VolumeEcShardInformationMessage) int { + return DataShardsCount +} + +// EcShardsVolumeParityShards returns the number of parity shards for the EC +// volume described by vi. As with EcShardsVolumeDataShards, open-source +// SeaweedFS uses the fixed 10+4 layout, so this returns ParityShardsCount. +func EcShardsVolumeParityShards(vi *master_pb.VolumeEcShardInformationMessage) int { + return ParityShardsCount +} + // EcShardsDataSize returns the sum of sizes for data shards only (parity // shards excluded). Data shards are those with id < dataShards; all higher // shard ids are treated as parity. Passing dataShards <= 0 falls back to