feat: Free Volume Deadband (M) for EC shard placement (#12029)

* feat: Free Volume Deadband (M) for EC Place nodes and racks

|ΔfreeSlots| ≤ M is a tie; remainder is liveShards. M=0 stays strict
freeSlots. M is free volume slots (× dataShards). pickNodeInRackEligible
and pickRackEligible share the rule. Worker plugin config
free_slots_deadband, default 0. Not on the write path.

* fix: anchor EC freeSlots deadband on class maxFree

Pairwise |Δfree|≤M vs the current winner is not transitive: 120→110→100
at M=10 slots picked 100. After evenness, rank by maxFree, keep
|free-maxFree|≤M, then min liveShards. Shared by node and rack. M=0 is
still only maxFree then live.

* worker: add free_slots_deadband to ErasureCodingTaskConfig

* erasure_coding: persist free_slots_deadband through TaskPolicy

ApplyMaintenanceConfigFromToml accepted the key but the admin save path
round-trips config through ToTaskPolicy, so the value was dropped on
reload; serialize it like the other EC settings.

* erasure_coding: bound free_slots_deadband at its declared maximum

The TOML loader and the plugin worker config only clamped negatives, so
a value above the spec's 100000 could reach the int32 TaskPolicy field
and wrap; cap at 100000 on both entry paths and declare MaxValue on the
plugin field.

* erasure_coding: document free_slots_deadband in scaffold admin.toml

---------

Co-authored-by: Chepurnov Simon <195125934+CHEPURNOV-S-S@users.noreply.github.com>
Co-authored-by: user.email <chrislusf@users.noreply.github.com>
This commit is contained in:
authored and GitHub committed 2026-10-24 07:06:46 -07:00
1 parent 52e05ccda4
commit 18f9d9a0b6
12 files changed
+551 -72

No files matched your search

+16 -6
View File
@@ -102,6 +102,15 @@ func (cp *ConfigPersistence) ApplyMaintenanceConfigFromToml(v TomlConfig) error
ecConf.StrictPlacement = v.GetBool(k)
ecChanged = true
}
if k := "maintenance.erasure_coding.free_slots_deadband"; v.IsSet(k) {
ecConf.FreeSlotsDeadband = v.GetInt(k)
if ecConf.FreeSlotsDeadband < 0 {
ecConf.FreeSlotsDeadband = 0
} else if ecConf.FreeSlotsDeadband > 100000 {
ecConf.FreeSlotsDeadband = 100000
}
ecChanged = true
}
if !maintenanceChanged && !vacuumChanged && !balanceChanged && !ecChanged {
return nil
@@ -223,12 +232,13 @@ var pluginConfigSections = []pluginConfigSection{
jobType: "erasure_coding",
prefix: "maintenance.erasure_coding",
workerKeys: map[string]func(v TomlConfig, key string) *plugin_pb.ConfigValue{
"fullness_ratio": doubleValue,
"quiet_for_seconds": int64Value,
"min_size_mb": int64Value,
"preferred_tags": stringListValue,
"replica_placement": stringValue,
"strict_placement": boolValue,
"fullness_ratio": doubleValue,
"quiet_for_seconds": int64Value,
"min_size_mb": int64Value,
"preferred_tags": stringListValue,
"replica_placement": stringValue,
"strict_placement": boolValue,
"free_slots_deadband": int64Value,
},
// workers read collection_filter from the admin values, not the worker values
adminKeys: map[string]func(v TomlConfig, key string) *plugin_pb.ConfigValue{
+3
View File
@@ -66,6 +66,9 @@
# replica_placement = ""
# fail planning when the placement constraints cannot be met instead of relaxing them
# strict_placement = false
# volume-slot deadband for shard placement: nodes whose free volume slots differ by at
# most this much tie and fewer live shards wins; applied times data shards (0 = strict)
# free_slots_deadband = 0
# max retry attempts for a failed erasure coding job
# retry_limit = 1
# seconds to wait between retry attempts
+1
View File
@@ -381,6 +381,7 @@ message ErasureCodingTaskConfig {
repeated string preferred_tags = 5; // Disk tags to prioritize for EC shard placement
string replica_placement = 6; // EC shard replica placement (e.g. "020"); empty falls back to master default replication
bool strict_placement = 7; // fail planning instead of relaxing placement constraints
int32 free_slots_deadband = 8; // Place free-volume-slot deadband; |ΔfreeSlots| ≤ M×dataShards ranks by liveShards; 0 = strict
}
// BalanceTaskConfig contains balance-specific configuration
+21 -12
View File
@@ -2973,16 +2973,17 @@ func (x *VacuumTaskConfig) GetMinIntervalSeconds() int32 {
// ErasureCodingTaskConfig contains EC-specific configuration
type ErasureCodingTaskConfig struct {
state protoimpl.MessageState `protogen:"open.v1"`
FullnessRatio float64 `protobuf:"fixed64,1,opt,name=fullness_ratio,json=fullnessRatio,proto3" json:"fullness_ratio,omitempty"` // Minimum fullness ratio to trigger EC (0.0-1.0)
QuietForSeconds int32 `protobuf:"varint,2,opt,name=quiet_for_seconds,json=quietForSeconds,proto3" json:"quiet_for_seconds,omitempty"` // Minimum quiet time before EC
MinVolumeSizeMb int32 `protobuf:"varint,3,opt,name=min_volume_size_mb,json=minVolumeSizeMb,proto3" json:"min_volume_size_mb,omitempty"` // Minimum volume size for EC
CollectionFilter string `protobuf:"bytes,4,opt,name=collection_filter,json=collectionFilter,proto3" json:"collection_filter,omitempty"` // Only process volumes from specific collections
PreferredTags []string `protobuf:"bytes,5,rep,name=preferred_tags,json=preferredTags,proto3" json:"preferred_tags,omitempty"` // Disk tags to prioritize for EC shard placement
ReplicaPlacement string `protobuf:"bytes,6,opt,name=replica_placement,json=replicaPlacement,proto3" json:"replica_placement,omitempty"` // EC shard replica placement (e.g. "020"); empty falls back to master default replication
StrictPlacement bool `protobuf:"varint,7,opt,name=strict_placement,json=strictPlacement,proto3" json:"strict_placement,omitempty"` // fail planning instead of relaxing placement constraints
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
state protoimpl.MessageState `protogen:"open.v1"`
FullnessRatio float64 `protobuf:"fixed64,1,opt,name=fullness_ratio,json=fullnessRatio,proto3" json:"fullness_ratio,omitempty"` // Minimum fullness ratio to trigger EC (0.0-1.0)
QuietForSeconds int32 `protobuf:"varint,2,opt,name=quiet_for_seconds,json=quietForSeconds,proto3" json:"quiet_for_seconds,omitempty"` // Minimum quiet time before EC
MinVolumeSizeMb int32 `protobuf:"varint,3,opt,name=min_volume_size_mb,json=minVolumeSizeMb,proto3" json:"min_volume_size_mb,omitempty"` // Minimum volume size for EC
CollectionFilter string `protobuf:"bytes,4,opt,name=collection_filter,json=collectionFilter,proto3" json:"collection_filter,omitempty"` // Only process volumes from specific collections
PreferredTags []string `protobuf:"bytes,5,rep,name=preferred_tags,json=preferredTags,proto3" json:"preferred_tags,omitempty"` // Disk tags to prioritize for EC shard placement
ReplicaPlacement string `protobuf:"bytes,6,opt,name=replica_placement,json=replicaPlacement,proto3" json:"replica_placement,omitempty"` // EC shard replica placement (e.g. "020"); empty falls back to master default replication
StrictPlacement bool `protobuf:"varint,7,opt,name=strict_placement,json=strictPlacement,proto3" json:"strict_placement,omitempty"` // fail planning instead of relaxing placement constraints
FreeSlotsDeadband int32 `protobuf:"varint,8,opt,name=free_slots_deadband,json=freeSlotsDeadband,proto3" json:"free_slots_deadband,omitempty"` // Place free-volume-slot deadband; |ΔfreeSlots| ≤ M×dataShards ranks by liveShards; 0 = strict
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *ErasureCodingTaskConfig) Reset() {
@@ -3064,6 +3065,13 @@ func (x *ErasureCodingTaskConfig) GetStrictPlacement() bool {
return false
}
func (x *ErasureCodingTaskConfig) GetFreeSlotsDeadband() int32 {
if x != nil {
return x.FreeSlotsDeadband
}
return 0
}
// BalanceTaskConfig contains balance-specific configuration
type BalanceTaskConfig struct {
state protoimpl.MessageState `protogen:"open.v1"`
@@ -4293,7 +4301,7 @@ const file_worker_proto_rawDesc = "" +
"\x10VacuumTaskConfig\x12+\n" +
"\x11garbage_threshold\x18\x01 \x01(\x01R\x10garbageThreshold\x12/\n" +
"\x14min_volume_age_hours\x18\x02 \x01(\x05R\x11minVolumeAgeHours\x120\n" +
"\x14min_interval_seconds\x18\x03 \x01(\x05R\x12minIntervalSeconds\"\xc5\x02\n" +
"\x14min_interval_seconds\x18\x03 \x01(\x05R\x12minIntervalSeconds\"\xf5\x02\n" +
"\x17ErasureCodingTaskConfig\x12%\n" +
"\x0efullness_ratio\x18\x01 \x01(\x01R\rfullnessRatio\x12*\n" +
"\x11quiet_for_seconds\x18\x02 \x01(\x05R\x0fquietForSeconds\x12+\n" +
@@ -4301,7 +4309,8 @@ const file_worker_proto_rawDesc = "" +
"\x11collection_filter\x18\x04 \x01(\tR\x10collectionFilter\x12%\n" +
"\x0epreferred_tags\x18\x05 \x03(\tR\rpreferredTags\x12+\n" +
"\x11replica_placement\x18\x06 \x01(\tR\x10replicaPlacement\x12)\n" +
"\x10strict_placement\x18\a \x01(\bR\x0fstrictPlacement\"\x9b\x01\n" +
"\x10strict_placement\x18\a \x01(\bR\x0fstrictPlacement\x12.\n" +
"\x13free_slots_deadband\x18\b \x01(\x05R\x11freeSlotsDeadband\"\x9b\x01\n" +
"\x11BalanceTaskConfig\x12/\n" +
"\x13imbalance_threshold\x18\x01 \x01(\x01R\x12imbalanceThreshold\x12(\n" +
"\x10min_server_count\x18\x02 \x01(\x05R\x0eminServerCount\x12+\n" +
@@ -1284,7 +1284,8 @@ func volumeShardCount(node *Node, vk volKey) int {
}
// nodeLiveShardCount is the disk-level EC shard cardinality (master EcShards),
// not the UI "EC volumes" count (len of the volume map).
// not the UI "EC volumes" count (len of the volume map). Place uses it as the
// remainder after even-this-volume and freeSlots (with deadband).
func nodeLiveShardCount(n *Node) int {
if n == nil {
return 0
+145 -33
View File
@@ -2,6 +2,7 @@ package ecbalancer
import (
"fmt"
"math"
"sort"
"strings"
@@ -28,6 +29,11 @@ type Constraints struct {
PreferredTags []string
ReplicaPlacement *super_block.ReplicaPlacement
Ratio func(collection string) (dataShards, parityShards int)
// FreeSlotsDeadband (M) is free volume slots. After even-this-volume,
// Place ranks that evenness class by maxFree, keeps |free−maxFree| ≤
// M×dataShards, then lower liveShards. 0 is only maxFree then live
// (old strict freeSlots; liveShards only on an exact free tie).
FreeSlotsDeadband int
}
// DiskTypePolicy controls how Constraints.DiskType constrains placement.
@@ -193,7 +199,7 @@ func (t *Topology) Place(vid uint32, collection string, need []int, c Constraint
eligible := func(d *disk) bool {
return typeEligible(d) && (len(tt) == 0 || diskHasAnyTag(d, tt))
}
res, err := t.tryPlace(vk, need, dataShards, parityShards, racks, rackKeys, mode, c.ReplicaPlacement, eligible, prefer)
res, err := t.tryPlace(vk, need, dataShards, parityShards, racks, rackKeys, mode, c.ReplicaPlacement, eligible, prefer, volumeDeadbandToShardSlots(c.FreeSlotsDeadband, dataShards))
if err != nil {
lastErr = err
continue
@@ -223,7 +229,7 @@ func tagTiers(preferredTags []string) [][]string {
// `eligible`, with `prefer` (may be nil) ranking soft-preferred disks first. It
// journals reservations and rolls them all back if any shard cannot be placed, so
// a failed tier leaves the snapshot unchanged for the next attempt.
func (t *Topology) tryPlace(vk volKey, need []int, dataShards, parityShards int, racks map[string]*rack, rackKeys []string, mode PlacementMode, rp *super_block.ReplicaPlacement, eligible func(*disk) bool, prefer func(*disk) bool) (*PlaceResult, error) {
func (t *Topology) tryPlace(vk volKey, need []int, dataShards, parityShards int, racks map[string]*rack, rackKeys []string, mode PlacementMode, rp *super_block.ReplicaPlacement, eligible func(*disk) bool, prefer func(*disk) bool, freeSlotsDeadband int) (*PlaceResult, error) {
result := &PlaceResult{Destinations: make(map[int]Destination, len(need))}
// Per-type shard ids per rack (even caps), total shard count per rack
@@ -297,7 +303,7 @@ func (t *Topology) tryPlace(vk volKey, need []int, dataShards, parityShards int,
typeTotal = parityShards
}
for _, rl := range attempts {
node, diskID, spilled, ok := chooseShardDest(vk, sid, isData, dataShards, typeTotal, numEligibleRacks, parityShards, rackCap(rl), racks, rackKeys, rp, eligible, prefer, shardsPerRack[isData], rackShardCount, bearing, rl)
node, diskID, spilled, ok := chooseShardDest(vk, sid, isData, dataShards, typeTotal, numEligibleRacks, parityShards, rackCap(rl), racks, rackKeys, rp, eligible, prefer, shardsPerRack[isData], rackShardCount, bearing, rl, freeSlotsDeadband)
if !ok {
continue
}
@@ -350,12 +356,13 @@ func (t *Topology) tryPlace(vk volKey, need []int, dataShards, parityShards int,
// chooseShardDest selects a (node, disk) for one shard at the given relaxation
// level: pick a rack (total-shards-per-rack cap + even per-type cap +
// ReplicaPlacement caps + two-pass anti-affinity to the opposite type), then the
// least-loaded eligible node, then the best eligible disk. If no node in the
// chosen rack fits (e.g. every node is at SameRackCount), the next-best rack is
// tried. The third return reports whether the disk spilled off the
// soft-preferred type. ok=false when no rack/node/disk fits.
func chooseShardDest(vk volKey, sid int, isData bool, dataShards, typeTotal, numEligibleRacks, maxPerDisk, maxTotalPerRack int, racks map[string]*rack, rackKeys []string, rp *super_block.ReplicaPlacement, eligible func(*disk) bool, prefer func(*disk) bool, shardsPerRackType map[string][]int, rackShardCount map[string]int, bearing map[bool]map[string]bool, rl relaxation) (*Node, uint32, bool, bool) {
// ReplicaPlacement caps + two-pass anti-affinity to the opposite type, then
// freeSlots deadband / liveShards), then the least-loaded eligible node, then
// the best eligible disk. If no node in the chosen rack fits (e.g. every node
// is at SameRackCount), the next-best rack is tried. The third return reports
// whether the disk spilled off the soft-preferred type. ok=false when no
// rack/node/disk fits.
func chooseShardDest(vk volKey, sid int, isData bool, dataShards, typeTotal, numEligibleRacks, maxPerDisk, maxTotalPerRack int, racks map[string]*rack, rackKeys []string, rp *super_block.ReplicaPlacement, eligible func(*disk) bool, prefer func(*disk) bool, shardsPerRackType map[string][]int, rackShardCount map[string]int, bearing map[bool]map[string]bool, rl relaxation, freeSlotsDeadband int) (*Node, uint32, bool, bool) {
maxPerRack := numEligibleRacks*typeTotal + 1 // effectively unlimited when caps are relaxed
if rl.caps {
if maxPerRack = ceilDivide(typeTotal, numEligibleRacks); maxPerRack < 1 {
@@ -389,12 +396,11 @@ func chooseShardDest(vk volKey, sid int, isData bool, dataShards, typeTotal, num
return !tried[r] && racks[r].freeSlots > 0 && rackShardRoom(racks[r], vk, eligible, maxPerDisk) > 0
}
for {
destRack, ok := pickTarget(rackKeys, shardsPerRackType, maxPerRack, anti, hasRoom, withinLimit,
func(r string) int { return racks[r].freeSlots })
destRack, ok := pickRackEligible(rackKeys, shardsPerRackType, maxPerRack, anti, hasRoom, withinLimit, racks, freeSlotsDeadband)
if !ok {
return nil, 0, false, false
}
if node := pickNodeInRackEligible(racks[destRack], vk, rp, eligible, maxPerDisk); node != nil {
if node := pickNodeInRackEligible(racks[destRack], vk, rp, eligible, maxPerDisk, freeSlotsDeadband); node != nil {
if diskID, ok, spilled := pickBestDiskEligible(node, vk, eligible, prefer, sid, dataShards, maxPerDisk); ok {
return node, diskID, spilled, true
}
@@ -482,18 +488,83 @@ func rackShardRoomUnderRP(r *rack, vk volKey, eligible func(*disk) bool, maxPerD
return room
}
// pickRackEligible is pickTarget for Place racks. Even this-type / cap /
// two-pass anti-affinity stay. Among that evenness class the remainder is
// max rack freeSlots, keep |free−maxFree| ≤ shard-slot deadband, then
// lower liveShards (sum of nodeLiveShardCount). M=0 is only maxFree then
// live (old strict freeSlots). pickTarget itself is unchanged so balancer
// / DC paths stay without M.
func pickRackEligible(candidates []string, shardsPerTarget map[string][]int, maxPerTarget int, antiAffinity map[string]bool, hasFreeSlots, withinLimit func(string) bool, racks map[string]*rack, freeSlotsDeadband int) (string, bool) {
try := func(skipAnti bool) (string, bool) {
var keys []string
var free, live []int
bestCount := maxPerTarget + 1
for _, c := range candidates {
if skipAnti && antiAffinity[c] {
continue
}
if !hasFreeSlots(c) {
continue
}
if len(shardsPerTarget[c]) >= maxPerTarget {
continue
}
if !withinLimit(c) {
continue
}
cnt := len(shardsPerTarget[c])
if len(keys) == 0 || cnt < bestCount {
keys, free, live = keys[:0], free[:0], live[:0]
bestCount = cnt
} else if cnt > bestCount {
continue
}
f := 0
if r := racks[c]; r != nil {
f = r.freeSlots
}
keys = append(keys, c)
free = append(free, f)
live = append(live, rackLiveShardCount(racks[c]))
}
if len(keys) == 0 {
return "", false
}
return keys[pickDeadbandRemainder(free, live, freeSlotsDeadband)], true
}
if len(antiAffinity) > 0 {
if t, ok := try(true); ok {
return t, true
}
}
return try(false)
}
func rackLiveShardCount(r *rack) int {
if r == nil {
return 0
}
total := 0
for _, n := range r.nodes {
total += nodeLiveShardCount(n)
}
return total
}
// pickNodeInRackEligible is pickNodeInRack restricted to nodes with an eligible
// disk that can take another shard of the volume (free slot, under maxPerDisk).
// FromActiveTopology keeps all disk types/tags in the snapshot, so without this a
// node with free volume slots but no eligible disk could be chosen.
//
// Even first (this volume, machine then node). Machine freeSlots remainder
// (mFree) was dropped: keep host even (mCount); remainder is disk freeSlots
// then liveShards so thin disks fill first. SameRackCount / 2+2 unchanged.
func pickNodeInRackEligible(r *rack, vk volKey, rp *super_block.ReplicaPlacement, eligible func(*disk) bool, maxPerDisk int) *Node {
// Even first (this volume, machine then node). Among that class: max node
// freeSlots, keep |free−maxFree| ≤ shard-slot deadband, then lower
// liveShards. Machine freeSlots (mFree) stays dropped as on origin/master.
// SameRackCount / 2+2 unchanged. M=0 is only maxFree then live. Callers
// convert UI volume M via volumeDeadbandToShardSlots.
func pickNodeInRackEligible(r *rack, vk volKey, rp *super_block.ReplicaPlacement, eligible func(*disk) bool, maxPerDisk, freeSlotsDeadband int) *Node {
machineShards := countShardsByHost(vk, r.nodes)
var best *Node
var bestMCount, bestNCount, bestNFree, bestNScore int
var nodes []*Node
var bestMCount, bestNCount int
for _, id := range sortedNodeKeys(r.nodes) {
node := r.nodes[id]
if nodeShardRoom(node, vk, eligible, maxPerDisk) <= 0 {
@@ -504,27 +575,68 @@ func pickNodeInRackEligible(r *rack, vk volKey, rp *super_block.ReplicaPlacement
continue
}
mCount := machineShards[node.host]
nScore := nodeLiveShardCount(node)
better := false
switch {
case best == nil:
better = true
case mCount != bestMCount:
better = mCount < bestMCount
case count != bestNCount:
better = count < bestNCount
case node.freeSlots != bestNFree:
better = node.freeSlots > bestNFree
default:
better = nScore < bestNScore
if len(nodes) == 0 || mCount < bestMCount || (mCount == bestMCount && count < bestNCount) {
nodes = nodes[:0]
bestMCount, bestNCount = mCount, count
} else if mCount != bestMCount || count != bestNCount {
continue
}
if better {
best, bestMCount, bestNCount, bestNFree, bestNScore = node, mCount, count, node.freeSlots, nScore
nodes = append(nodes, node)
}
if len(nodes) == 0 {
return nil
}
free := make([]int, len(nodes))
live := make([]int, len(nodes))
for i, n := range nodes {
free[i] = n.freeSlots
live[i] = nodeLiveShardCount(n)
}
return nodes[pickDeadbandRemainder(free, live, freeSlotsDeadband)]
}
// pickDeadbandRemainder ranks same-evenness candidates: maxFree of the
// class, keep |free−maxFree|≤m, then min live. Ties keep the earlier
// index (scan order). M=0 keeps only maxFree then live. Shared by node
// and rack so the band cannot drift.
func pickDeadbandRemainder(free, live []int, m int) int {
maxFree := free[0]
for _, f := range free[1:] {
if f > maxFree {
maxFree = f
}
}
best := -1
for i, f := range free {
if absIntDiff(f, maxFree) > m {
continue
}
if best < 0 || live[i] < live[best] {
best = i
}
}
return best
}
func absIntDiff(a, b int) int {
if a > b {
return a - b
}
return b - a
}
// volumeDeadbandToShardSlots converts UI volume-slot M to the shard-slot
// deadband Place compares against node.freeSlots. OSS 10+4 uses dataShards=10.
func volumeDeadbandToShardSlots(volumes, dataShards int) int {
if volumes <= 0 || dataShards <= 0 {
return 0
}
if volumes > math.MaxInt/dataShards {
return math.MaxInt
}
return volumes * dataShards
}
// pickBestDiskEligible chooses the best eligible disk on a node, ranking
// soft-preferred disks (prefer != nil && prefer(d)) ahead of others so disk-type
// Prefer uses the preferred type when available but spills otherwise. Returns the
@@ -2,6 +2,7 @@ package ecbalancer
import (
"fmt"
"math"
"sort"
"testing"
@@ -914,3 +915,235 @@ func TestPlace10Plus4ImbalanceTiebreak(t *testing.T) {
})
}
}
func TestVolumeDeadbandToShardSlots(t *testing.T) {
if got := volumeDeadbandToShardSlots(0, 10); got != 0 {
t.Fatalf("M=0 → %d, want 0", got)
}
if got := volumeDeadbandToShardSlots(1, 10); got != 10 {
t.Fatalf("M=1 at 10+4 → %d, want 10", got)
}
if got := volumeDeadbandToShardSlots(1, 8); got != 8 {
t.Fatalf("M=1 at 8+4 → %d, want 8", got)
}
if got := volumeDeadbandToShardSlots(50, 10); got != 500 {
t.Fatalf("M=50 at 10+4 → %d, want 500", got)
}
if got := volumeDeadbandToShardSlots(-3, 10); got != 0 {
t.Fatalf("negative M → %d, want 0", got)
}
}
func TestAbsIntDiff(t *testing.T) {
if got := absIntDiff(105, 100); got != 5 {
t.Fatalf("absIntDiff(105, 100)=%d, want 5", got)
}
if got := absIntDiff(100, 105); got != 5 {
t.Fatalf("absIntDiff(100, 105)=%d, want 5", got)
}
if got := absIntDiff(100, 100); got != 0 {
t.Fatalf("absIntDiff(100, 100)=%d, want 0", got)
}
if got := absIntDiff(math.MaxInt, math.MaxInt-5); got != 5 {
t.Fatalf("absIntDiff(MaxInt, MaxInt-5)=%d, want 5", got)
}
if got := absIntDiff(math.MaxInt-5, math.MaxInt); got != 5 {
t.Fatalf("absIntDiff(MaxInt-5, MaxInt)=%d, want 5", got)
}
if got := absIntDiff(math.MinInt+5, math.MinInt); got != 5 {
t.Fatalf("absIntDiff(MinInt+5, MinInt)=%d, want 5", got)
}
if got := absIntDiff(math.MinInt, math.MinInt+5); got != 5 {
t.Fatalf("absIntDiff(MinInt, MinInt+5)=%d, want 5", got)
}
}
func placeFirstShardNode(t *testing.T, topo *Topology, m int) string {
t.Helper()
res, err := topo.Place(1, "c1", []int{0}, Constraints{FreeSlotsDeadband: m}, PlaceStrict)
if err != nil {
t.Fatalf("Place M=%d: %v", m, err)
}
d, ok := res.Destinations[0]
if !ok {
t.Fatal("missing shard 0")
}
return d.Node
}
// TestFreeSlotsDeadbandDefersToLiveShards: two nodes, empty this-volume.
// b has slightly more freeSlots but more live shards. M=0 follows freeSlots
// (b); M=1 volume (×10 shard slots at 10+4) covers the gap and follows
// liveShards (a). Sorted-id would always pick a.
func TestFreeSlotsDeadbandDefersToLiveShards(t *testing.T) {
build := func() *Topology {
topo := NewTopology()
a := topo.AddNode("a:8080", "dc1", "dc1:rack0", 100)
a.AddDisk(0, "", 100, 10)
b := topo.AddNode("b:8081", "dc1", "dc1:rack0", 105)
b.AddDisk(0, "", 105, 50)
return topo
}
if got := placeFirstShardNode(t, build(), 0); got != "b:8081" {
t.Fatalf("M=0: got %s, want b (strict freeSlots)", got)
}
if got := placeFirstShardNode(t, build(), 1); got != "a:8080" {
t.Fatalf("M=1 volume: got %s, want a (liveShards remainder)", got)
}
}
// TestFreeSlotsDeadbandZeroExactTieUsesLiveShards: equal freeSlots, M=0
// remainder is liveShards, not the lower sorted id.
func TestFreeSlotsDeadbandZeroExactTieUsesLiveShards(t *testing.T) {
topo := NewTopology()
a := topo.AddNode("a:8080", "dc1", "dc1:rack0", 100)
a.AddDisk(0, "", 100, 50)
b := topo.AddNode("b:8081", "dc1", "dc1:rack0", 100)
b.AddDisk(0, "", 100, 10)
if got := placeFirstShardNode(t, topo, 0); got != "b:8081" {
t.Fatalf("equal freeSlots M=0: got %s, want b (lower liveShards)", got)
}
}
// TestFreeSlotsDeadbandStillHonorsLargeGap: gap above M still follows freeSlots
// even if that node is fatter on liveShards.
func TestFreeSlotsDeadbandStillHonorsLargeGap(t *testing.T) {
topo := NewTopology()
a := topo.AddNode("a:8080", "dc1", "dc1:rack0", 100)
a.AddDisk(0, "", 100, 10)
b := topo.AddNode("b:8081", "dc1", "dc1:rack0", 200)
b.AddDisk(0, "", 200, 80)
if got := placeFirstShardNode(t, topo, 1); got != "b:8081" {
t.Fatalf("M=1 volume gap=100 shards: got %s, want b (freeSlots still wins)", got)
}
}
// TestFreeSlotsDeadbandScalesWithRatio: M=1 volume is 8 shard slots at 8+4
// and 10 at default 10+4. Gap of 9 shards: 8+4 still follows freeSlots; 10+4
// ties and follows liveShards.
func TestFreeSlotsDeadbandScalesWithRatio(t *testing.T) {
build := func() *Topology {
topo := NewTopology()
a := topo.AddNode("a:8080", "dc1", "dc1:rack0", 100)
a.AddDisk(0, "", 100, 10)
b := topo.AddNode("b:8081", "dc1", "dc1:rack0", 109)
b.AddDisk(0, "", 109, 50)
return topo
}
place := func(topo *Topology, ratio func(string) (int, int)) string {
t.Helper()
res, err := topo.Place(1, "c1", []int{0}, Constraints{FreeSlotsDeadband: 1, Ratio: ratio}, PlaceStrict)
if err != nil {
t.Fatalf("Place: %v", err)
}
d, ok := res.Destinations[0]
if !ok {
t.Fatal("missing shard 0")
}
return d.Node
}
if got := place(build(), func(string) (int, int) { return 8, 4 }); got != "b:8081" {
t.Fatalf("8+4 M=1 gap=9: got %s, want b (deadband 8, freeSlots)", got)
}
if got := place(build(), nil); got != "a:8080" {
t.Fatalf("10+4 M=1 gap=9: got %s, want a (deadband 10, liveShards)", got)
}
}
func rackExtraTopo(thinFree, thinLive, fatFree, fatLive int) *Topology {
topo := NewTopology()
add := func(rack, id string, free, live int) {
n := topo.AddNode(id, "dc1", rack, free)
n.AddDisk(0, "", free, live)
}
add("dc1:rack0", "thin:8080", thinFree, thinLive)
add("dc1:rack1", "fat1:8080", fatFree, fatLive)
add("dc1:rack2", "fat2:8080", fatFree, fatLive)
add("dc1:rack3", "fat3:8080", fatFree, fatLive)
return topo
}
func place104PerRack(t *testing.T, topo *Topology, m int) map[string]int {
t.Helper()
res, err := topo.Place(1, "c1", allShards(), Constraints{FreeSlotsDeadband: m}, PlaceStrict)
if err != nil {
t.Fatalf("Place M=%d: %v", m, err)
}
got := map[string]int{}
for _, d := range res.Destinations {
got[d.Rack]++
}
return got
}
// TestRackDeadbandDefersToLiveShards: 10+4 extras are 4+4+3+3. Thin rack has
// fewer freeSlots (gap 5) but fewer liveShards. M=1 volume is 10 shard slots,
// so the gap is a tie and extras go to the thinner live rack.
func TestRackDeadbandDefersToLiveShards(t *testing.T) {
got := place104PerRack(t, rackExtraTopo(100, 10, 105, 50), 1)
if got["rack0"] != 4 {
t.Fatalf("M=1 gap=5: thin rack0 got %d, want 4 (liveShards remainder); per-rack %v", got["rack0"], got)
}
}
// TestRackDeadbandStillHonorsLargeGap: same thin-vs-fat live, but freeSlots
// gap 100 is above M=1 (10 shard slots), so extras still follow freeSlots.
func TestRackDeadbandStillHonorsLargeGap(t *testing.T) {
got := place104PerRack(t, rackExtraTopo(100, 10, 200, 50), 1)
if got["rack0"] != 3 {
t.Fatalf("M=1 gap=100: thin rack0 got %d, want 3 (freeSlots still wins); per-rack %v", got["rack0"], got)
}
}
// Greptile chain at M=1 volume (10 shard slots): 120/300, 110/200, 100/100.
// Pairwise vs the current winner walks 120→110→100 or 100→110→120 depending
// on scan order. maxFree=120 keeps |Δ|≤10, so 110 wins — not 100 (out of
// band) and not 120 (fatter live).
func TestNodeDeadbandAnchorsOnMaxFree(t *testing.T) {
run := func(t *testing.T, frees, lives [3]int, want string) {
t.Helper()
topo := NewTopology()
ids := [3]string{"n0:8080", "n1:8081", "n2:8082"}
for i, id := range ids {
n := topo.AddNode(id, "dc1", "dc1:rack0", frees[i])
n.AddDisk(0, "", frees[i], lives[i])
}
if got := placeFirstShardNode(t, topo, 1); got != want {
t.Fatalf("got %s, want %s (110 in-band, not 100/120)", got, want)
}
}
t.Run("high-to-low", func(t *testing.T) {
run(t, [3]int{120, 110, 100}, [3]int{300, 200, 100}, "n1:8081")
})
t.Run("low-to-high", func(t *testing.T) {
run(t, [3]int{100, 110, 120}, [3]int{100, 200, 300}, "n1:8081")
})
}
func TestRackDeadbandAnchorsOnMaxFree(t *testing.T) {
run := func(t *testing.T, frees, lives [3]int, want string) {
t.Helper()
topo := NewTopology()
for i := 0; i < 3; i++ {
n := topo.AddNode(fmt.Sprintf("n%d:8080", i), "dc1", fmt.Sprintf("dc1:rack%d", i), frees[i])
n.AddDisk(0, "", frees[i], lives[i])
}
res, err := topo.Place(1, "c1", []int{0}, Constraints{FreeSlotsDeadband: 1}, PlaceStrict)
if err != nil {
t.Fatalf("Place: %v", err)
}
d, ok := res.Destinations[0]
if !ok {
t.Fatal("missing shard 0")
}
if d.Rack != want {
t.Fatalf("got rack %s, want %s (110 in-band, not 100/120)", d.Rack, want)
}
}
t.Run("high-to-low", func(t *testing.T) {
run(t, [3]int{120, 110, 100}, [3]int{300, 200, 100}, "rack1")
})
t.Run("low-to-high", func(t *testing.T) {
run(t, [3]int{100, 110, 120}, [3]int{100, 200, 300}, "rack1")
})
}
+33 -14
View File
@@ -12,13 +12,14 @@ import (
// Config extends BaseConfig with erasure coding specific settings
type Config struct {
base.BaseConfig
QuietForSeconds int `json:"quiet_for_seconds"`
FullnessRatio float64 `json:"fullness_ratio"`
CollectionFilter string `json:"collection_filter"`
MinSizeMB int `json:"min_size_mb"`
PreferredTags []string `json:"preferred_tags"`
ReplicaPlacement string `json:"replica_placement"` // e.g. "020"; empty falls back to the master default replication
StrictPlacement bool `json:"strict_placement"` // fail planning instead of relaxing placement constraints
QuietForSeconds int `json:"quiet_for_seconds"`
FullnessRatio float64 `json:"fullness_ratio"`
CollectionFilter string `json:"collection_filter"`
MinSizeMB int `json:"min_size_mb"`
PreferredTags []string `json:"preferred_tags"`
ReplicaPlacement string `json:"replica_placement"` // e.g. "020"; empty falls back to the master default replication
StrictPlacement bool `json:"strict_placement"` // fail planning instead of relaxing placement constraints
FreeSlotsDeadband int `json:"free_slots_deadband"` // Place volume-slot M; |ΔfreeSlots| ≤ M×dataShards is a tie, then liveShards; 0 = strict
}
// NewDefaultConfig creates a new default erasure coding configuration
@@ -184,6 +185,22 @@ func GetConfigSpec() base.ConfigSpec {
InputType: "checkbox",
CSSClasses: "form-check-input",
},
{
Name: "free_slots_deadband",
JSONName: "free_slots_deadband",
Type: config.FieldTypeInt,
DefaultValue: 0,
MinValue: 0,
MaxValue: 100000,
Required: false,
DisplayName: "Free Volume Deadband (M)",
Description: "Treat |Δfree volume slots| ≤ M as a Place tie and rank by liveShards instead",
HelpText: "0 (default) keeps strict freeSlots ranking. A positive M is volume slots; Place applies M × data shards (10+4 → ×10).",
Placeholder: "0",
Unit: config.UnitCount,
InputType: "number",
CSSClasses: "form-control",
},
},
}
}
@@ -199,13 +216,14 @@ func (c *Config) ToTaskPolicy() *worker_pb.TaskPolicy {
CheckIntervalSeconds: int32(c.ScanIntervalSeconds),
TaskConfig: &worker_pb.TaskPolicy_ErasureCodingConfig{
ErasureCodingConfig: &worker_pb.ErasureCodingTaskConfig{
FullnessRatio: float64(c.FullnessRatio),
QuietForSeconds: int32(c.QuietForSeconds),
MinVolumeSizeMb: int32(c.MinSizeMB),
CollectionFilter: c.CollectionFilter,
PreferredTags: preferredTagsCopy,
ReplicaPlacement: c.ReplicaPlacement,
StrictPlacement: c.StrictPlacement,
FullnessRatio: float64(c.FullnessRatio),
QuietForSeconds: int32(c.QuietForSeconds),
MinVolumeSizeMb: int32(c.MinSizeMB),
CollectionFilter: c.CollectionFilter,
PreferredTags: preferredTagsCopy,
ReplicaPlacement: c.ReplicaPlacement,
StrictPlacement: c.StrictPlacement,
FreeSlotsDeadband: int32(c.FreeSlotsDeadband),
},
},
}
@@ -231,6 +249,7 @@ func (c *Config) FromTaskPolicy(policy *worker_pb.TaskPolicy) error {
c.PreferredTags = append([]string(nil), ecConfig.PreferredTags...)
c.ReplicaPlacement = ecConfig.ReplicaPlacement
c.StrictPlacement = ecConfig.StrictPlacement
c.FreeSlotsDeadband = int(ecConfig.FreeSlotsDeadband)
}
return nil
@@ -519,11 +519,12 @@ func planECDestinations(snap *ecbalancer.Topology, nodeAddresses map[string]stri
mode = ecbalancer.PlaceStrict
}
res, err := snap.Place(metric.VolumeID, metric.Collection, need, ecbalancer.Constraints{
DiskType: metric.DiskType,
DiskTypePolicy: ecbalancer.DiskTypePrefer,
PreferredTags: ecConfig.PreferredTags,
ReplicaPlacement: rp,
Ratio: func(string) (int, int) { return dataShards, parityShards },
DiskType: metric.DiskType,
DiskTypePolicy: ecbalancer.DiskTypePrefer,
PreferredTags: ecConfig.PreferredTags,
ReplicaPlacement: rp,
Ratio: func(string) (int, int) { return dataShards, parityShards },
FreeSlotsDeadband: ecConfig.FreeSlotsDeadband,
}, mode)
if err != nil {
return nil, nil, err
@@ -39,12 +39,16 @@ func TestPlanECDestinationsStrictPlacement(t *testing.T) {
func TestStrictPlacementRoundTripsThroughTaskPolicy(t *testing.T) {
cfg := NewDefaultConfig()
cfg.StrictPlacement = true
cfg.FreeSlotsDeadband = 3
restored := NewDefaultConfig()
require.NoError(t, restored.FromTaskPolicy(cfg.ToTaskPolicy()))
require.True(t, restored.StrictPlacement, "strict placement must survive the persisted policy round trip")
require.Equal(t, 3, restored.FreeSlotsDeadband, "free_slots_deadband must survive the persisted policy round trip")
restored.StrictPlacement = false
restored.FreeSlotsDeadband = 0
require.NoError(t, restored.FromTaskPolicy(restored.ToTaskPolicy()))
require.False(t, restored.StrictPlacement)
require.Zero(t, restored.FreeSlotsDeadband)
}
@@ -98,7 +98,7 @@ func (h *ErasureCodingHandler) Descriptor() *plugin_pb.JobTypeDescriptor {
WorkerConfigForm: &plugin_pb.ConfigForm{
FormId: "erasure-coding-worker",
Title: "Erasure Coding Worker Config",
Description: "Worker-side detection thresholds.",
Description: "Worker-side detection and shard placement.",
Sections: []*plugin_pb.ConfigSection{
{
SectionId: "thresholds",
@@ -133,6 +133,13 @@ func (h *ErasureCodingHandler) Descriptor() *plugin_pb.JobTypeDescriptor {
Required: true,
MinValue: &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 1}},
},
},
},
{
SectionId: "placement",
Title: "Shard Placement",
Description: "Where EC shards go: disk tags, replica-placement caps, Place ranking.",
Fields: []*plugin_pb.ConfigField{
{
Name: "preferred_tags",
Label: "Preferred Tags",
@@ -149,6 +156,15 @@ func (h *ErasureCodingHandler) Descriptor() *plugin_pb.JobTypeDescriptor {
FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_STRING,
Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_TEXT,
},
{
Name: "free_slots_deadband",
Label: "Free Volume Deadband (M)",
Description: "Nodes whose free volume slots differ by at most M are a tie; then fewer live shards wins. Applied as M × data shards (10+4 → ×10). 0 = always pick more free slots.",
FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_INT64,
Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_NUMBER,
MinValue: &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
MaxValue: &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 100000}},
},
{
Name: "strict_placement",
Label: "Strict Placement",
@@ -178,6 +194,9 @@ func (h *ErasureCodingHandler) Descriptor() *plugin_pb.JobTypeDescriptor {
"strict_placement": {
Kind: &plugin_pb.ConfigValue_BoolValue{BoolValue: false},
},
"free_slots_deadband": {
Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0},
},
},
},
AdminRuntimeDefaults: &plugin_pb.AdminRuntimeDefaults{
@@ -205,6 +224,9 @@ func (h *ErasureCodingHandler) Descriptor() *plugin_pb.JobTypeDescriptor {
"preferred_tags": {
Kind: &plugin_pb.ConfigValue_StringValue{StringValue: ""},
},
"free_slots_deadband": {
Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0},
},
},
}
}
@@ -640,6 +662,14 @@ func deriveErasureCodingWorkerConfig(values map[string]*plugin_pb.ConfigValue) *
taskConfig.StrictPlacement = pluginworker.ReadBoolConfig(values, "strict_placement", taskConfig.StrictPlacement)
deadband := pluginworker.ReadIntConfig(values, "free_slots_deadband", taskConfig.FreeSlotsDeadband)
if deadband < 0 {
deadband = 0
} else if deadband > 100000 {
deadband = 100000
}
taskConfig.FreeSlotsDeadband = deadband
return &erasureCodingWorkerConfig{
TaskConfig: taskConfig,
}
@@ -128,6 +128,17 @@ func TestDeriveErasureCodingWorkerConfig(t *testing.T) {
if cfg.TaskConfig.MinSizeMB != 128 {
t.Fatalf("expected min_size_mb 128, got %d", cfg.TaskConfig.MinSizeMB)
}
if cfg.TaskConfig.FreeSlotsDeadband != 0 {
t.Fatalf("expected default free_slots_deadband 0, got %d", cfg.TaskConfig.FreeSlotsDeadband)
}
values["free_slots_deadband"] = &plugin_pb.ConfigValue{
Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 12},
}
cfg = deriveErasureCodingWorkerConfig(values)
if cfg.TaskConfig.FreeSlotsDeadband != 12 {
t.Fatalf("expected free_slots_deadband 12, got %d", cfg.TaskConfig.FreeSlotsDeadband)
}
}
func TestBuildErasureCodingProposal(t *testing.T) {
@@ -249,6 +260,51 @@ func TestEmitErasureCodingDetectionDecisionTraceNoTasks(t *testing.T) {
}
}
func TestErasureCodingDescriptorExposesFreeSlotsDeadband(t *testing.T) {
descriptor := NewErasureCodingHandler(nil, "").Descriptor()
if descriptor == nil || descriptor.WorkerConfigForm == nil {
t.Fatalf("expected worker config form in descriptor")
}
if descriptor.DescriptorVersion != 1 {
t.Fatalf("descriptor version = %d, want 1", descriptor.DescriptorVersion)
}
if !workerConfigFormHasField(descriptor.WorkerConfigForm, "free_slots_deadband") {
t.Fatal("worker config form omitted free_slots_deadband")
}
var sectionID, prev string
for _, section := range descriptor.WorkerConfigForm.Sections {
if section == nil {
continue
}
for _, field := range section.Fields {
if field != nil && field.Name == "free_slots_deadband" {
sectionID = section.SectionId
if field.Label != "Free Volume Deadband (M)" {
t.Fatalf("free_slots_deadband label = %q, want Free Volume Deadband (M)", field.Label)
}
}
if field != nil && field.Name == "replica_placement" {
prev = section.SectionId
}
}
}
if sectionID != "placement" {
t.Fatalf("free_slots_deadband section = %q, want placement", sectionID)
}
if prev != "placement" {
t.Fatalf("replica_placement section = %q, want placement next to free_slots_deadband", prev)
}
for _, section := range descriptor.WorkerConfigForm.Sections {
if section != nil && section.SectionId == "thresholds" {
for _, field := range section.Fields {
if field != nil && field.Name == "free_slots_deadband" {
t.Fatal("free_slots_deadband must not sit under Detection Thresholds")
}
}
}
}
}
func TestErasureCodingDescriptorOmitsLocalExecutionFields(t *testing.T) {
descriptor := NewErasureCodingHandler(nil, "").Descriptor()
if descriptor == nil || descriptor.WorkerConfigForm == nil {