Files
seaweedfs/weed/admin/dash/config_toml_test.go
T
fefac028ec feat: send EC shards in parallel per destination disk (#12050)
* feat: send EC shards in parallel per destination disk

One stream per destination disk. max_shard_transfers unset means no global cap, 0 sends shards sequentially, and N is a global cap that still keeps one stream per disk. The value is persisted through TaskPolicy. The maintenance form is a checkbox plus Max transfers because unset and 0 differ. The key is intentionally omitted from GetConfigSpec.

* ec shard distribution: treat negative max_shard_transfers as sequential

The config layer (normalizedMaxShardTransfers) clamps negative values to 0,
which selects sequential distribution. shardTransferCap diverged: a negative
cap that reached ShardDistributionOptions directly would have selected
uncapped parallel mode, the most aggressive setting. Align the options layer
with the same normalization so invalid input degrades to the conservative
mode.

* ec: serialize shard transfers to a server when a target leaves disk choice to the receiver

disk_id 0 is auto-placement on the receiving volume server, so an
auto-placed stream can land on a disk another queued stream is pinned
to. Grouping by (destNode, disk_id) alone let both run at once; collapse
all of that server's transfers under the destNode key instead.

* worker: clamp max_shard_transfers to the int32 wire range

ReadIntConfig returned the 0 fallback for values above MaxInt32, turning
a large positive cap into sequential mode. Read int64 and clamp into
range instead.

* worker: drop unused shard-transfer form helper

The maintenance form renders and collects max_shard_transfers in
plugin.templ JavaScript; the Go copy had no callers.

* worker: bound max_shard_transfers conversion under an in-range guard

---------

Co-authored-by: Chepurnov Simon <195125934+CHEPURNOV-S-S@users.noreply.github.com>
Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>
2026-10-25 06:37:26 -07:00

404 lines
12 KiB
Go

package dash
import (
"math"
"os"
"path/filepath"
"reflect"
"strconv"
"strings"
"testing"
adminplugin "github.com/seaweedfs/seaweedfs/weed/admin/plugin"
"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
"github.com/seaweedfs/seaweedfs/weed/worker/tasks/balance"
"github.com/seaweedfs/seaweedfs/weed/worker/tasks/erasure_coding"
"github.com/seaweedfs/seaweedfs/weed/worker/tasks/vacuum"
"github.com/spf13/viper"
)
func tomlConfig(t *testing.T, content string) *viper.Viper {
t.Helper()
v := viper.New()
v.SetConfigType("toml")
if err := v.ReadConfig(strings.NewReader(content)); err != nil {
t.Fatalf("read toml: %v", err)
}
return v
}
func TestApplyMaintenanceConfigFromToml(t *testing.T) {
dir := t.TempDir()
cp := NewConfigPersistence(dir)
// simulate an earlier UI edit
saved := vacuum.NewDefaultConfig()
saved.MaxConcurrent = 5
saved.GarbageThreshold = 0.5
if err := cp.SaveVacuumTaskPolicy(saved.ToTaskPolicy()); err != nil {
t.Fatalf("save vacuum policy: %v", err)
}
balanceBefore := balance.LoadConfigFromPersistence(cp)
v := tomlConfig(t, `
[maintenance.vacuum]
garbage_threshold = 0.03
min_volume_age_seconds = 1800
[maintenance.erasure_coding]
enabled = false
fullness_ratio = 0.8
preferred_tags = "Fast, ssd"
`)
if err := cp.ApplyMaintenanceConfigFromToml(v); err != nil {
t.Fatalf("apply: %v", err)
}
vacuumConf := vacuum.LoadConfigFromPersistence(cp)
if vacuumConf.GarbageThreshold != 0.03 {
t.Errorf("garbage threshold = %v, want 0.03", vacuumConf.GarbageThreshold)
}
// sub-hour values round up to a whole hour instead of truncating to 0
if vacuumConf.MinVolumeAgeSeconds != 3600 {
t.Errorf("min volume age = %v, want 3600", vacuumConf.MinVolumeAgeSeconds)
}
if vacuumConf.MaxConcurrent != 5 {
t.Errorf("max concurrent = %v, want UI-saved 5 preserved", vacuumConf.MaxConcurrent)
}
ecConf := erasure_coding.LoadConfigFromPersistence(cp)
if ecConf.Enabled {
t.Errorf("ec enabled = true, want false")
}
if ecConf.FullnessRatio != 0.8 {
t.Errorf("fullness ratio = %v, want 0.8", ecConf.FullnessRatio)
}
if !reflect.DeepEqual(ecConf.PreferredTags, []string{"fast", "ssd"}) {
t.Errorf("preferred tags = %v, want [fast ssd]", ecConf.PreferredTags)
}
if ecConf.QuietForSeconds != 3600 {
t.Errorf("quiet for = %v, want default 3600 preserved", ecConf.QuietForSeconds)
}
// balance section absent: nothing written
if _, err := os.Stat(filepath.Join(dir, ConfigSubdir, BalanceTaskConfigFile)); !os.IsNotExist(err) {
t.Errorf("balance config written without [maintenance.balance] section")
}
if got := balance.LoadConfigFromPersistence(cp); !reflect.DeepEqual(got, balanceBefore) {
t.Errorf("balance config changed without [maintenance.balance] section")
}
}
func TestApplyMaintenanceConfigFromTomlEnabledToggle(t *testing.T) {
dir := t.TempDir()
cp := NewConfigPersistence(dir)
if err := cp.ApplyMaintenanceConfigFromToml(tomlConfig(t, "[maintenance]\nenabled = false\n")); err != nil {
t.Fatalf("apply: %v", err)
}
conf, err := cp.LoadMaintenanceConfig()
if err != nil {
t.Fatalf("load maintenance config: %v", err)
}
if conf.GetEnabled() {
t.Errorf("maintenance still enabled after [maintenance] enabled = false")
}
if conf.ScanIntervalSeconds != 30*60 {
t.Errorf("scan interval = %d, want default 1800 kept alongside the toggle", conf.ScanIntervalSeconds)
}
if err := cp.ApplyMaintenanceConfigFromToml(tomlConfig(t, "[maintenance]\nenabled = true\n")); err != nil {
t.Fatalf("apply: %v", err)
}
conf, err = cp.LoadMaintenanceConfig()
if err != nil {
t.Fatalf("load maintenance config: %v", err)
}
if !conf.GetEnabled() {
t.Errorf("maintenance still disabled after [maintenance] enabled = true")
}
}
func TestApplyMaintenanceConfigFromTomlNoKeys(t *testing.T) {
dir := t.TempDir()
cp := NewConfigPersistence(dir)
if err := cp.ApplyMaintenanceConfigFromToml(viper.New()); err != nil {
t.Fatalf("apply: %v", err)
}
if _, err := os.Stat(filepath.Join(dir, ConfigSubdir)); !os.IsNotExist(err) {
t.Errorf("conf dir created without any maintenance keys set")
}
}
func TestApplyMaintenanceConfigFromTomlRequiresDataDir(t *testing.T) {
cp := NewConfigPersistence("")
v := tomlConfig(t, `
[maintenance.vacuum]
garbage_threshold = 0.03
`)
if err := cp.ApplyMaintenanceConfigFromToml(v); err == nil {
t.Errorf("expected error when maintenance keys are set without a data dir")
}
}
func newPluginServer(t *testing.T) *AdminServer {
t.Helper()
p, err := adminplugin.New(adminplugin.Options{})
if err != nil {
t.Fatalf("new plugin: %v", err)
}
t.Cleanup(p.Shutdown)
return &AdminServer{plugin: p}
}
func TestApplyPluginConfigFromTomlUpdatesExistingConfig(t *testing.T) {
s := newPluginServer(t)
seed := &plugin_pb.PersistedJobTypeConfig{
JobType: "vacuum",
AdminRuntime: &plugin_pb.AdminRuntimeConfig{Enabled: true, RetryLimit: 1, RetryBackoffSeconds: 10},
WorkerConfigValues: map[string]*plugin_pb.ConfigValue{
"garbage_threshold": {Kind: &plugin_pb.ConfigValue_DoubleValue{DoubleValue: 0.3}},
},
}
if err := s.GetPlugin().SaveJobTypeConfig(seed); err != nil {
t.Fatalf("seed: %v", err)
}
v := tomlConfig(t, `
[maintenance.vacuum]
garbage_threshold = 0.1
retry_limit = 5
`)
if err := s.ApplyPluginConfigFromToml(v); err != nil {
t.Fatalf("apply: %v", err)
}
cfg, err := s.GetPlugin().LoadJobTypeConfig("vacuum")
if err != nil || cfg == nil {
t.Fatalf("load: %v %v", cfg, err)
}
if got := cfg.WorkerConfigValues["garbage_threshold"].GetDoubleValue(); got != 0.1 {
t.Errorf("garbage_threshold = %v, want 0.1", got)
}
if cfg.AdminRuntime.RetryLimit != 5 {
t.Errorf("retry_limit = %d, want 5", cfg.AdminRuntime.RetryLimit)
}
if !cfg.AdminRuntime.Enabled {
t.Errorf("enabled flipped off by unrelated toml keys")
}
if cfg.AdminRuntime.RetryBackoffSeconds != 10 {
t.Errorf("retry_backoff_seconds = %d, want 10 kept", cfg.AdminRuntime.RetryBackoffSeconds)
}
if cfg.UpdatedBy != "admin.toml" {
t.Errorf("updated_by = %q, want admin.toml", cfg.UpdatedBy)
}
}
func TestApplyPluginConfigFromTomlClampsRetryValues(t *testing.T) {
s := newPluginServer(t)
seed := &plugin_pb.PersistedJobTypeConfig{
JobType: "vacuum",
AdminRuntime: &plugin_pb.AdminRuntimeConfig{Enabled: true},
}
if err := s.GetPlugin().SaveJobTypeConfig(seed); err != nil {
t.Fatalf("seed: %v", err)
}
v := tomlConfig(t, `
[maintenance.vacuum]
retry_limit = 68719476736
retry_backoff_seconds = -5
`)
if err := s.ApplyPluginConfigFromToml(v); err != nil {
t.Fatalf("apply: %v", err)
}
cfg, err := s.GetPlugin().LoadJobTypeConfig("vacuum")
if err != nil || cfg == nil {
t.Fatalf("load: %v %v", cfg, err)
}
wantLimit := int32(math.MaxInt32)
if strconv.IntSize == 32 {
wantLimit = 0 // viper zeroes values that overflow int before the clamp sees them
}
if cfg.AdminRuntime.RetryLimit != wantLimit {
t.Errorf("retry_limit = %d, want clamped to %d", cfg.AdminRuntime.RetryLimit, wantLimit)
}
if cfg.AdminRuntime.RetryBackoffSeconds != 0 {
t.Errorf("retry_backoff_seconds = %d, want clamped to 0", cfg.AdminRuntime.RetryBackoffSeconds)
}
}
func TestApplyPluginConfigFromTomlLeavesMissingConfigToBootstrap(t *testing.T) {
s := newPluginServer(t)
v := tomlConfig(t, `
[maintenance.vacuum]
garbage_threshold = 0.1
`)
if err := s.ApplyPluginConfigFromToml(v); err != nil {
t.Fatalf("apply: %v", err)
}
if cfg, _ := s.GetPlugin().LoadJobTypeConfig("vacuum"); cfg != nil {
t.Errorf("bare config created; descriptor bootstrap would be skipped and disable the job type")
}
}
func TestApplyPluginTomlDefaultsOverlaysBootstrapConfig(t *testing.T) {
cfg := &plugin_pb.PersistedJobTypeConfig{
JobType: "vacuum",
AdminRuntime: &plugin_pb.AdminRuntimeConfig{Enabled: true, RetryLimit: 1},
WorkerConfigValues: map[string]*plugin_pb.ConfigValue{
"garbage_threshold": {Kind: &plugin_pb.ConfigValue_DoubleValue{DoubleValue: 0.3}},
},
UpdatedBy: "plugin",
}
v := tomlConfig(t, `
[maintenance.vacuum]
garbage_threshold = 0.1
`)
applyPluginTomlDefaults(v, cfg)
if got := cfg.WorkerConfigValues["garbage_threshold"].GetDoubleValue(); got != 0.1 {
t.Errorf("garbage_threshold = %v, want 0.1", got)
}
if !cfg.AdminRuntime.Enabled || cfg.AdminRuntime.RetryLimit != 1 {
t.Errorf("descriptor defaults lost: %v", cfg.AdminRuntime)
}
if cfg.UpdatedBy != "admin.toml" {
t.Errorf("updated_by = %q, want admin.toml", cfg.UpdatedBy)
}
}
func TestApplyPluginConfigFromTomlErasureCodingKeyPlacement(t *testing.T) {
s := newPluginServer(t)
seed := &plugin_pb.PersistedJobTypeConfig{
JobType: "erasure_coding",
AdminRuntime: &plugin_pb.AdminRuntimeConfig{Enabled: true},
}
if err := s.GetPlugin().SaveJobTypeConfig(seed); err != nil {
t.Fatalf("seed: %v", err)
}
v := tomlConfig(t, `
[maintenance.erasure_coding]
collection_filter = "important"
replica_placement = "020"
preferred_tags = ["fast,SSD"]
`)
if err := s.ApplyPluginConfigFromToml(v); err != nil {
t.Fatalf("apply: %v", err)
}
cfg, err := s.GetPlugin().LoadJobTypeConfig("erasure_coding")
if err != nil || cfg == nil {
t.Fatalf("load: %v %v", cfg, err)
}
// collection_filter is read from the admin values by all workers
if got := cfg.AdminConfigValues["collection_filter"].GetStringValue(); got != "important" {
t.Errorf("admin collection_filter = %q, want important", got)
}
if _, ok := cfg.WorkerConfigValues["collection_filter"]; ok {
t.Errorf("collection_filter must not land in worker values")
}
if got := cfg.WorkerConfigValues["replica_placement"].GetStringValue(); got != "020" {
t.Errorf("replica_placement = %q, want 020", got)
}
if got := cfg.WorkerConfigValues["preferred_tags"].GetStringList().GetValues(); !reflect.DeepEqual(got, []string{"fast", "ssd"}) {
t.Errorf("preferred_tags = %v, want [fast ssd]", got)
}
}
func TestApplyMaintenanceConfigFromTomlMaxShardTransfers(t *testing.T) {
dir := t.TempDir()
cp := NewConfigPersistence(dir)
if err := cp.ApplyMaintenanceConfigFromToml(tomlConfig(t, `
[maintenance.erasure_coding]
fullness_ratio = 0.9
`)); err != nil {
t.Fatalf("apply fullness: %v", err)
}
ec := erasure_coding.LoadConfigFromPersistence(cp)
if ec.MaxShardTransfers != nil {
t.Fatalf("absent key persisted as %d, want unset", *ec.MaxShardTransfers)
}
if ec.FullnessRatio != 0.9 {
t.Fatalf("fullness_ratio = %v, want 0.9", ec.FullnessRatio)
}
if err := cp.ApplyMaintenanceConfigFromToml(tomlConfig(t, `
[maintenance.erasure_coding]
max_shard_transfers = 0
`)); err != nil {
t.Fatalf("apply 0: %v", err)
}
ec = erasure_coding.LoadConfigFromPersistence(cp)
if ec.MaxShardTransfers == nil || *ec.MaxShardTransfers != 0 {
t.Fatalf("0: got %v, want sequential 0", ec.MaxShardTransfers)
}
if err := cp.ApplyMaintenanceConfigFromToml(tomlConfig(t, `
[maintenance.erasure_coding]
max_shard_transfers = 4
`)); err != nil {
t.Fatalf("apply 4: %v", err)
}
ec = erasure_coding.LoadConfigFromPersistence(cp)
if ec.MaxShardTransfers == nil || *ec.MaxShardTransfers != 4 {
t.Fatalf("4: got %v, want 4", ec.MaxShardTransfers)
}
if err := cp.ApplyMaintenanceConfigFromToml(tomlConfig(t, `
[maintenance.erasure_coding]
max_shard_transfers = -3
`)); err != nil {
t.Fatalf("apply negative: %v", err)
}
ec = erasure_coding.LoadConfigFromPersistence(cp)
if ec.MaxShardTransfers == nil || *ec.MaxShardTransfers != 0 {
t.Fatalf("negative: got %v, want sequential 0", ec.MaxShardTransfers)
}
}
func TestApplyPluginConfigFromTomlMaxShardTransfers(t *testing.T) {
s := newPluginServer(t)
seed := &plugin_pb.PersistedJobTypeConfig{
JobType: "erasure_coding",
AdminRuntime: &plugin_pb.AdminRuntimeConfig{Enabled: true},
}
if err := s.GetPlugin().SaveJobTypeConfig(seed); err != nil {
t.Fatalf("seed: %v", err)
}
if err := s.ApplyPluginConfigFromToml(tomlConfig(t, `
[maintenance.erasure_coding]
replica_placement = "020"
`)); err != nil {
t.Fatalf("apply placement: %v", err)
}
cfg, err := s.GetPlugin().LoadJobTypeConfig("erasure_coding")
if err != nil || cfg == nil {
t.Fatalf("load: %v %v", cfg, err)
}
if _, ok := cfg.WorkerConfigValues["max_shard_transfers"]; ok {
t.Fatal("absent max_shard_transfers was written into worker config")
}
if err := s.ApplyPluginConfigFromToml(tomlConfig(t, `
[maintenance.erasure_coding]
max_shard_transfers = 0
`)); err != nil {
t.Fatalf("apply 0: %v", err)
}
cfg, err = s.GetPlugin().LoadJobTypeConfig("erasure_coding")
if err != nil || cfg == nil {
t.Fatalf("load 0: %v %v", cfg, err)
}
got, ok := cfg.WorkerConfigValues["max_shard_transfers"]
if !ok {
t.Fatal("max_shard_transfers = 0 missing from worker config")
}
if got.GetInt64Value() != 0 {
t.Fatalf("max_shard_transfers = %d, want 0", got.GetInt64Value())
}
}