Github-Mirror/seaweedfs
1
0 Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-10-16 02:58:35 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor: extract EC orchestration into a shared weed/ec package (#10760)

Browse Source 
* 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
This commit is contained in:
Chris Lu authored and GitHub committed 2026-08-14 13:54:12 -07:00
1 parent f66d6ffc4a
commit 944d967502
30 files changed
+3830 -3648
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

weed/shell/command_ec_check_replication.go
+5 -201
View File
@@ -4,12 +4,10 @@ import (
"flag"
"fmt"
"io"
"slices"
"strconv"
"strings"
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
"github.com/seaweedfs/seaweedfs/weed/ec"
)
func init() {
@@ -63,13 +61,7 @@ func (c *commandEcCheckReplication) Do(args []string, commandEnv *CommandEnv, wr
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{},
}
volumeIDMap := map[uint32]bool{}
if *volumeIDsStr != "" {
for _, vids := range strings.Split(*volumeIDsStr, ",") {
vids = strings.TrimSpace(vids)
@@ -77,207 +69,19 @@ func (c *commandEcCheckReplication) Do(args []string, commandEnv *CommandEnv, wr
continue
}
if vid, err := strconv.ParseUint(vids, 10, 32); err == nil {
runner.volumeIDMap[uint32(vid)] = true
volumeIDMap[uint32(vid)] = true
} else {
return fmt.Errorf("invalid volume ID %q", vids)
}
}
}
runner.dataNodes, err = collectDataNodes(commandEnv, 0)
dataNodes, err := collectDataNodes(commandEnv, 0)
if err != nil {
return err
}
return runner.checkEcVolumes(*showDetails)
return ec.CheckEcVolumeReplication(writer, dataNodes, volumeIDMap, *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])
}
}
}