From 6c4eb95a3a4acd98226433b21ddf4babcfe5602c Mon Sep 17 00:00:00 2001 From: baracudaz Date: Tue, 21 Jul 2026 23:00:37 +0200 Subject: [PATCH] fix(admin): implement ApplyPluginConfigFromToml to propagate settings (#10388) * fix(admin): implement ApplyPluginConfigFromToml to propagate settings to plugin config store * Update weed/admin/dash/config_toml.go Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> * admin: overlay admin.toml onto plugin configs through bootstrap defaults Creating a config from scratch at startup skipped the descriptor-defaults bootstrap, so a job type with only worker keys in admin.toml persisted Enabled=false and RetryLimit=0 and silently stopped running. Overlay existing configs at startup, and apply the same overlay in enrichConfigDefaults when the plugin bootstraps a fresh config from descriptor defaults. Also place collection_filter in the admin values where workers read it, map preferred_tags as a string list, and stamp UpdatedAt. * admin: trim the admin.toml help text and call-site comment * admin: clamp toml retry values to the int32 range * admin: fail startup when admin.toml cannot reach the plugin config The legacy overlay already aborts startup when declared settings cannot persist; continuing here would let workers bootstrap with stale values. * admin: fix the retry clamp test on 32-bit A 32-bit int cannot hold the oversized toml value, so viper returns 0 before the clamp runs. --------- Co-authored-by: baracudaz Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> Co-authored-by: Chris Lu --- weed/admin/dash/admin_server.go | 12 +- weed/admin/dash/config_toml.go | 165 +++++++++++++++++++++++++++ weed/admin/dash/config_toml_test.go | 170 ++++++++++++++++++++++++++++ weed/command/admin.go | 6 + weed/command/scaffold/admin.toml | 12 ++ 5 files changed, 361 insertions(+), 4 deletions(-) diff --git a/weed/admin/dash/admin_server.go b/weed/admin/dash/admin_server.go index 4f2489379..75587c99d 100644 --- a/weed/admin/dash/admin_server.go +++ b/weed/admin/dash/admin_server.go @@ -474,12 +474,16 @@ func (s *AdminServer) loadTaskConfigurationsFromPersistence() { } // enrichConfigDefaults is called by the plugin when bootstrapping a job type's -// default config from its descriptor. For admin_script, it fetches maintenance -// scripts from the master and uses them as the script default. +// default config from its descriptor. It overlays admin.toml maintenance +// settings, and for admin_script fetches maintenance scripts from the master +// to use as the script default. // -// MIGRATION: This exists to help users migrate from master.toml [master.maintenance] -// to the admin script plugin worker. Remove after March 2027. +// MIGRATION: the admin_script part exists to help users migrate from +// master.toml [master.maintenance] to the admin script plugin worker. +// Remove after March 2027. func (s *AdminServer) enrichConfigDefaults(cfg *plugin_pb.PersistedJobTypeConfig) *plugin_pb.PersistedJobTypeConfig { + applyPluginTomlDefaults(util.GetViper(), cfg) + if cfg.JobType != "admin_script" { return cfg } diff --git a/weed/admin/dash/config_toml.go b/weed/admin/dash/config_toml.go index 81e31da72..3872d5b9e 100644 --- a/weed/admin/dash/config_toml.go +++ b/weed/admin/dash/config_toml.go @@ -2,14 +2,17 @@ package dash import ( "fmt" + "math" "strings" "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" "github.com/seaweedfs/seaweedfs/weed/util" "github.com/seaweedfs/seaweedfs/weed/worker/tasks/balance" "github.com/seaweedfs/seaweedfs/weed/worker/tasks/base" "github.com/seaweedfs/seaweedfs/weed/worker/tasks/erasure_coding" "github.com/seaweedfs/seaweedfs/weed/worker/tasks/vacuum" + "google.golang.org/protobuf/types/known/timestamppb" ) // TomlConfig is the subset of viper used to read admin.toml values. @@ -125,3 +128,165 @@ func applyBaseConfigFromToml(v TomlConfig, prefix string, c *base.BaseConfig) bo } return changed } + +// ApplyPluginConfigFromToml overlays admin.toml [maintenance.*] settings onto +// existing plugin job type configs. Configs that do not exist yet get the +// overlay in applyPluginTomlDefaults when the plugin bootstraps them from a +// worker descriptor, so descriptor defaults are never lost. +func (s *AdminServer) ApplyPluginConfigFromToml(v TomlConfig) error { + plugin := s.GetPlugin() + if plugin == nil { + return nil + } + for _, section := range pluginConfigSections { + cfg, err := plugin.LoadJobTypeConfig(section.jobType) + if err != nil { + return fmt.Errorf("load %s plugin config: %w", section.jobType, err) + } + if cfg == nil { + continue + } + if !section.applyToml(v, cfg) { + continue + } + if err := plugin.SaveJobTypeConfig(cfg); err != nil { + return fmt.Errorf("save %s plugin config: %w", section.jobType, err) + } + glog.V(0).Infof("Applied [%s] settings from admin.toml to plugin config", section.prefix) + } + return nil +} + +// applyPluginTomlDefaults overlays admin.toml onto a job type config the +// plugin is bootstrapping from descriptor defaults. +func applyPluginTomlDefaults(v TomlConfig, cfg *plugin_pb.PersistedJobTypeConfig) { + for _, section := range pluginConfigSections { + if section.jobType == cfg.JobType { + if section.applyToml(v, cfg) { + glog.V(0).Infof("Applied [%s] settings from admin.toml to plugin defaults", section.prefix) + } + return + } + } +} + +type pluginConfigSection struct { + jobType string + prefix string + workerKeys map[string]func(v TomlConfig, key string) *plugin_pb.ConfigValue + adminKeys map[string]func(v TomlConfig, key string) *plugin_pb.ConfigValue +} + +var pluginConfigSections = []pluginConfigSection{ + { + jobType: "vacuum", + prefix: "maintenance.vacuum", + workerKeys: map[string]func(v TomlConfig, key string) *plugin_pb.ConfigValue{ + "garbage_threshold": doubleValue, + "min_volume_age_seconds": int64Value, + }, + }, + { + jobType: "volume_balance", + prefix: "maintenance.balance", + workerKeys: map[string]func(v TomlConfig, key string) *plugin_pb.ConfigValue{ + "imbalance_threshold": doubleValue, + "min_server_count": int64Value, + }, + }, + { + 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, + }, + // workers read collection_filter from the admin values, not the worker values + adminKeys: map[string]func(v TomlConfig, key string) *plugin_pb.ConfigValue{ + "collection_filter": stringValue, + }, + }, +} + +func doubleValue(v TomlConfig, key string) *plugin_pb.ConfigValue { + return &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_DoubleValue{DoubleValue: v.GetFloat64(key)}} +} + +func int64Value(v TomlConfig, key string) *plugin_pb.ConfigValue { + return &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: int64(v.GetInt(key))}} +} + +func stringValue(v TomlConfig, key string) *plugin_pb.ConfigValue { + return &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_StringValue{StringValue: v.GetString(key)}} +} + +func stringListValue(v TomlConfig, key string) *plugin_pb.ConfigValue { + // viper does not split comma-separated values from env vars + var items []string + for _, item := range v.GetStringSlice(key) { + items = append(items, strings.Split(item, ",")...) + } + return &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_StringList{StringList: &plugin_pb.StringList{Values: util.NormalizeTagList(items)}}} +} + +func (section pluginConfigSection) applyToml(v TomlConfig, cfg *plugin_pb.PersistedJobTypeConfig) bool { + changed := false + if k := section.prefix + ".enabled"; v.IsSet(k) { + ensureAdminRuntime(cfg).Enabled = v.GetBool(k) + changed = true + } + if k := section.prefix + ".retry_limit"; v.IsSet(k) { + ensureAdminRuntime(cfg).RetryLimit = int32Setting(v, k) + changed = true + } + if k := section.prefix + ".retry_backoff_seconds"; v.IsSet(k) { + ensureAdminRuntime(cfg).RetryBackoffSeconds = int32Setting(v, k) + changed = true + } + for tomlKey, mapper := range section.workerKeys { + if k := section.prefix + "." + tomlKey; v.IsSet(k) { + if cfg.WorkerConfigValues == nil { + cfg.WorkerConfigValues = make(map[string]*plugin_pb.ConfigValue) + } + cfg.WorkerConfigValues[tomlKey] = mapper(v, k) + changed = true + } + } + for tomlKey, mapper := range section.adminKeys { + if k := section.prefix + "." + tomlKey; v.IsSet(k) { + if cfg.AdminConfigValues == nil { + cfg.AdminConfigValues = make(map[string]*plugin_pb.ConfigValue) + } + cfg.AdminConfigValues[tomlKey] = mapper(v, k) + changed = true + } + } + if changed { + cfg.UpdatedAt = timestamppb.Now() + cfg.UpdatedBy = "admin.toml" + } + return changed +} + +// int32Setting clamps to [0, MaxInt32] so oversized toml values cannot wrap negative +func int32Setting(v TomlConfig, key string) int32 { + n := v.GetInt(key) + if n < 0 { + return 0 + } + if n > math.MaxInt32 { + return math.MaxInt32 + } + return int32(n) +} + +func ensureAdminRuntime(cfg *plugin_pb.PersistedJobTypeConfig) *plugin_pb.AdminRuntimeConfig { + if cfg.AdminRuntime == nil { + // maintenance job types are enabled by default; a bare runtime must not disable them + cfg.AdminRuntime = &plugin_pb.AdminRuntimeConfig{Enabled: true} + } + return cfg.AdminRuntime +} diff --git a/weed/admin/dash/config_toml_test.go b/weed/admin/dash/config_toml_test.go index 4e55a8119..c68e31a68 100644 --- a/weed/admin/dash/config_toml_test.go +++ b/weed/admin/dash/config_toml_test.go @@ -1,12 +1,16 @@ 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" @@ -107,3 +111,169 @@ garbage_threshold = 0.03 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) + } +} diff --git a/weed/command/admin.go b/weed/command/admin.go index 161575952..910848cdd 100644 --- a/weed/command/admin.go +++ b/weed/command/admin.go @@ -181,6 +181,8 @@ var cmdAdmin = &Command{ saved from the admin UI, so they can be managed declaratively - Requires -dataDir; values can also be set via WEED_* environment variables, e.g. WEED_MAINTENANCE_VACUUM_GARBAGE_THRESHOLD=0.3 + - Settings are applied to both the legacy task policy and the plugin + config store, so they take effect for the admin UI and plugin workers - Generate example admin.toml: weed scaffold -config=admin Configuration File: @@ -388,6 +390,10 @@ func startAdminServer(ctx context.Context, options AdminOptions, enableUI bool, // Create admin server (plugin is always enabled) adminServer := dash.NewAdminServer(*options.master, *options.filerGroup, nil, dataDir, icebergPort) + if err := adminServer.ApplyPluginConfigFromToml(util.GetViper()); err != nil { + return fmt.Errorf("apply admin.toml to plugin config: %w", err) + } + // Show discovered filers filers := adminServer.GetAllFilers() if len(filers) > 0 { diff --git a/weed/command/scaffold/admin.toml b/weed/command/scaffold/admin.toml index 679053e96..3ea6f554b 100644 --- a/weed/command/scaffold/admin.toml +++ b/weed/command/scaffold/admin.toml @@ -20,6 +20,10 @@ # max_concurrent = 2 # only vacuum volumes older than this (rounded up to whole hours) # min_volume_age_seconds = 86400 +# max retry attempts for a failed vacuum job (admin runtime default) +# retry_limit = 1 +# seconds to wait between retry attempts +# retry_backoff_seconds = 10 [maintenance.balance] # enabled = true @@ -29,6 +33,10 @@ # max_concurrent = 1 # minimum number of volume servers before balancing # min_server_count = 2 +# max retry attempts for a failed balance job +# retry_limit = 1 +# seconds to wait between retry attempts +# retry_backoff_seconds = 15 [maintenance.erasure_coding] # enabled = true @@ -45,3 +53,7 @@ # preferred_tags = ["fast", "ssd"] # EC shard placement constraint, e.g. "020"; empty uses the master default replication # replica_placement = "" +# max retry attempts for a failed erasure coding job +# retry_limit = 1 +# seconds to wait between retry attempts +# retry_backoff_seconds = 30