From 02b5b66a3f2de032f3bd13710bd1f1aeaaeab57e Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 8 Jul 2026 20:06:15 -0700 Subject: [PATCH] topology/balancer: share replica selection between shell and workers (#10276) Move pickOneReplicaToCopyFrom, pickOneReplicaToDelete, pickOneMisplacedVolume, and isMisplaced into weed/topology/balancer, following satisfyReplicaPlacement. Selection functions take the shared balancer.Replica shape and return indices, so the shell keeps its own replica type; behavior is unchanged and covered by the existing shell tests. --- weed/shell/command_volume_fix_replication.go | 168 +++++-------------- weed/topology/balancer/replica_selection.go | 161 ++++++++++++++++++ 2 files changed, 199 insertions(+), 130 deletions(-) create mode 100644 weed/topology/balancer/replica_selection.go diff --git a/weed/shell/command_volume_fix_replication.go b/weed/shell/command_volume_fix_replication.go index f2903c3aa..526256e5f 100644 --- a/weed/shell/command_volume_fix_replication.go +++ b/weed/shell/command_volume_fix_replication.go @@ -499,24 +499,11 @@ func keepDataNodesSorted(dataNodes []location, diskType types.DiskType) { } func satisfyReplicaCurrentLocation(replicaPlacement *super_block.ReplicaPlacement, replicas []*VolumeReplica) bool { - existingDataCenters, existingRacks, _ := countReplicas(replicas) - - if replicaPlacement.DiffDataCenterCount+1 > len(existingDataCenters) { - return false + locs := make([]balancer.Location, len(replicas)) + for i, r := range replicas { + locs[i] = toBalancerLocation(r.location) } - if replicaPlacement.DiffRackCount+1 > len(existingRacks) { - return false - } - if replicaPlacement.SameRackCount > 0 { - foundSatisfyRack := false - for _, rackCount := range existingRacks { - if rackCount >= replicaPlacement.SameRackCount+1 { - foundSatisfyRack = true - } - } - return foundSatisfyRack - } - return true + return balancer.SatisfyReplicaCurrentLocation(replicaPlacement, locs) } /* @@ -598,129 +585,50 @@ func (l location) DataCenter() string { return l.dc } -func pickOneReplicaToCopyFrom(replicas []*VolumeReplica) *VolumeReplica { - mostRecent := replicas[0] - for _, replica := range replicas { - if replica.info.ModifiedAtSecond > mostRecent.info.ModifiedAtSecond { - mostRecent = replica - } - } - return mostRecent -} - -func countReplicas(replicas []*VolumeReplica) (diffDc, diffRack, diffNode map[string]int) { - diffDc = make(map[string]int) - diffRack = make(map[string]int) - diffNode = make(map[string]int) - for _, replica := range replicas { - diffDc[replica.location.DataCenter()] += 1 - diffRack[replica.location.Rack()] += 1 - diffNode[replica.location.String()] += 1 - } - return -} - -// pickOneReplicaToDelete selects the replica to trim when over-replicated. -// It only ever removes the smallest of multiple healthy writable replicas: a -// ReadOnly/integrity-flagged replica is never chosen for deletion, and the -// trim is refused (returns nil) when removing a writable replica would leave -// only ReadOnly survivors. VolumeStatus file_count>0 alone cannot prove the -// survivors' .dat is readable, so we do not over-claim survivor health. -// pickSmallestReplica returns the smallest replica (ties broken by oldest then -// lowest compact revision), or nil for an empty set. -func pickSmallestReplica(replicas []*VolumeReplica) *VolumeReplica { - if len(replicas) == 0 { - return nil - } - sorted := slices.Clone(replicas) - slices.SortFunc(sorted, func(a, b *VolumeReplica) int { - if a.info.Size != b.info.Size { - return int(a.info.Size - b.info.Size) - } - if a.info.ModifiedAtSecond != b.info.ModifiedAtSecond { - return int(a.info.ModifiedAtSecond - b.info.ModifiedAtSecond) - } - if a.info.CompactRevision != b.info.CompactRevision { - return int(a.info.CompactRevision - b.info.CompactRevision) - } - return 0 - }) - return sorted[0] -} - -func pickOneReplicaToDelete(replicas []*VolumeReplica, replicaPlacement *super_block.ReplicaPlacement) *VolumeReplica { - // Over-replication trim: only ever remove a writable replica, and only - // when another writable one survives, so a healthy copy is never deleted - // down to a read-only (e.g. full or integrity-flagged) survivor. - var writable []*VolumeReplica - for _, r := range replicas { - if !r.info.ReadOnly { - writable = append(writable, r) - } - } - if len(writable) < 2 { - return nil - } - // Prefer a writable replica whose removal still satisfies placement, so the - // trim does not strip the only replica in a required failure domain. Fall - // back to the smallest writable if none keeps placement (a later misplaced - // cycle then re-balances). - var placementSafe []*VolumeReplica +// toBalancerReplicas adapts shell replicas to the shared selection shape in +// weed/topology/balancer; selection results come back as indices into the +// same slice. +func toBalancerReplicas(replicas []*VolumeReplica) []balancer.Replica { + out := make([]balancer.Replica, len(replicas)) for i, r := range replicas { - if r.info.ReadOnly { - continue - } - if !isMisplaced(otherThan(replicas, i), replicaPlacement) { - placementSafe = append(placementSafe, r) + out[i] = balancer.Replica{Location: toBalancerLocation(r.location)} + if r.info != nil { + out[i].Size = r.info.Size + out[i].ModifiedAtSecond = r.info.ModifiedAtSecond + out[i].CompactRevision = r.info.CompactRevision + out[i].ReadOnly = r.info.ReadOnly } } - if len(placementSafe) > 0 { - return pickSmallestReplica(placementSafe) + return out +} + +func pickOneReplicaToCopyFrom(replicas []*VolumeReplica) *VolumeReplica { + if i := balancer.PickOneReplicaToCopyFrom(toBalancerReplicas(replicas)); i >= 0 { + return replicas[i] } - return pickSmallestReplica(writable) + return nil +} + +// pickOneReplicaToDelete selects the replica to trim when over-replicated; +// see balancer.PickOneReplicaToDelete for the survivor-safety rules. +// VolumeStatus file_count>0 alone cannot prove the survivors' .dat is +// readable, so we do not over-claim survivor health. +func pickOneReplicaToDelete(replicas []*VolumeReplica, replicaPlacement *super_block.ReplicaPlacement) *VolumeReplica { + if i := balancer.PickOneReplicaToDelete(toBalancerReplicas(replicas), replicaPlacement); i >= 0 { + return replicas[i] + } + return nil } // check and fix misplaced volumes func isMisplaced(replicas []*VolumeReplica, replicaPlacement *super_block.ReplicaPlacement) bool { - - for i := 0; i < len(replicas); i++ { - others := otherThan(replicas, i) - if !satisfyReplicaPlacement(replicaPlacement, others, *replicas[i].location) { - return true - } - } - - return false - + return balancer.IsMisplaced(toBalancerReplicas(replicas), replicaPlacement) } -func otherThan(replicas []*VolumeReplica, index int) (others []*VolumeReplica) { - for i := 0; i < len(replicas); i++ { - if index != i { - others = append(others, replicas[i]) - } +func pickOneMisplacedVolume(replicas []*VolumeReplica, replicaPlacement *super_block.ReplicaPlacement) *VolumeReplica { + if i := balancer.PickOneMisplacedVolume(toBalancerReplicas(replicas), replicaPlacement); i >= 0 { + return replicas[i] } - return -} - -func pickOneMisplacedVolume(replicas []*VolumeReplica, replicaPlacement *super_block.ReplicaPlacement) (toDelete *VolumeReplica) { - - // Relocation, not over-replication: pick the smallest replica to delete - // and recreate at a correct placement. Unlike the trim this must still act - // on read-only replicas (e.g. a full but misplaced volume), so it does not - // use pickOneReplicaToDelete's writable-survivor guard. - var deletionCandidates []*VolumeReplica - for i := 0; i < len(replicas); i++ { - others := otherThan(replicas, i) - if !isMisplaced(others, replicaPlacement) { - deletionCandidates = append(deletionCandidates, replicas[i]) - } - } - if toDelete = pickSmallestReplica(deletionCandidates); toDelete != nil { - return toDelete - } - - return pickSmallestReplica(replicas) - + return nil } diff --git a/weed/topology/balancer/replica_selection.go b/weed/topology/balancer/replica_selection.go new file mode 100644 index 000000000..fba195d75 --- /dev/null +++ b/weed/topology/balancer/replica_selection.go @@ -0,0 +1,161 @@ +package balancer + +import ( + "github.com/seaweedfs/seaweedfs/weed/storage/super_block" +) + +// Replica pairs a replica's placement location with the volume attributes +// replica-selection decisions need. Like Location for placement checks, it is +// the shared shape the shell (from master_pb) and workers adapt to. Selection +// functions return an index into the given slice (-1 when nothing qualifies), +// so callers can map the choice back to their own richer replica type. +type Replica struct { + Location Location + Size uint64 + ModifiedAtSecond int64 + CompactRevision uint32 + ReadOnly bool +} + +// SatisfyReplicaCurrentLocation reports whether the existing replicas already +// span the failure domains the policy requires (enough data centers, enough +// racks, and the same-rack count somewhere). A volume whose replica count is +// right but whose spread falls short still needs another replica. +func SatisfyReplicaCurrentLocation(rp *super_block.ReplicaPlacement, replicas []Location) bool { + dcCounts, rackCounts, _ := countReplicas(replicas) + + if rp.DiffDataCenterCount+1 > len(dcCounts) { + return false + } + if rp.DiffRackCount+1 > len(rackCounts) { + return false + } + if rp.SameRackCount > 0 { + for _, rackCount := range rackCounts { + if rackCount >= rp.SameRackCount+1 { + return true + } + } + return false + } + return true +} + +// PickOneReplicaToCopyFrom returns the most recently modified replica, the +// best source for adding a new replica. +func PickOneReplicaToCopyFrom(replicas []Replica) int { + mostRecent := -1 + for i, r := range replicas { + if mostRecent < 0 || r.ModifiedAtSecond > replicas[mostRecent].ModifiedAtSecond { + mostRecent = i + } + } + return mostRecent +} + +// IsMisplaced reports whether any replica sits at a location inconsistent +// with the replication policy given the other replicas' locations. +func IsMisplaced(replicas []Replica, rp *super_block.ReplicaPlacement) bool { + for i := range replicas { + others := otherThan(replicas, i) + locs := make([]Location, len(others)) + for j, r := range others { + locs[j] = r.Location + } + if !SatisfyReplicaPlacement(rp, locs, replicas[i].Location) { + return true + } + } + return false +} + +// PickSmallestReplica returns the smallest replica (ties broken by oldest +// then lowest compact revision). +func PickSmallestReplica(replicas []Replica) int { + smallest := -1 + for i := range replicas { + if smallest < 0 || lessReplica(replicas[i], replicas[smallest]) { + smallest = i + } + } + return smallest +} + +// PickOneReplicaToDelete selects the replica to trim when over-replicated. +// It only ever removes the smallest of multiple healthy writable replicas: a +// ReadOnly/integrity-flagged replica is never chosen for deletion, and the +// trim is refused (returns -1) when removing a writable replica would leave +// only ReadOnly survivors. Among the writable replicas it prefers one whose +// removal still satisfies placement, so the trim does not strip the only +// replica in a required failure domain; it falls back to the smallest +// writable if none keeps placement (a later misplaced cycle then +// re-balances). +func PickOneReplicaToDelete(replicas []Replica, rp *super_block.ReplicaPlacement) int { + var writable []int + for i, r := range replicas { + if !r.ReadOnly { + writable = append(writable, i) + } + } + if len(writable) < 2 { + return -1 + } + var placementSafe []int + for _, i := range writable { + if !IsMisplaced(otherThan(replicas, i), rp) { + placementSafe = append(placementSafe, i) + } + } + if len(placementSafe) > 0 { + return pickSmallestAmong(replicas, placementSafe) + } + return pickSmallestAmong(replicas, writable) +} + +// PickOneMisplacedVolume selects the replica to delete and recreate at a +// correct placement. This is relocation, not over-replication: unlike +// PickOneReplicaToDelete it must still act on read-only replicas (e.g. a full +// but misplaced volume), so it does not use the writable-survivor guard. It +// prefers a replica whose removal leaves the others well placed, and falls +// back to the smallest replica. +func PickOneMisplacedVolume(replicas []Replica, rp *super_block.ReplicaPlacement) int { + var deletionCandidates []int + for i := range replicas { + if !IsMisplaced(otherThan(replicas, i), rp) { + deletionCandidates = append(deletionCandidates, i) + } + } + if smallest := pickSmallestAmong(replicas, deletionCandidates); smallest >= 0 { + return smallest + } + return PickSmallestReplica(replicas) +} + +func pickSmallestAmong(replicas []Replica, candidates []int) int { + smallest := -1 + for _, i := range candidates { + if smallest < 0 || lessReplica(replicas[i], replicas[smallest]) { + smallest = i + } + } + return smallest +} + +func lessReplica(a, b Replica) bool { + if a.Size != b.Size { + return a.Size < b.Size + } + if a.ModifiedAtSecond != b.ModifiedAtSecond { + return a.ModifiedAtSecond < b.ModifiedAtSecond + } + return a.CompactRevision < b.CompactRevision +} + +func otherThan(replicas []Replica, index int) (others []Replica) { + for i := range replicas { + if index != i { + others = append(others, replicas[i]) + } + } + return +}