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 <baracudaz@users.noreply.github.com>
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
Co-authored-by: Chris Lu <chris.lu@gmail.com>

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
 	}

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
+}

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)
+	}
+}

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 {

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