diff --git a/weed/admin/dash/admin_server.go b/weed/admin/dash/admin_server.go index 623dfeba8..c76815736 100644 --- a/weed/admin/dash/admin_server.go +++ b/weed/admin/dash/admin_server.go @@ -278,12 +278,17 @@ func NewAdminServer(masters string, filerGroup string, templateFS http.FileSyste glog.V(1).Infof("No data directory configured, maintenance system will run in memory-only mode (enabled: %v)", maintenanceConfig.Enabled) } + // Load saved task configurations from persistence. This has to run before the maintenance + // manager is created: creating it applies the maintenance policy to the registered + // detectors and schedulers, while this call replaces each task's whole config object, so + // running it afterwards would discard what the policy just applied. Both read the same + // persisted task config files, so the policy ends up as the last writer and stays + // authoritative for the task types it covers. + server.loadTaskConfigurationsFromPersistence() + // Always initialize maintenance manager server.InitMaintenanceManager(maintenanceConfig) - // Load saved task configurations from persistence - server.loadTaskConfigurationsFromPersistence() - // Start maintenance manager if enabled if maintenanceConfig.Enabled { go func() { @@ -1839,7 +1844,16 @@ func (s *AdminServer) ListPluginSchedulerStates() ([]adminplugin.SchedulerJobTyp // InitMaintenanceManager initializes the maintenance manager func (s *AdminServer) InitMaintenanceManager(config *maintenance.MaintenanceConfig) { - s.maintenanceManager = maintenance.NewMaintenanceManager(s, config) + // Hand the real config store to the manager so that, if it has to build the maintenance policy + // itself, it reads the persisted task configs instead of compiled-in defaults. Only pass it when + // a data directory is actually configured: an unconfigured store has nothing to read, and a typed + // nil pointer would satisfy the loaders' type assertion and then panic on use. + var configPersistence interface{} + if s.configPersistence != nil && s.configPersistence.IsConfigured() { + configPersistence = s.configPersistence + } + + s.maintenanceManager = maintenance.NewMaintenanceManager(s, config, configPersistence) // Set up task persistence if config persistence is available if s.configPersistence != nil { diff --git a/weed/admin/dash/config_persistence.go b/weed/admin/dash/config_persistence.go index fa3a0b43d..fe52abcab 100644 --- a/weed/admin/dash/config_persistence.go +++ b/weed/admin/dash/config_persistence.go @@ -14,6 +14,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/worker_pb" "github.com/seaweedfs/seaweedfs/weed/worker/tasks/balance" + "github.com/seaweedfs/seaweedfs/weed/worker/tasks/ec_balance" "github.com/seaweedfs/seaweedfs/weed/worker/tasks/erasure_coding" "github.com/seaweedfs/seaweedfs/weed/worker/tasks/vacuum" "google.golang.org/protobuf/encoding/protojson" @@ -29,6 +30,7 @@ const ( VacuumTaskConfigFile = "task_vacuum.pb" ECTaskConfigFile = "task_erasure_coding.pb" BalanceTaskConfigFile = "task_balance.pb" + EcBalanceTaskConfigFile = "task_ec_balance.pb" ReplicationTaskConfigFile = "task_replication.pb" // JSON reference files @@ -36,6 +38,7 @@ const ( VacuumTaskConfigJSONFile = "task_vacuum.json" ECTaskConfigJSONFile = "task_erasure_coding.json" BalanceTaskConfigJSONFile = "task_balance.json" + EcBalanceTaskConfigJSONFile = "task_ec_balance.json" ReplicationTaskConfigJSONFile = "task_replication.json" // Task persistence subdirectories and settings @@ -53,6 +56,7 @@ type ( VacuumTaskConfig = worker_pb.VacuumTaskConfig ErasureCodingTaskConfig = worker_pb.ErasureCodingTaskConfig BalanceTaskConfig = worker_pb.BalanceTaskConfig + EcBalanceTaskConfig = worker_pb.EcBalanceTaskConfig ReplicationTaskConfig = worker_pb.ReplicationTaskConfig ) @@ -156,7 +160,7 @@ func (cp *ConfigPersistence) LoadMaintenanceConfig() (*MaintenanceConfig, error) var config MaintenanceConfig if err := proto.Unmarshal(configData, &config); err == nil { // Always populate policy from separate task configuration files - config.Policy = buildPolicyFromTaskConfigs() + config.Policy = cp.buildPolicyFromTaskConfigs() return &config, nil } } @@ -268,6 +272,28 @@ func (cp *ConfigPersistence) RestoreConfig(filename, backupName string) error { return nil } +// Default task policies. These derive from each task's own NewDefaultConfig() so that a +// task type has exactly one definition of its defaults. They used to be hand-written copies +// here, and had drifted from the values the tasks themselves and the admin UI schema use: +// vacuum scanned every 24h instead of 2h, balance every 6h instead of 30m with a 0.1 instead +// of 0.2 imbalance threshold, and erasure coding every 168h instead of 1h with a 0.90 instead +// of 0.95 fullness ratio and a 1024MB instead of 30MB minimum volume size. +func defaultVacuumTaskPolicy() *worker_pb.TaskPolicy { + return vacuum.NewDefaultConfig().ToTaskPolicy() +} + +func defaultErasureCodingTaskPolicy() *worker_pb.TaskPolicy { + return erasure_coding.NewDefaultConfig().ToTaskPolicy() +} + +func defaultBalanceTaskPolicy() *worker_pb.TaskPolicy { + return balance.NewDefaultConfig().ToTaskPolicy() +} + +func defaultEcBalanceTaskPolicy() *worker_pb.TaskPolicy { + return ec_balance.NewDefaultConfig().ToTaskPolicy() +} + // SaveVacuumTaskConfig saves vacuum task configuration to protobuf file func (cp *ConfigPersistence) SaveVacuumTaskConfig(config *VacuumTaskConfig) error { return cp.saveTaskConfig(VacuumTaskConfigFile, config) @@ -288,28 +314,14 @@ func (cp *ConfigPersistence) LoadVacuumTaskConfig() (*VacuumTaskConfig, error) { } // Return default config if no valid config found - return &VacuumTaskConfig{ - GarbageThreshold: 0.3, - MinVolumeAgeHours: 24, - }, nil + return defaultVacuumTaskPolicy().GetVacuumConfig(), nil } // LoadVacuumTaskPolicy loads complete vacuum task policy from protobuf file func (cp *ConfigPersistence) LoadVacuumTaskPolicy() (*worker_pb.TaskPolicy, error) { if cp.dataDir == "" { // Return default policy if no data directory - return &worker_pb.TaskPolicy{ - Enabled: true, - MaxConcurrent: 2, - RepeatIntervalSeconds: 24 * 3600, // 24 hours in seconds - CheckIntervalSeconds: 6 * 3600, // 6 hours in seconds - TaskConfig: &worker_pb.TaskPolicy_VacuumConfig{ - VacuumConfig: &worker_pb.VacuumTaskConfig{ - GarbageThreshold: 0.3, - MinVolumeAgeHours: 24, - }, - }, - }, nil + return defaultVacuumTaskPolicy(), nil } confDir := filepath.Join(cp.dataDir, ConfigSubdir) @@ -318,18 +330,7 @@ func (cp *ConfigPersistence) LoadVacuumTaskPolicy() (*worker_pb.TaskPolicy, erro // Check if file exists if _, err := os.Stat(configPath); os.IsNotExist(err) { // Return default policy if file doesn't exist - return &worker_pb.TaskPolicy{ - Enabled: true, - MaxConcurrent: 2, - RepeatIntervalSeconds: 24 * 3600, // 24 hours in seconds - CheckIntervalSeconds: 6 * 3600, // 6 hours in seconds - TaskConfig: &worker_pb.TaskPolicy_VacuumConfig{ - VacuumConfig: &worker_pb.VacuumTaskConfig{ - GarbageThreshold: 0.3, - MinVolumeAgeHours: 24, - }, - }, - }, nil + return defaultVacuumTaskPolicy(), nil } // Read file @@ -371,32 +372,14 @@ func (cp *ConfigPersistence) LoadErasureCodingTaskConfig() (*ErasureCodingTaskCo } // Return default config if no valid config found - return &ErasureCodingTaskConfig{ - FullnessRatio: 0.9, - QuietForSeconds: 3600, - MinVolumeSizeMb: 1024, - CollectionFilter: "", - }, nil + return defaultErasureCodingTaskPolicy().GetErasureCodingConfig(), nil } // LoadErasureCodingTaskPolicy loads complete EC task policy from protobuf file func (cp *ConfigPersistence) LoadErasureCodingTaskPolicy() (*worker_pb.TaskPolicy, error) { if cp.dataDir == "" { // Return default policy if no data directory - return &worker_pb.TaskPolicy{ - Enabled: true, - MaxConcurrent: 1, - RepeatIntervalSeconds: 168 * 3600, // 1 week in seconds - CheckIntervalSeconds: 24 * 3600, // 24 hours in seconds - TaskConfig: &worker_pb.TaskPolicy_ErasureCodingConfig{ - ErasureCodingConfig: &worker_pb.ErasureCodingTaskConfig{ - FullnessRatio: 0.9, - QuietForSeconds: 3600, - MinVolumeSizeMb: 1024, - CollectionFilter: "", - }, - }, - }, nil + return defaultErasureCodingTaskPolicy(), nil } confDir := filepath.Join(cp.dataDir, ConfigSubdir) @@ -405,20 +388,7 @@ func (cp *ConfigPersistence) LoadErasureCodingTaskPolicy() (*worker_pb.TaskPolic // Check if file exists if _, err := os.Stat(configPath); os.IsNotExist(err) { // Return default policy if file doesn't exist - return &worker_pb.TaskPolicy{ - Enabled: true, - MaxConcurrent: 1, - RepeatIntervalSeconds: 168 * 3600, // 1 week in seconds - CheckIntervalSeconds: 24 * 3600, // 24 hours in seconds - TaskConfig: &worker_pb.TaskPolicy_ErasureCodingConfig{ - ErasureCodingConfig: &worker_pb.ErasureCodingTaskConfig{ - FullnessRatio: 0.9, - QuietForSeconds: 3600, - MinVolumeSizeMb: 1024, - CollectionFilter: "", - }, - }, - }, nil + return defaultErasureCodingTaskPolicy(), nil } // Read file @@ -460,28 +430,14 @@ func (cp *ConfigPersistence) LoadBalanceTaskConfig() (*BalanceTaskConfig, error) } // Return default config if no valid config found - return &BalanceTaskConfig{ - ImbalanceThreshold: 0.1, - MinServerCount: 2, - }, nil + return defaultBalanceTaskPolicy().GetBalanceConfig(), nil } // LoadBalanceTaskPolicy loads complete balance task policy from protobuf file func (cp *ConfigPersistence) LoadBalanceTaskPolicy() (*worker_pb.TaskPolicy, error) { if cp.dataDir == "" { // Return default policy if no data directory - return &worker_pb.TaskPolicy{ - Enabled: true, - MaxConcurrent: 1, - RepeatIntervalSeconds: 6 * 3600, // 6 hours in seconds - CheckIntervalSeconds: 12 * 3600, // 12 hours in seconds - TaskConfig: &worker_pb.TaskPolicy_BalanceConfig{ - BalanceConfig: &worker_pb.BalanceTaskConfig{ - ImbalanceThreshold: 0.1, - MinServerCount: 2, - }, - }, - }, nil + return defaultBalanceTaskPolicy(), nil } confDir := filepath.Join(cp.dataDir, ConfigSubdir) @@ -490,18 +446,7 @@ func (cp *ConfigPersistence) LoadBalanceTaskPolicy() (*worker_pb.TaskPolicy, err // Check if file exists if _, err := os.Stat(configPath); os.IsNotExist(err) { // Return default policy if file doesn't exist - return &worker_pb.TaskPolicy{ - Enabled: true, - MaxConcurrent: 1, - RepeatIntervalSeconds: 6 * 3600, // 6 hours in seconds - CheckIntervalSeconds: 12 * 3600, // 12 hours in seconds - TaskConfig: &worker_pb.TaskPolicy_BalanceConfig{ - BalanceConfig: &worker_pb.BalanceTaskConfig{ - ImbalanceThreshold: 0.1, - MinServerCount: 2, - }, - }, - }, nil + return defaultBalanceTaskPolicy(), nil } // Read file @@ -523,6 +468,60 @@ func (cp *ConfigPersistence) LoadBalanceTaskPolicy() (*worker_pb.TaskPolicy, err return nil, fmt.Errorf("failed to unmarshal balance task configuration") } +// SaveEcBalanceTaskPolicy saves complete EC balance task policy to protobuf file +func (cp *ConfigPersistence) SaveEcBalanceTaskPolicy(policy *worker_pb.TaskPolicy) error { + return cp.saveTaskConfig(EcBalanceTaskConfigFile, policy) +} + +// LoadEcBalanceTaskConfig loads EC balance task configuration from protobuf file +func (cp *ConfigPersistence) LoadEcBalanceTaskConfig() (*EcBalanceTaskConfig, error) { + if taskPolicy, err := cp.LoadEcBalanceTaskPolicy(); err == nil && taskPolicy != nil { + if ecBalanceConfig := taskPolicy.GetEcBalanceConfig(); ecBalanceConfig != nil { + return ecBalanceConfig, nil + } + } + + // Return default config if no valid config found + return defaultEcBalanceTaskPolicy().GetEcBalanceConfig(), nil +} + +// LoadEcBalanceTaskPolicy loads complete EC balance task policy from protobuf file. +// ec_balance is registered like the other maintenance tasks and ec_balance.LoadConfigFromPersistence +// asserts on this accessor, so without it the task could never be configured at all. +func (cp *ConfigPersistence) LoadEcBalanceTaskPolicy() (*worker_pb.TaskPolicy, error) { + if cp.dataDir == "" { + // Return default policy if no data directory + return defaultEcBalanceTaskPolicy(), nil + } + + confDir := filepath.Join(cp.dataDir, ConfigSubdir) + configPath := filepath.Join(confDir, EcBalanceTaskConfigFile) + + // Check if file exists + if _, err := os.Stat(configPath); os.IsNotExist(err) { + // Return default policy if file doesn't exist + return defaultEcBalanceTaskPolicy(), nil + } + + // Read file + configData, err := os.ReadFile(configPath) + if err != nil { + return nil, fmt.Errorf("failed to read EC balance task config file: %w", err) + } + + // Try to unmarshal as TaskPolicy + var policy worker_pb.TaskPolicy + if err := proto.Unmarshal(configData, &policy); err == nil { + // Validate that it's actually a TaskPolicy with EC balance config + if policy.GetEcBalanceConfig() != nil { + glog.V(1).Infof("Loaded EC balance task policy from %s", configPath) + return &policy, nil + } + } + + return nil, fmt.Errorf("failed to unmarshal EC balance task configuration") +} + // SaveReplicationTaskConfig saves replication task configuration to protobuf file func (cp *ConfigPersistence) SaveReplicationTaskConfig(config *ReplicationTaskConfig) error { return cp.saveTaskConfig(ReplicationTaskConfigFile, config) @@ -632,6 +631,8 @@ func (cp *ConfigPersistence) SaveTaskPolicy(taskType string, policy *worker_pb.T return cp.SaveErasureCodingTaskPolicy(policy) case "balance": return cp.SaveBalanceTaskPolicy(policy) + case "ec_balance": + return cp.SaveEcBalanceTaskPolicy(policy) case "replication": return cp.SaveReplicationTaskPolicy(policy) } @@ -687,67 +688,13 @@ func (cp *ConfigPersistence) GetConfigInfo() map[string]interface{} { return info } -// buildPolicyFromTaskConfigs loads task configurations from separate files and builds a MaintenancePolicy -func buildPolicyFromTaskConfigs() *worker_pb.MaintenancePolicy { - policy := &worker_pb.MaintenancePolicy{ - GlobalMaxConcurrent: 4, - DefaultRepeatIntervalSeconds: 6 * 3600, // 6 hours in seconds - DefaultCheckIntervalSeconds: 12 * 3600, // 12 hours in seconds - TaskPolicies: make(map[string]*worker_pb.TaskPolicy), - } - - // Load vacuum task configuration - if vacuumConfig := vacuum.LoadConfigFromPersistence(nil); vacuumConfig != nil { - policy.TaskPolicies["vacuum"] = &worker_pb.TaskPolicy{ - Enabled: vacuumConfig.Enabled, - MaxConcurrent: int32(vacuumConfig.MaxConcurrent), - RepeatIntervalSeconds: int32(vacuumConfig.ScanIntervalSeconds), - CheckIntervalSeconds: int32(vacuumConfig.ScanIntervalSeconds), - TaskConfig: &worker_pb.TaskPolicy_VacuumConfig{ - VacuumConfig: &worker_pb.VacuumTaskConfig{ - GarbageThreshold: float64(vacuumConfig.GarbageThreshold), - MinVolumeAgeHours: int32((vacuumConfig.MinVolumeAgeSeconds + 3599) / 3600), // round up so sub-hour values don't become 0 - }, - }, - } - } - - // Load erasure coding task configuration - if ecConfig := erasure_coding.LoadConfigFromPersistence(nil); ecConfig != nil { - policy.TaskPolicies["erasure_coding"] = &worker_pb.TaskPolicy{ - Enabled: ecConfig.Enabled, - MaxConcurrent: int32(ecConfig.MaxConcurrent), - RepeatIntervalSeconds: int32(ecConfig.ScanIntervalSeconds), - CheckIntervalSeconds: int32(ecConfig.ScanIntervalSeconds), - TaskConfig: &worker_pb.TaskPolicy_ErasureCodingConfig{ - ErasureCodingConfig: &worker_pb.ErasureCodingTaskConfig{ - FullnessRatio: float64(ecConfig.FullnessRatio), - QuietForSeconds: int32(ecConfig.QuietForSeconds), - MinVolumeSizeMb: int32(ecConfig.MinSizeMB), - CollectionFilter: ecConfig.CollectionFilter, - }, - }, - } - } - - // Load balance task configuration - if balanceConfig := balance.LoadConfigFromPersistence(nil); balanceConfig != nil { - policy.TaskPolicies["balance"] = &worker_pb.TaskPolicy{ - Enabled: balanceConfig.Enabled, - MaxConcurrent: int32(balanceConfig.MaxConcurrent), - RepeatIntervalSeconds: int32(balanceConfig.ScanIntervalSeconds), - CheckIntervalSeconds: int32(balanceConfig.ScanIntervalSeconds), - TaskConfig: &worker_pb.TaskPolicy_BalanceConfig{ - BalanceConfig: &worker_pb.BalanceTaskConfig{ - ImbalanceThreshold: float64(balanceConfig.ImbalanceThreshold), - MinServerCount: int32(balanceConfig.MinServerCount), - }, - }, - } - } - - glog.V(1).Infof("Built maintenance policy from separate task configs - %d task policies loaded", len(policy.TaskPolicies)) - return policy +// buildPolicyFromTaskConfigs builds the maintenance policy from the persisted task configs. +// +// The body lives in weed/admin/maintenance because the maintenance manager needs the same +// policy when it has to build one itself, and this package already imports that one. Keeping +// a second copy here is what let the two drift apart in the first place. +func (cp *ConfigPersistence) buildPolicyFromTaskConfigs() *worker_pb.MaintenancePolicy { + return maintenance.BuildPolicyFromTaskConfigs(cp) } // SaveTaskDetail saves detailed task information to disk diff --git a/weed/admin/dash/maintenance_policy_persistence_test.go b/weed/admin/dash/maintenance_policy_persistence_test.go new file mode 100644 index 000000000..77e650fe5 --- /dev/null +++ b/weed/admin/dash/maintenance_policy_persistence_test.go @@ -0,0 +1,74 @@ +package dash + +import ( + "testing" + + "github.com/seaweedfs/seaweedfs/weed/worker/tasks/balance" + "github.com/seaweedfs/seaweedfs/weed/worker/tasks/vacuum" +) + +// TestLoadMaintenanceConfigHonoursPersistedTaskConfigs guards against the regression in +// https://github.com/seaweedfs/seaweedfs/issues/10874: buildPolicyFromTaskConfigs used to call +// LoadConfigFromPersistence(nil), which can never satisfy the loaders' type assertion, so every +// task silently fell back to its compiled-in defaults (Enabled: true) and a task disabled in the +// admin UI kept being scheduled. +func TestLoadMaintenanceConfigHonoursPersistedTaskConfigs(t *testing.T) { + dir := t.TempDir() + cp := NewConfigPersistence(dir) + + // A maintenance.pb must exist, otherwise LoadMaintenanceConfig returns early with defaults. + if err := cp.SaveMaintenanceConfig(DefaultMaintenanceConfig()); err != nil { + t.Fatalf("save maintenance config: %v", err) + } + + // Disable balance and vacuum the way the admin UI does, and change a value that is not a bool + // so a fallback to defaults cannot pass by coincidence. + disabledBalance := balance.NewDefaultConfig() + disabledBalance.Enabled = false + disabledBalance.MinServerCount = 7 + if err := cp.SaveBalanceTaskPolicy(disabledBalance.ToTaskPolicy()); err != nil { + t.Fatalf("save balance policy: %v", err) + } + + disabledVacuum := vacuum.NewDefaultConfig() + disabledVacuum.Enabled = false + if err := cp.SaveVacuumTaskPolicy(disabledVacuum.ToTaskPolicy()); err != nil { + t.Fatalf("save vacuum policy: %v", err) + } + + config, err := cp.LoadMaintenanceConfig() + if err != nil { + t.Fatalf("load maintenance config: %v", err) + } + if config.Policy == nil { + t.Fatal("policy is nil, want it populated from the persisted task configs") + } + + balancePolicy := config.Policy.TaskPolicies["balance"] + if balancePolicy == nil { + t.Fatal("no balance task policy in the built maintenance policy") + } + if balancePolicy.Enabled { + t.Error("balance enabled = true, want false from the persisted config") + } + if got := balancePolicy.GetBalanceConfig().GetMinServerCount(); got != 7 { + t.Errorf("balance min server count = %d, want persisted 7", got) + } + + vacuumPolicy := config.Policy.TaskPolicies["vacuum"] + if vacuumPolicy == nil { + t.Fatal("no vacuum task policy in the built maintenance policy") + } + if vacuumPolicy.Enabled { + t.Error("vacuum enabled = true, want false from the persisted config") + } + + // erasure_coding was never saved, so it keeps the loader's default of enabled. + ecPolicy := config.Policy.TaskPolicies["erasure_coding"] + if ecPolicy == nil { + t.Fatal("no erasure_coding task policy in the built maintenance policy") + } + if !ecPolicy.Enabled { + t.Error("erasure_coding enabled = false, want the default true for a config never saved") + } +} diff --git a/weed/admin/dash/maintenance_startup_test.go b/weed/admin/dash/maintenance_startup_test.go new file mode 100644 index 000000000..7bd9502c5 --- /dev/null +++ b/weed/admin/dash/maintenance_startup_test.go @@ -0,0 +1,123 @@ +package dash + +import ( + "testing" + + "github.com/seaweedfs/seaweedfs/weed/admin/maintenance" + "github.com/seaweedfs/seaweedfs/weed/worker/tasks" + "github.com/seaweedfs/seaweedfs/weed/worker/tasks/balance" + "github.com/seaweedfs/seaweedfs/weed/worker/tasks/vacuum" + "github.com/seaweedfs/seaweedfs/weed/worker/types" +) + +// The task definitions these tests configure are process-global, so put them back the way a +// fresh process would have them. Passing no config store makes every task fall back to its +// own NewDefaultConfig, which is exactly the state package init left them in. +func restoreGlobalTaskState(t *testing.T) { + t.Helper() + + t.Cleanup(func() { + tasks.GetGlobalConfigUpdateRegistry().UpdateAllConfigs(nil) + }) +} + +// TestDisabledTaskIsNotScannedAfterStartup walks the admin server's startup sequence over a +// data directory that has a disabled balance task saved in it, and checks the end state that +// actually matters: the balance detector reports disabled, so ScanWithTaskDetectors skips it. +// +// This is the whole of issue #10874 in one test. The reporter disabled balance, and the +// scanner kept detecting balance tasks, cancelling them and re-detecting them. Two separate +// defects had to line up for the disabled flag to survive to here: the policy had to be built +// from the persisted configs rather than from a nil store, and the policy had to reach +// detector.IsEnabled() rather than dying in a failed type assertion. +func TestDisabledTaskIsNotScannedAfterStartup(t *testing.T) { + restoreGlobalTaskState(t) + + dir := t.TempDir() + cp := NewConfigPersistence(dir) + + // What the admin writes when a user turns balance off, and leaves vacuum on. + disabledBalance := balance.NewDefaultConfig() + disabledBalance.Enabled = false + if err := cp.SaveBalanceTaskPolicy(disabledBalance.ToTaskPolicy()); err != nil { + t.Fatalf("save balance policy: %v", err) + } + + enabledVacuum := vacuum.NewDefaultConfig() + enabledVacuum.Enabled = true + if err := cp.SaveVacuumTaskPolicy(enabledVacuum.ToTaskPolicy()); err != nil { + t.Fatalf("save vacuum policy: %v", err) + } + + // The admin server's startup sequence, in order: + // loadTaskConfigurationsFromPersistence, then InitMaintenanceManager. + tasks.GetGlobalConfigUpdateRegistry().UpdateAllConfigs(cp) + + maintenanceConfig, err := cp.LoadMaintenanceConfig() + if err != nil { + t.Fatalf("load maintenance config: %v", err) + } + manager := maintenance.NewMaintenanceManager(nil, maintenanceConfig, cp) + if manager == nil { + t.Fatal("NewMaintenanceManager returned nil") + } + + registry := tasks.GetGlobalTypesRegistry() + + balanceDetector := registry.GetDetector(types.TaskTypeBalance) + if balanceDetector == nil { + t.Fatal("no balance detector registered") + } + if balanceDetector.IsEnabled() { + t.Error("balance detector reports enabled after startup over a data directory where " + + "balance is saved as disabled; the scanner will keep detecting and cancelling balance tasks") + } + + vacuumDetector := registry.GetDetector(types.TaskTypeVacuum) + if vacuumDetector == nil { + t.Fatal("no vacuum detector registered") + } + if !vacuumDetector.IsEnabled() { + t.Error("vacuum detector reports disabled although vacuum is saved as enabled; " + + "the fix must not switch off tasks the user left on") + } + + // Tasks the user never touched keep their compiled-in default of enabled rather than + // being switched off by a policy entry built from a config that was never saved. + for _, taskType := range []types.TaskType{types.TaskTypeErasureCoding, types.TaskTypeECBalance} { + detector := registry.GetDetector(taskType) + if detector == nil { + t.Fatalf("no %s detector registered", taskType) + } + if !detector.IsEnabled() { + t.Errorf("%s detector reports disabled although its config was never saved", taskType) + } + } +} + +// TestPolicyMirrorsWhatTheDetectorsReport checks that the maintenance policy the queue and +// the scanner run on agrees with the detectors. A disagreement means one of the two paths +// into the task configs has gone stale again. +func TestPolicyMirrorsWhatTheDetectorsReport(t *testing.T) { + restoreGlobalTaskState(t) + + dir := t.TempDir() + cp := NewConfigPersistence(dir) + + disabledBalance := balance.NewDefaultConfig() + disabledBalance.Enabled = false + if err := cp.SaveBalanceTaskPolicy(disabledBalance.ToTaskPolicy()); err != nil { + t.Fatalf("save balance policy: %v", err) + } + + tasks.GetGlobalConfigUpdateRegistry().UpdateAllConfigs(cp) + policy := cp.buildPolicyFromTaskConfigs() + + for taskType, detector := range tasks.GetGlobalTypesRegistry().GetAllDetectors() { + policyEnabled := maintenance.IsTaskEnabled(policy, maintenance.MaintenanceTaskType(taskType)) + if policyEnabled != detector.IsEnabled() { + t.Errorf("%s: policy says enabled=%v but the detector says enabled=%v", + taskType, policyEnabled, detector.IsEnabled()) + } + } +} diff --git a/weed/admin/dash/task_policy_defaults_test.go b/weed/admin/dash/task_policy_defaults_test.go new file mode 100644 index 000000000..29da93352 --- /dev/null +++ b/weed/admin/dash/task_policy_defaults_test.go @@ -0,0 +1,178 @@ +package dash + +import ( + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/worker_pb" + "github.com/seaweedfs/seaweedfs/weed/worker/tasks/balance" + "github.com/seaweedfs/seaweedfs/weed/worker/tasks/ec_balance" + "github.com/seaweedfs/seaweedfs/weed/worker/tasks/erasure_coding" + "github.com/seaweedfs/seaweedfs/weed/worker/tasks/vacuum" + "google.golang.org/protobuf/proto" +) + +// TestLoadTaskPolicyDefaultsMatchTaskDefaults pins the persistence layer's "nothing saved +// yet" defaults to each task's own NewDefaultConfig(). They used to be a second, hand-written +// copy and had drifted: with a data directory but no config file on disk, vacuum ran on a 24h +// scan interval instead of 2h, balance on 6h with a 0.1 imbalance threshold instead of 30m +// with 0.2, and erasure coding on 168h with a 0.90 fullness ratio and a 1024MB minimum volume +// size instead of 1h with 0.95 and 30MB - none of which is what the admin UI shows as the +// default for those fields. +func TestLoadTaskPolicyDefaultsMatchTaskDefaults(t *testing.T) { + cases := []struct { + name string + want *worker_pb.TaskPolicy + load func(cp *ConfigPersistence) (*worker_pb.TaskPolicy, error) + }{ + { + name: "vacuum", + want: vacuum.NewDefaultConfig().ToTaskPolicy(), + load: func(cp *ConfigPersistence) (*worker_pb.TaskPolicy, error) { return cp.LoadVacuumTaskPolicy() }, + }, + { + name: "erasure_coding", + want: erasure_coding.NewDefaultConfig().ToTaskPolicy(), + load: func(cp *ConfigPersistence) (*worker_pb.TaskPolicy, error) { + return cp.LoadErasureCodingTaskPolicy() + }, + }, + { + name: "balance", + want: balance.NewDefaultConfig().ToTaskPolicy(), + load: func(cp *ConfigPersistence) (*worker_pb.TaskPolicy, error) { return cp.LoadBalanceTaskPolicy() }, + }, + { + name: "ec_balance", + want: ec_balance.NewDefaultConfig().ToTaskPolicy(), + load: func(cp *ConfigPersistence) (*worker_pb.TaskPolicy, error) { return cp.LoadEcBalanceTaskPolicy() }, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + // Both no-file branches have to agree with the task's own defaults: no data + // directory at all, and a data directory that has never been written to. + for _, cp := range []*ConfigPersistence{NewConfigPersistence(""), NewConfigPersistence(t.TempDir())} { + got, err := tc.load(cp) + if err != nil { + t.Fatalf("load %s policy: %v", tc.name, err) + } + if !proto.Equal(got, tc.want) { + t.Errorf("%s default policy (dataDir=%q) =\n %v\nwant NewDefaultConfig().ToTaskPolicy() =\n %v", + tc.name, cp.GetDataDir(), got, tc.want) + } + } + }) + } +} + +// TestLoadTaskConfigDefaultsMatchTaskDefaults covers the narrower Load*TaskConfig accessors, +// which carried a third copy of the same defaults. +func TestLoadTaskConfigDefaultsMatchTaskDefaults(t *testing.T) { + cp := NewConfigPersistence(t.TempDir()) + + vacuumConfig, err := cp.LoadVacuumTaskConfig() + if err != nil { + t.Fatalf("load vacuum config: %v", err) + } + if want := vacuum.NewDefaultConfig().ToTaskPolicy().GetVacuumConfig(); !proto.Equal(vacuumConfig, want) { + t.Errorf("vacuum default config = %v, want %v", vacuumConfig, want) + } + + ecConfig, err := cp.LoadErasureCodingTaskConfig() + if err != nil { + t.Fatalf("load erasure coding config: %v", err) + } + if want := erasure_coding.NewDefaultConfig().ToTaskPolicy().GetErasureCodingConfig(); !proto.Equal(ecConfig, want) { + t.Errorf("erasure coding default config = %v, want %v", ecConfig, want) + } + + balanceConfig, err := cp.LoadBalanceTaskConfig() + if err != nil { + t.Fatalf("load balance config: %v", err) + } + if want := balance.NewDefaultConfig().ToTaskPolicy().GetBalanceConfig(); !proto.Equal(balanceConfig, want) { + t.Errorf("balance default config = %v, want %v", balanceConfig, want) + } +} + +// TestEcBalanceTaskPolicyRoundTrip checks the accessor ec_balance.LoadConfigFromPersistence +// asserts on. Before it existed, ec_balance was the one registered maintenance task whose +// configuration could not be persisted at all. +func TestEcBalanceTaskPolicyRoundTrip(t *testing.T) { + cp := NewConfigPersistence(t.TempDir()) + + saved := ec_balance.NewDefaultConfig() + saved.Enabled = false + saved.MinServerCount = 9 + saved.ImbalanceThreshold = 0.42 + saved.CollectionFilter = "pictures" + + if err := cp.SaveEcBalanceTaskPolicy(saved.ToTaskPolicy()); err != nil { + t.Fatalf("save ec_balance policy: %v", err) + } + + loaded := ec_balance.LoadConfigFromPersistence(cp) + if loaded == nil { + t.Fatal("ec_balance.LoadConfigFromPersistence returned nil") + } + if loaded.Enabled { + t.Error("ec_balance enabled = true, want the persisted false") + } + if loaded.MinServerCount != 9 { + t.Errorf("ec_balance min server count = %d, want the persisted 9", loaded.MinServerCount) + } + if loaded.ImbalanceThreshold != 0.42 { + t.Errorf("ec_balance imbalance threshold = %v, want the persisted 0.42", loaded.ImbalanceThreshold) + } + if loaded.CollectionFilter != "pictures" { + t.Errorf("ec_balance collection filter = %q, want the persisted %q", loaded.CollectionFilter, "pictures") + } + + // The generic dispatcher the maintenance manager uses has to know the type too. + if err := cp.SaveTaskPolicy("ec_balance", saved.ToTaskPolicy()); err != nil { + t.Errorf("SaveTaskPolicy(ec_balance): %v", err) + } +} + +// TestBuildPolicyKeepsTaskSpecificFields guards the fields the hand-written policy builder +// used to drop on the floor: the erasure coding preferred tags and replica placement, and +// the balance IO rate limit. Building each entry from the task's own ToTaskPolicy() keeps +// them, so a value set in admin.toml survives into the maintenance policy. +func TestBuildPolicyKeepsTaskSpecificFields(t *testing.T) { + cp := NewConfigPersistence(t.TempDir()) + + ecConfig := erasure_coding.NewDefaultConfig() + ecConfig.PreferredTags = []string{"ssd", "archive"} + ecConfig.ReplicaPlacement = "020" + if err := cp.SaveErasureCodingTaskPolicy(ecConfig.ToTaskPolicy()); err != nil { + t.Fatalf("save erasure coding policy: %v", err) + } + + balanceConfig := balance.NewDefaultConfig() + balanceConfig.IoBytePerSecond = 5 << 20 + if err := cp.SaveBalanceTaskPolicy(balanceConfig.ToTaskPolicy()); err != nil { + t.Fatalf("save balance policy: %v", err) + } + + policy := cp.buildPolicyFromTaskConfigs() + + ecPolicy := policy.TaskPolicies["erasure_coding"].GetErasureCodingConfig() + if ecPolicy == nil { + t.Fatal("no erasure coding config in the built policy") + } + if got := ecPolicy.GetReplicaPlacement(); got != "020" { + t.Errorf("erasure coding replica placement = %q, want the persisted %q", got, "020") + } + if got := ecPolicy.GetPreferredTags(); len(got) != 2 { + t.Errorf("erasure coding preferred tags = %v, want the persisted 2 entries", got) + } + + balancePolicy := policy.TaskPolicies["balance"].GetBalanceConfig() + if balancePolicy == nil { + t.Fatal("no balance config in the built policy") + } + if got := balancePolicy.GetIoBytePerSecond(); got != 5<<20 { + t.Errorf("balance IO limit = %d, want the persisted %d", got, 5<<20) + } +} diff --git a/weed/admin/maintenance/maintenance_integration.go b/weed/admin/maintenance/maintenance_integration.go index 68e72ff0b..04aa06c5d 100644 --- a/weed/admin/maintenance/maintenance_integration.go +++ b/weed/admin/maintenance/maintenance_integration.go @@ -122,6 +122,22 @@ func (s *MaintenanceIntegration) registerAllTasks() { glog.V(1).Infof("Registered tasks: %v", registeredTaskTypes) } +// SetPolicy replaces the maintenance policy the integration configures tasks from and +// applies it immediately. Without this the integration kept the policy it was built with, +// so a policy updated at runtime reached the queue but never the detectors that decide +// which task types are scanned for. +// +// Not safe to call concurrently with a running scan. ConfigureTasksFromPolicy writes +// TaskDefinition.Config and TaskDefinition.MaxConcurrent, which ScanWithTaskDetectors reads +// through detector.IsEnabled(); nothing synchronises the two. Today every caller runs during +// admin server startup, before the scan loop exists. Anything that wires this to an HTTP +// handler has to add that synchronisation first - the same applies to +// tasks.ConfigUpdateRegistry.UpdateAllConfigs, which replaces TaskDefinition.Config outright. +func (s *MaintenanceIntegration) SetPolicy(policy *MaintenancePolicy) { + s.maintenancePolicy = policy + s.ConfigureTasksFromPolicy() +} + // ConfigureTasksFromPolicy dynamically configures all registered tasks based on the maintenance policy func (s *MaintenanceIntegration) ConfigureTasksFromPolicy() { if s.maintenancePolicy == nil { @@ -155,20 +171,32 @@ func (s *MaintenanceIntegration) configureDetectorFromPolicy(taskType types.Task return } - // Apply basic configuration that all detectors should support - if basicDetector, ok := detector.(interface{ SetEnabled(bool) }); ok { - // Convert task system type to maintenance task type for policy lookup - maintenanceTaskType, exists := s.taskTypeMap[taskType] - if exists { - enabled := IsTaskEnabled(s.maintenancePolicy, maintenanceTaskType) - basicDetector.SetEnabled(enabled) - glog.V(3).Infof("Set enabled=%v for detector %s", enabled, taskType) - } + // Convert task system type to maintenance task type for policy lookup + maintenanceTaskType, exists := s.taskTypeMap[taskType] + if !exists { + glog.V(3).Infof("No maintenance task type mapping for %s, skipping configuration", taskType) + return } - // For detectors that don't implement PolicyConfigurableDetector interface, - // they should be updated to implement it for full policy-based configuration - glog.V(2).Infof("Detector %s should implement PolicyConfigurableDetector interface for full policy support", taskType) + // A task type the policy says nothing about is left alone. IsTaskEnabled reports + // false for a missing entry, so applying it unconditionally would silently disable + // every task the policy does not list - which is how ec_balance would have been + // switched off the moment SetEnabled started working. + if GetTaskPolicy(s.maintenancePolicy, maintenanceTaskType) == nil { + glog.V(2).Infof("Maintenance policy has no entry for %s, leaving its detector configuration untouched", taskType) + return + } + + // Apply basic configuration that all detectors should support + if basicDetector, ok := detector.(interface{ SetEnabled(bool) }); ok { + enabled := IsTaskEnabled(s.maintenancePolicy, maintenanceTaskType) + basicDetector.SetEnabled(enabled) + glog.V(3).Infof("Set enabled=%v for detector %s", enabled, taskType) + } else { + // For detectors that don't implement PolicyConfigurableDetector interface, + // they should be updated to implement it for full policy-based configuration + glog.V(2).Infof("Detector %s supports neither PolicyConfigurableDetector nor SetEnabled, its policy is ignored", taskType) + } } // configureSchedulerFromPolicy configures a scheduler using policy-based configuration @@ -187,11 +215,21 @@ func (s *MaintenanceIntegration) configureSchedulerFromPolicy(taskType types.Tas return } + // Same guard as on the detector side: no policy entry means no opinion, not disabled. + if GetTaskPolicy(s.maintenancePolicy, maintenanceTaskType) == nil { + glog.V(2).Infof("Maintenance policy has no entry for %s, leaving its scheduler configuration untouched", taskType) + return + } + // Set enabled status if scheduler supports it if enableableScheduler, ok := scheduler.(interface{ SetEnabled(bool) }); ok { enabled := IsTaskEnabled(s.maintenancePolicy, maintenanceTaskType) enableableScheduler.SetEnabled(enabled) glog.V(3).Infof("Set enabled=%v for scheduler %s", enabled, taskType) + } else { + // For schedulers that don't implement PolicyConfigurableScheduler interface, + // they should be updated to implement it for full policy-based configuration + glog.V(2).Infof("Scheduler %s supports neither PolicyConfigurableScheduler nor SetEnabled, its policy is ignored", taskType) } // Set max concurrent if scheduler supports it @@ -202,10 +240,6 @@ func (s *MaintenanceIntegration) configureSchedulerFromPolicy(taskType types.Tas glog.V(3).Infof("Set max concurrent=%d for scheduler %s", maxConcurrent, taskType) } } - - // For schedulers that don't implement PolicyConfigurableScheduler interface, - // they should be updated to implement it for full policy-based configuration - glog.V(2).Infof("Scheduler %s should implement PolicyConfigurableScheduler interface for full policy support", taskType) } // ScanWithTaskDetectors performs a scan using the task system diff --git a/weed/admin/maintenance/maintenance_manager.go b/weed/admin/maintenance/maintenance_manager.go index 898478107..26c60091c 100644 --- a/weed/admin/maintenance/maintenance_manager.go +++ b/weed/admin/maintenance/maintenance_manager.go @@ -10,12 +10,29 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/worker_pb" stats_collect "github.com/seaweedfs/seaweedfs/weed/stats" "github.com/seaweedfs/seaweedfs/weed/worker/tasks/balance" + "github.com/seaweedfs/seaweedfs/weed/worker/tasks/ec_balance" "github.com/seaweedfs/seaweedfs/weed/worker/tasks/erasure_coding" "github.com/seaweedfs/seaweedfs/weed/worker/tasks/vacuum" + "github.com/seaweedfs/seaweedfs/weed/worker/types" ) -// buildPolicyFromTaskConfigs loads task configurations from separate files and builds a MaintenancePolicy -func buildPolicyFromTaskConfigs() *worker_pb.MaintenancePolicy { +// BuildPolicyFromTaskConfigs loads each registered task's configuration and builds the +// MaintenancePolicy the maintenance system runs on. +// +// Every entry is produced by the task's own ToTaskPolicy(), so a policy entry always carries +// exactly what that task's config holds. Hand-copying the fields here instead made this a +// second, silently diverging definition of every task's policy: it dropped the erasure +// coding preferred tags and replica placement and the balance IO rate limit outright. +// +// Every task registered through base.RegisterTask needs an entry, because a task type the +// policy does not list has no enabled flag and no concurrency limit of its own - +// IsTaskEnabled reports false for a missing entry. +// +// configPersistence is duck-typed as interface{} because weed/admin/dash already imports this +// package, so importing *dash.ConfigPersistence back here would create an import cycle. It must be +// a value implementing the LoadXTaskPolicy() accessors the task loaders assert on; passing nil (or +// anything else) makes every task fall back to its compiled-in defaults. +func BuildPolicyFromTaskConfigs(configPersistence interface{}) *worker_pb.MaintenancePolicy { policy := &worker_pb.MaintenancePolicy{ GlobalMaxConcurrent: 4, DefaultRepeatIntervalSeconds: 6 * 3600, // 6 hours in seconds @@ -23,54 +40,20 @@ func buildPolicyFromTaskConfigs() *worker_pb.MaintenancePolicy { TaskPolicies: make(map[string]*worker_pb.TaskPolicy), } - // Load vacuum task configuration - if vacuumConfig := vacuum.LoadConfigFromPersistence(nil); vacuumConfig != nil { - policy.TaskPolicies["vacuum"] = &worker_pb.TaskPolicy{ - Enabled: vacuumConfig.Enabled, - MaxConcurrent: int32(vacuumConfig.MaxConcurrent), - RepeatIntervalSeconds: int32(vacuumConfig.ScanIntervalSeconds), - CheckIntervalSeconds: int32(vacuumConfig.ScanIntervalSeconds), - TaskConfig: &worker_pb.TaskPolicy_VacuumConfig{ - VacuumConfig: &worker_pb.VacuumTaskConfig{ - GarbageThreshold: float64(vacuumConfig.GarbageThreshold), - MinVolumeAgeHours: int32((vacuumConfig.MinVolumeAgeSeconds + 3599) / 3600), // round up so sub-hour values don't become 0 - }, - }, - } + if vacuumConfig := vacuum.LoadConfigFromPersistence(configPersistence); vacuumConfig != nil { + policy.TaskPolicies[string(types.TaskTypeVacuum)] = vacuumConfig.ToTaskPolicy() } - // Load erasure coding task configuration - if ecConfig := erasure_coding.LoadConfigFromPersistence(nil); ecConfig != nil { - policy.TaskPolicies["erasure_coding"] = &worker_pb.TaskPolicy{ - Enabled: ecConfig.Enabled, - MaxConcurrent: int32(ecConfig.MaxConcurrent), - RepeatIntervalSeconds: int32(ecConfig.ScanIntervalSeconds), - CheckIntervalSeconds: int32(ecConfig.ScanIntervalSeconds), - TaskConfig: &worker_pb.TaskPolicy_ErasureCodingConfig{ - ErasureCodingConfig: &worker_pb.ErasureCodingTaskConfig{ - FullnessRatio: float64(ecConfig.FullnessRatio), - QuietForSeconds: int32(ecConfig.QuietForSeconds), - MinVolumeSizeMb: int32(ecConfig.MinSizeMB), - CollectionFilter: ecConfig.CollectionFilter, - }, - }, - } + if ecConfig := erasure_coding.LoadConfigFromPersistence(configPersistence); ecConfig != nil { + policy.TaskPolicies[string(types.TaskTypeErasureCoding)] = ecConfig.ToTaskPolicy() } - // Load balance task configuration - if balanceConfig := balance.LoadConfigFromPersistence(nil); balanceConfig != nil { - policy.TaskPolicies["balance"] = &worker_pb.TaskPolicy{ - Enabled: balanceConfig.Enabled, - MaxConcurrent: int32(balanceConfig.MaxConcurrent), - RepeatIntervalSeconds: int32(balanceConfig.ScanIntervalSeconds), - CheckIntervalSeconds: int32(balanceConfig.ScanIntervalSeconds), - TaskConfig: &worker_pb.TaskPolicy_BalanceConfig{ - BalanceConfig: &worker_pb.BalanceTaskConfig{ - ImbalanceThreshold: float64(balanceConfig.ImbalanceThreshold), - MinServerCount: int32(balanceConfig.MinServerCount), - }, - }, - } + if balanceConfig := balance.LoadConfigFromPersistence(configPersistence); balanceConfig != nil { + policy.TaskPolicies[string(types.TaskTypeBalance)] = balanceConfig.ToTaskPolicy() + } + + if ecBalanceConfig := ec_balance.LoadConfigFromPersistence(configPersistence); ecBalanceConfig != nil { + policy.TaskPolicies[string(types.TaskTypeECBalance)] = ecBalanceConfig.ToTaskPolicy() } glog.V(1).Infof("Built maintenance policy from separate task configs - %d task policies loaded", len(policy.TaskPolicies)) @@ -94,8 +77,12 @@ type MaintenanceManager struct { scanInProgress bool } -// NewMaintenanceManager creates a new maintenance manager -func NewMaintenanceManager(adminClient AdminClient, config *MaintenanceConfig) *MaintenanceManager { +// NewMaintenanceManager creates a new maintenance manager. +// +// configPersistence is the config store to read persisted task configs from when the policy has to +// be built here. See BuildPolicyFromTaskConfigs for why it is duck-typed; pass nil when no config +// store is available. +func NewMaintenanceManager(adminClient AdminClient, config *MaintenanceConfig, configPersistence interface{}) *MaintenanceManager { if config == nil { config = DefaultMaintenanceConfig() } @@ -104,7 +91,7 @@ func NewMaintenanceManager(adminClient AdminClient, config *MaintenanceConfig) * policy := config.Policy if policy == nil { // Fallback: build policy from separate task configuration files if not already populated - policy = buildPolicyFromTaskConfigs() + policy = BuildPolicyFromTaskConfigs(configPersistence) } queue := NewMaintenanceQueue(policy) @@ -132,7 +119,9 @@ func (mm *MaintenanceManager) Start() error { return fmt.Errorf("invalid maintenance configuration: %w", err) } + mm.mutex.Lock() mm.running = true + mm.mutex.Unlock() // Start background processes go mm.scanLoop() @@ -178,30 +167,54 @@ func (mm *MaintenanceManager) validateConfig() error { return nil } -// IsRunning returns whether the maintenance manager is currently running +// IsRunning returns whether the maintenance manager is currently running. +// running is guarded by mm.mutex because the background loops read it on every +// iteration while Start and Stop are called from the admin server's goroutines. func (mm *MaintenanceManager) IsRunning() bool { + mm.mutex.RLock() + defer mm.mutex.RUnlock() return mm.running } -// Stop terminates the maintenance manager +// Stop terminates the maintenance manager. It is a no-op when the manager is not +// running, so a second call cannot close the already closed stop channel. func (mm *MaintenanceManager) Stop() { + mm.mutex.Lock() + if !mm.running { + mm.mutex.Unlock() + return + } mm.running = false close(mm.stopChan) + mm.mutex.Unlock() + glog.Infof("Maintenance manager stopped") } // scanLoop periodically scans for maintenance tasks with adaptive timing func (mm *MaintenanceManager) scanLoop() { scanInterval := time.Duration(mm.config.ScanIntervalSeconds) * time.Second - ticker := time.NewTicker(scanInterval) - defer ticker.Stop() - for mm.running { + // activeInterval is the interval the ticker is actually running at right now. It has + // to be tracked separately from the configured scanInterval, because the error backoff + // replaces the ticker with a much shorter one. Comparing the target against the + // configured interval instead never restores the normal cadence: once the errors stop, + // getScanInterval returns scanInterval again, the comparison comes out false, and the + // ticker is left at the backoff delay. A single transient scan failure therefore pinned + // the scanner to one scan per second forever - see issue #10874, where that produced a + // ~658 KB/s log flood and 193k orphaned task files. + activeInterval := scanInterval + ticker := time.NewTicker(activeInterval) + // Wrapped in a closure so the replacement ticker is stopped, not the one that happened + // to be current when the defer was registered. + defer func() { ticker.Stop() }() + + for mm.IsRunning() { select { case <-mm.stopChan: return case <-ticker.C: - glog.V(1).Infof("Performing maintenance scan every %v", scanInterval) + glog.V(1).Infof("Performing maintenance scan every %v", activeInterval) // Use the same synchronization as TriggerScan to prevent concurrent scans if err := mm.triggerScanInternal(false); err != nil { @@ -211,10 +224,13 @@ func (mm *MaintenanceManager) scanLoop() { // Adjust ticker interval based on error state (read error state safely) currentInterval := mm.getScanInterval(scanInterval) - // Reset ticker with new interval if needed - if currentInterval != scanInterval { + // Reset ticker whenever the target differs from what the ticker is running at, + // which covers both entering the backoff and returning to the normal cadence. + if currentInterval != activeInterval { ticker.Stop() ticker = time.NewTicker(currentInterval) + glog.V(1).Infof("Maintenance scan cadence changed from %v to %v", activeInterval, currentInterval) + activeInterval = currentInterval } } } @@ -246,7 +262,7 @@ func (mm *MaintenanceManager) cleanupLoop() { ticker := time.NewTicker(cleanupInterval) defer ticker.Stop() - for mm.running { + for mm.IsRunning() { select { case <-mm.stopChan: return @@ -263,7 +279,7 @@ func (mm *MaintenanceManager) topologyStatusLoop() { ticker := time.NewTicker(statusInterval) defer ticker.Stop() - for mm.running { + for mm.IsRunning() { select { case <-mm.stopChan: return @@ -532,12 +548,15 @@ func (mm *MaintenanceManager) TriggerScan() error { // triggerScanInternal handles both manual and automatic scan triggers func (mm *MaintenanceManager) triggerScanInternal(isManual bool) error { + // running and scanInProgress are checked under one lock so a Stop that lands + // between the two checks cannot leave a scan running after shutdown. + mm.mutex.Lock() if !mm.running { + mm.mutex.Unlock() return fmt.Errorf("maintenance manager is not running") } // Prevent multiple concurrent scans - mm.mutex.Lock() if mm.scanInProgress { mm.mutex.Unlock() if isManual { @@ -569,6 +588,14 @@ func (mm *MaintenanceManager) UpdateConfig(config *MaintenanceConfig) error { mm.saveTaskConfigsFromPolicy(config.Policy) } + // The integration holds its own reference to the policy and is what pushes the + // enabled flag and the concurrency limit into the registered detectors and + // schedulers. Without this the queue and the scanner saw the new policy while the + // detectors kept scanning under the old one until the next restart. + if mm.scanner != nil && mm.scanner.integration != nil { + mm.scanner.integration.SetPolicy(config.Policy) + } + glog.V(1).Infof("Maintenance configuration updated") return nil } diff --git a/weed/admin/maintenance/maintenance_manager_test.go b/weed/admin/maintenance/maintenance_manager_test.go index 243a88f5e..06b446baf 100644 --- a/weed/admin/maintenance/maintenance_manager_test.go +++ b/weed/admin/maintenance/maintenance_manager_test.go @@ -10,7 +10,7 @@ func TestMaintenanceManager_ErrorHandling(t *testing.T) { config := DefaultMaintenanceConfig() config.ScanIntervalSeconds = 1 // Short interval for testing (1 second) - manager := NewMaintenanceManager(nil, config) + manager := NewMaintenanceManager(nil, config, nil) // Test initial state if manager.errorCount != 0 { @@ -96,7 +96,7 @@ func TestIsConnectionError(t *testing.T) { func TestMaintenanceManager_GetErrorState(t *testing.T) { config := DefaultMaintenanceConfig() - manager := NewMaintenanceManager(nil, config) + manager := NewMaintenanceManager(nil, config, nil) // Test initial state errorCount, lastError, backoffDelay := manager.GetErrorState() @@ -118,7 +118,7 @@ func TestMaintenanceManager_GetErrorState(t *testing.T) { func TestMaintenanceManager_LogThrottling(t *testing.T) { config := DefaultMaintenanceConfig() - manager := NewMaintenanceManager(nil, config) + manager := NewMaintenanceManager(nil, config, nil) // This is a basic test to ensure the error handling doesn't panic // In practice, you'd want to capture log output to verify throttling diff --git a/weed/admin/maintenance/maintenance_policy_test.go b/weed/admin/maintenance/maintenance_policy_test.go new file mode 100644 index 000000000..adc606c92 --- /dev/null +++ b/weed/admin/maintenance/maintenance_policy_test.go @@ -0,0 +1,134 @@ +package maintenance + +import ( + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/worker_pb" +) + +// stubConfigPersistence implements the LoadXTaskPolicy accessors that the task config loaders +// type-assert on. The real implementation is *dash.ConfigPersistence, which this package cannot +// import: weed/admin/dash already imports weed/admin/maintenance, so the dependency only runs one +// way and the persistence argument has to stay duck-typed. +type stubConfigPersistence struct { + vacuum *worker_pb.TaskPolicy + ec *worker_pb.TaskPolicy + balance *worker_pb.TaskPolicy + ecBalance *worker_pb.TaskPolicy +} + +func (s *stubConfigPersistence) LoadVacuumTaskPolicy() (*worker_pb.TaskPolicy, error) { + return s.vacuum, nil +} + +func (s *stubConfigPersistence) LoadErasureCodingTaskPolicy() (*worker_pb.TaskPolicy, error) { + return s.ec, nil +} + +func (s *stubConfigPersistence) LoadBalanceTaskPolicy() (*worker_pb.TaskPolicy, error) { + return s.balance, nil +} + +func (s *stubConfigPersistence) LoadEcBalanceTaskPolicy() (*worker_pb.TaskPolicy, error) { + return s.ecBalance, nil +} + +func disabledStub() *stubConfigPersistence { + return &stubConfigPersistence{ + vacuum: &worker_pb.TaskPolicy{ + Enabled: false, + MaxConcurrent: 2, + RepeatIntervalSeconds: 2 * 3600, + TaskConfig: &worker_pb.TaskPolicy_VacuumConfig{ + VacuumConfig: &worker_pb.VacuumTaskConfig{GarbageThreshold: 0.3, MinVolumeAgeHours: 24}, + }, + }, + ec: &worker_pb.TaskPolicy{ + Enabled: false, + MaxConcurrent: 1, + RepeatIntervalSeconds: 3600, + TaskConfig: &worker_pb.TaskPolicy_ErasureCodingConfig{ + ErasureCodingConfig: &worker_pb.ErasureCodingTaskConfig{FullnessRatio: 0.95, QuietForSeconds: 3600, MinVolumeSizeMb: 30}, + }, + }, + balance: &worker_pb.TaskPolicy{ + Enabled: false, + MaxConcurrent: 1, + RepeatIntervalSeconds: 30 * 60, + TaskConfig: &worker_pb.TaskPolicy_BalanceConfig{ + BalanceConfig: &worker_pb.BalanceTaskConfig{ImbalanceThreshold: 0.2, MinServerCount: 7}, + }, + }, + ecBalance: &worker_pb.TaskPolicy{ + Enabled: false, + MaxConcurrent: 1, + RepeatIntervalSeconds: 60 * 60, + TaskConfig: &worker_pb.TaskPolicy_EcBalanceConfig{ + EcBalanceConfig: &worker_pb.EcBalanceTaskConfig{ImbalanceThreshold: 0.2, MinServerCount: 5}, + }, + }, + } +} + +// TestBuildPolicyFromTaskConfigsUsesPersistence covers the bug reported in +// https://github.com/seaweedfs/seaweedfs/issues/10874: the persistence argument used to be a +// literal nil, which no type assertion can satisfy, so a task disabled on disk came back enabled. +func TestBuildPolicyFromTaskConfigsUsesPersistence(t *testing.T) { + policy := BuildPolicyFromTaskConfigs(disabledStub()) + + for _, taskType := range []string{"vacuum", "erasure_coding", "balance", "ec_balance"} { + taskPolicy := policy.TaskPolicies[taskType] + if taskPolicy == nil { + t.Fatalf("no %s task policy built", taskType) + } + if taskPolicy.Enabled { + t.Errorf("%s enabled = true, want false from the persisted config", taskType) + } + } + + if got := policy.TaskPolicies["balance"].GetBalanceConfig().GetMinServerCount(); got != 7 { + t.Errorf("balance min server count = %d, want persisted 7", got) + } +} + +// TestBuildPolicyFromTaskConfigsWithoutPersistence keeps the documented fallback: with no config +// store there is nothing to read, so the compiled-in defaults apply. +func TestBuildPolicyFromTaskConfigsWithoutPersistence(t *testing.T) { + policy := BuildPolicyFromTaskConfigs(nil) + + for _, taskType := range []string{"vacuum", "erasure_coding", "balance", "ec_balance"} { + taskPolicy := policy.TaskPolicies[taskType] + if taskPolicy == nil { + t.Fatalf("no %s task policy built", taskType) + } + if !taskPolicy.Enabled { + t.Errorf("%s enabled = false, want the compiled-in default true", taskType) + } + } +} + +// TestNewMaintenanceManagerUsesPersistenceForPolicyFallback covers the path the admin server takes +// when no maintenance.pb has been written yet: the config carries no policy, so the manager builds +// one itself and must read the persisted task configs to do it. +func TestNewMaintenanceManagerUsesPersistenceForPolicyFallback(t *testing.T) { + config := DefaultMaintenanceConfig() + if config.Policy != nil { + t.Fatal("default config unexpectedly carries a policy, test no longer covers the fallback") + } + + manager := NewMaintenanceManager(nil, config, disabledStub()) + + policy := manager.queue.policy + if policy == nil { + t.Fatal("queue policy is nil") + } + if IsTaskEnabled(policy, MaintenanceTaskType("balance")) { + t.Error("balance enabled = true, want false from the persisted config") + } + if IsTaskEnabled(policy, MaintenanceTaskType("vacuum")) { + t.Error("vacuum enabled = true, want false from the persisted config") + } + if manager.scanner.policy != policy { + t.Error("scanner and queue disagree on the policy") + } +} diff --git a/weed/admin/maintenance/maintenance_policy_wiring_test.go b/weed/admin/maintenance/maintenance_policy_wiring_test.go new file mode 100644 index 000000000..c80744b18 --- /dev/null +++ b/weed/admin/maintenance/maintenance_policy_wiring_test.go @@ -0,0 +1,209 @@ +package maintenance + +import ( + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/worker_pb" + "github.com/seaweedfs/seaweedfs/weed/worker/tasks" + "github.com/seaweedfs/seaweedfs/weed/worker/types" +) + +// The detectors and schedulers live in a process-global registry, so these tests restore +// whatever they change. Otherwise a test that disables a task leaks that state into every +// later test in the package. +func snapshotDetectorState(t *testing.T) { + t.Helper() + + registry := tasks.GetGlobalTypesRegistry() + enabled := make(map[types.TaskType]bool) + maxConcurrent := make(map[types.TaskType]int) + for taskType, detector := range registry.GetAllDetectors() { + enabled[taskType] = detector.IsEnabled() + } + for taskType, scheduler := range registry.GetAllSchedulers() { + maxConcurrent[taskType] = scheduler.GetMaxConcurrent() + } + + t.Cleanup(func() { + for taskType, detector := range registry.GetAllDetectors() { + if setter, ok := detector.(interface{ SetEnabled(bool) }); ok { + setter.SetEnabled(enabled[taskType]) + } + } + for taskType, scheduler := range registry.GetAllSchedulers() { + if setter, ok := scheduler.(interface{ SetMaxConcurrent(int) }); ok { + setter.SetMaxConcurrent(maxConcurrent[taskType]) + } + } + }) +} + +// TestPolicyReachesRegisteredDetectors is the regression test for the half of issue #10874 +// that a corrected policy alone did not fix: MaintenanceIntegration pushes the policy into +// detectors and schedulers through interface{ SetEnabled(bool) } and +// interface{ SetMaxConcurrent(int) } type assertions, but every task is backed by +// base.GenericDetector/base.GenericScheduler, which implemented neither. The assertions +// failed silently for every task on every startup, so the policy never reached +// detector.IsEnabled() - which is what ScanWithTaskDetectors gates scanning on. +func TestPolicyReachesRegisteredDetectors(t *testing.T) { + snapshotDetectorState(t) + + registry := tasks.GetGlobalTypesRegistry() + if len(registry.GetAllDetectors()) == 0 { + t.Fatal("no detectors registered, the test cannot prove anything") + } + + // Enable everything first so the disabling below cannot pass by accident. + policy := policyWithAllTasks(t, true) + NewMaintenanceIntegration(NewMaintenanceQueue(policy), policy) + + for taskType, detector := range registry.GetAllDetectors() { + if !detector.IsEnabled() { + t.Fatalf("detector %s is disabled after an all-enabled policy was applied", taskType) + } + } + + // Now disable everything through the policy and check it lands on the detectors. + policy = policyWithAllTasks(t, false) + NewMaintenanceIntegration(NewMaintenanceQueue(policy), policy) + + for taskType, detector := range registry.GetAllDetectors() { + if detector.IsEnabled() { + t.Errorf("detector %s still reports enabled after the policy disabled it; "+ + "the policy is not reaching the flag ScanWithTaskDetectors gates on", taskType) + } + } + for taskType, scheduler := range registry.GetAllSchedulers() { + if scheduler.IsEnabled() { + t.Errorf("scheduler %s still reports enabled after the policy disabled it", taskType) + } + } +} + +// TestPolicyMaxConcurrentReachesSchedulers covers the SetMaxConcurrent half of the same +// wiring. GetMaxConcurrent is what MaintenanceQueue.getMaxConcurrentForTaskType asks before +// starting another task of a type. +func TestPolicyMaxConcurrentReachesSchedulers(t *testing.T) { + snapshotDetectorState(t) + + const wantMaxConcurrent = 7 + + policy := policyWithAllTasks(t, true) + for _, taskPolicy := range policy.TaskPolicies { + taskPolicy.MaxConcurrent = wantMaxConcurrent + } + NewMaintenanceIntegration(NewMaintenanceQueue(policy), policy) + + registry := tasks.GetGlobalTypesRegistry() + for taskType, scheduler := range registry.GetAllSchedulers() { + if _, covered := policy.TaskPolicies[string(taskType)]; !covered { + continue + } + if got := scheduler.GetMaxConcurrent(); got != wantMaxConcurrent { + t.Errorf("scheduler %s max concurrent = %d, want %d from the policy", taskType, got, wantMaxConcurrent) + } + } +} + +// TestPolicyWithoutEntryLeavesTaskAlone guards the direction that would have been a silent +// outage: IsTaskEnabled reports false for a task type the policy does not list, so applying +// it unconditionally would disable every task the policy has no entry for. +func TestPolicyWithoutEntryLeavesTaskAlone(t *testing.T) { + snapshotDetectorState(t) + + registry := tasks.GetGlobalTypesRegistry() + + // Start from a policy that enables everything. + enabling := policyWithAllTasks(t, true) + NewMaintenanceIntegration(NewMaintenanceQueue(enabling), enabling) + + // An empty policy has an entry for nothing at all. + empty := &MaintenancePolicy{TaskPolicies: make(map[string]*worker_pb.TaskPolicy)} + NewMaintenanceIntegration(NewMaintenanceQueue(empty), empty) + + for taskType, detector := range registry.GetAllDetectors() { + if !detector.IsEnabled() { + t.Errorf("detector %s was disabled by a policy that has no entry for it; "+ + "a missing entry means no opinion, not disabled", taskType) + } + } +} + +// TestBuildPolicyCoversEveryRegisteredTask keeps BuildPolicyFromTaskConfigs and the task +// registry in step. A registered task with no policy entry has no enabled flag and no +// concurrency limit of its own, which is the state ec_balance was in. +func TestBuildPolicyCoversEveryRegisteredTask(t *testing.T) { + policy := BuildPolicyFromTaskConfigs(nil) + + for taskType := range tasks.GetGlobalTypesRegistry().GetAllDetectors() { + if _, ok := policy.TaskPolicies[string(taskType)]; !ok { + t.Errorf("task %s is registered as a detector but BuildPolicyFromTaskConfigs "+ + "builds no policy entry for it, so IsTaskEnabled reports false for it", taskType) + } + } +} + +// TestBuildPolicyIncludesEcBalance pins the specific gap above. +func TestBuildPolicyIncludesEcBalance(t *testing.T) { + policy := BuildPolicyFromTaskConfigs(nil) + + ecBalance := policy.TaskPolicies[string(types.TaskTypeECBalance)] + if ecBalance == nil { + t.Fatal("no ec_balance entry in the built maintenance policy") + } + if !IsTaskEnabled(policy, MaintenanceTaskType(types.TaskTypeECBalance)) { + t.Error("ec_balance reports disabled with no persisted config, want the compiled-in default of enabled") + } + if ecBalance.GetEcBalanceConfig() == nil { + t.Error("ec_balance policy entry carries no EcBalanceConfig") + } +} + +// policyWithAllTasks builds a policy that has an entry for every registered task type, +// all sharing the same enabled flag. +func policyWithAllTasks(t *testing.T, enabled bool) *MaintenancePolicy { + t.Helper() + + policy := &MaintenancePolicy{ + GlobalMaxConcurrent: 4, + TaskPolicies: make(map[string]*worker_pb.TaskPolicy), + } + for taskType := range tasks.GetGlobalTypesRegistry().GetAllDetectors() { + policy.TaskPolicies[string(taskType)] = &worker_pb.TaskPolicy{ + Enabled: enabled, + MaxConcurrent: 1, + RepeatIntervalSeconds: 3600, + } + } + return policy +} + +// TestPolicyHelpersTolerateNilPolicy pins the nil handling in the exported policy helpers. +// MaintenanceConfig.Policy is nil until something builds one, and UpdateConfig accepts a +// config that has none, so these are reachable with a nil policy - GetTaskPolicy used to +// dereference it and take the admin process down. +func TestPolicyHelpersTolerateNilPolicy(t *testing.T) { + const taskType = MaintenanceTaskType("balance") + + if got := GetTaskPolicy(nil, taskType); got != nil { + t.Errorf("GetTaskPolicy(nil) = %v, want nil", got) + } + if IsTaskEnabled(nil, taskType) { + t.Error("IsTaskEnabled(nil) = true, want false") + } + if got := GetMaxConcurrent(nil, taskType); got != 1 { + t.Errorf("GetMaxConcurrent(nil) = %d, want the safe default 1", got) + } + if got := GetRepeatInterval(nil, taskType); got != 0 { + t.Errorf("GetRepeatInterval(nil) = %d, want 0 so callers fall back to their own default", got) + } + + // A policy that exists but lists nothing must behave the same way. + empty := &MaintenancePolicy{} + if got := GetTaskPolicy(empty, taskType); got != nil { + t.Errorf("GetTaskPolicy(empty) = %v, want nil", got) + } + if IsTaskEnabled(empty, taskType) { + t.Error("IsTaskEnabled(empty) = true, want false") + } +} diff --git a/weed/admin/maintenance/maintenance_queue_test.go b/weed/admin/maintenance/maintenance_queue_test.go index 7a1328919..5da1de337 100644 --- a/weed/admin/maintenance/maintenance_queue_test.go +++ b/weed/admin/maintenance/maintenance_queue_test.go @@ -618,7 +618,7 @@ func TestMaintenanceQueue_StaleWorkerCapacityRelease(t *testing.T) { func TestMaintenanceManager_CancelTaskCapacityRelease(t *testing.T) { // Setup Manager config := DefaultMaintenanceConfig() - mm := NewMaintenanceManager(nil, config) + mm := NewMaintenanceManager(nil, config, nil) integration := mm.scanner.integration mq := mm.queue at := integration.GetActiveTopology() diff --git a/weed/admin/maintenance/maintenance_scan_cadence_test.go b/weed/admin/maintenance/maintenance_scan_cadence_test.go new file mode 100644 index 000000000..7dc195d78 --- /dev/null +++ b/weed/admin/maintenance/maintenance_scan_cadence_test.go @@ -0,0 +1,168 @@ +package maintenance + +import ( + "sync" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" +) + +// scanCadenceClient is a minimal AdminClient whose first call blocks until the test +// releases it. Blocking the first scan is what makes the cadence test deterministic: +// the scan loop evaluates the error state right after triggering a scan, and that scan +// runs in its own goroutine, so without the block the loop could observe the error +// counter either before or after the scan finished. +type scanCadenceClient struct { + mu sync.Mutex + calls int + release chan struct{} +} + +func (c *scanCadenceClient) WithMasterClient(fn func(client master_pb.SeaweedClient) error) error { + c.mu.Lock() + c.calls++ + first := c.calls == 1 + c.mu.Unlock() + + if first { + <-c.release + } + + // Returning nil without invoking fn yields an empty, successful scan. + return nil +} + +func (c *scanCadenceClient) callCount() int { + c.mu.Lock() + defer c.mu.Unlock() + return c.calls +} + +// TestScanLoopRestoresIntervalAfterBackoff is the regression test for the scan flood in +// https://github.com/seaweedfs/seaweedfs/issues/10874. The loop shortens its ticker to the +// error backoff delay after a failed scan. It used to compare the target interval against +// the *configured* scan interval rather than against the interval the ticker was actually +// running at, so once the errors stopped the comparison came out false and the ticker was +// never restored: one transient failure pinned the scanner to one scan per second for as +// long as the process lived. +// +// The test seeds the error state, lets the loop drop into the 1s backoff, then lets a scan +// succeed and asserts the loop goes back to the configured 3s cadence instead of keeping +// the 1s one. +func TestScanLoopRestoresIntervalAfterBackoff(t *testing.T) { + const baseInterval = 3 * time.Second + + client := &scanCadenceClient{release: make(chan struct{})} + + config := DefaultMaintenanceConfig() + config.ScanIntervalSeconds = int32(baseInterval / time.Second) + manager := NewMaintenanceManager(client, config, nil) + + // Seed a failed scan so the very first cadence decision drops to the backoff delay. + // backoffDelay is what getScanInterval returns while errorCount > 0. + manager.mutex.Lock() + manager.errorCount = 1 + manager.backoffDelay = time.Second + manager.running = true + manager.mutex.Unlock() + + go manager.scanLoop() + defer manager.Stop() + + // First tick at ~3s: the scan blocks in the client, so the loop still sees errorCount == 1 + // and switches the ticker to the 1s backoff. + deadline := time.Now().Add(baseInterval + 2*time.Second) + for client.callCount() == 0 { + if time.Now().After(deadline) { + t.Fatal("scan loop never triggered its first scan") + } + time.Sleep(20 * time.Millisecond) + } + + // Let the blocked scan complete successfully, which clears the error state. + close(client.release) + + // Wait for the error state to clear so the next cadence decision is unambiguous. + deadline = time.Now().Add(2 * time.Second) + for { + errorCount, _, _ := manager.GetErrorState() + if errorCount == 0 { + break + } + if time.Now().After(deadline) { + t.Fatalf("error tracking never reset, errorCount=%d", errorCount) + } + time.Sleep(20 * time.Millisecond) + } + + // The ticker is at 1s now. Wait for the next tick, where the loop must notice the + // recovery and restore the 3s cadence. + time.Sleep(1500 * time.Millisecond) + before := client.callCount() + + // Observe a window that a 1s cadence would fill with scans and a 3s cadence would not. + const window = 5 * time.Second + time.Sleep(window) + scansInWindow := client.callCount() - before + + // 3s cadence: at most 2 scans in 5s. 1s cadence: about 5. + if scansInWindow > 2 { + t.Errorf("scan loop ran %d scans in %v after recovering from a failed scan; "+ + "the ticker was left at the %v backoff instead of returning to the configured %v", + scansInWindow, window, time.Second, baseInterval) + } + if scansInWindow == 0 { + t.Errorf("scan loop ran no scans in %v, expected the %v cadence to fire at least once", window, baseInterval) + } +} + +// TestScanLoopEntersBackoffOnError checks the other half of the cadence logic: a failing +// scan still has to shorten the ticker, so the fix above did not simply pin the loop to +// the configured interval. +func TestScanLoopEntersBackoffOnError(t *testing.T) { + config := DefaultMaintenanceConfig() + config.ScanIntervalSeconds = 600 + manager := NewMaintenanceManager(nil, config, nil) + + baseInterval := time.Duration(config.ScanIntervalSeconds) * time.Second + + if got := manager.getScanInterval(baseInterval); got != baseInterval { + t.Errorf("healthy scan interval = %v, want the configured %v", got, baseInterval) + } + + manager.mutex.Lock() + manager.errorCount = 1 + manager.backoffDelay = time.Second + manager.mutex.Unlock() + + if got := manager.getScanInterval(baseInterval); got != time.Second { + t.Errorf("scan interval while failing = %v, want the %v backoff", got, time.Second) + } + + manager.mutex.Lock() + manager.resetErrorTracking() + manager.mutex.Unlock() + + if got := manager.getScanInterval(baseInterval); got != baseInterval { + t.Errorf("scan interval after recovery = %v, want the configured %v", got, baseInterval) + } +} + +// TestStopIsIdempotent guards the stop channel against a double close, which used to panic +// when StopMaintenanceManager ran twice (for example on a shutdown path that also runs on +// a signal handler). +func TestStopIsIdempotent(t *testing.T) { + manager := NewMaintenanceManager(nil, DefaultMaintenanceConfig(), nil) + + manager.mutex.Lock() + manager.running = true + manager.mutex.Unlock() + + manager.Stop() + manager.Stop() + + if manager.IsRunning() { + t.Error("manager still reports running after Stop") + } +} diff --git a/weed/admin/maintenance/maintenance_types.go b/weed/admin/maintenance/maintenance_types.go index 949440486..1a20d764d 100644 --- a/weed/admin/maintenance/maintenance_types.go +++ b/weed/admin/maintenance/maintenance_types.go @@ -144,9 +144,14 @@ func DefaultMaintenanceConfig() *MaintenanceConfig { // Policy helper functions (since we can't add methods to type aliases) -// GetTaskPolicy returns the policy for a specific task type +// GetTaskPolicy returns the policy for a specific task type, or nil when the maintenance +// policy has no entry for it. +// +// A nil maintenance policy is a legitimate state, not a programming error: +// MaintenanceConfig.Policy is unset until something builds one, and UpdateConfig accepts a +// config that carries none. Dereferencing it here panicked the whole admin process. func GetTaskPolicy(mp *MaintenancePolicy, taskType MaintenanceTaskType) *TaskPolicy { - if mp.TaskPolicies == nil { + if mp == nil || mp.TaskPolicies == nil { return nil } return mp.TaskPolicies[string(taskType)] @@ -174,6 +179,9 @@ func GetMaxConcurrent(mp *MaintenancePolicy, taskType MaintenanceTaskType) int { func GetRepeatInterval(mp *MaintenancePolicy, taskType MaintenanceTaskType) int { policy := GetTaskPolicy(mp, taskType) if policy == nil { + if mp == nil { + return 0 + } return int(mp.DefaultRepeatIntervalSeconds) } return int(policy.RepeatIntervalSeconds) diff --git a/weed/worker/tasks/balance/config.go b/weed/worker/tasks/balance/config.go index 1616788ee..f1372f1cc 100644 --- a/weed/worker/tasks/balance/config.go +++ b/weed/worker/tasks/balance/config.go @@ -178,12 +178,25 @@ func LoadConfigFromPersistence(configPersistence interface{}) *Config { if persistence, ok := configPersistence.(interface { LoadBalanceTaskPolicy() (*worker_pb.TaskPolicy, error) }); ok { - if policy, err := persistence.LoadBalanceTaskPolicy(); err == nil && policy != nil { - if err := config.FromTaskPolicy(policy); err == nil { + policy, err := persistence.LoadBalanceTaskPolicy() + switch { + case err != nil: + glog.Warningf("Could not read the persisted balance configuration, falling back to defaults: %v", err) + case policy == nil: + glog.V(1).Infof("No balance configuration persisted yet, using defaults") + default: + if err := config.FromTaskPolicy(policy); err != nil { + glog.Warningf("Could not apply the persisted balance configuration, falling back to defaults: %v", err) + } else { glog.V(1).Infof("Loaded balance configuration from persistence") return config } } + } else if configPersistence != nil { + // A store was handed in but does not expose the accessor, so the persisted + // settings are silently ignored - always a wiring bug, never a normal state. + glog.Warningf("%T cannot provide the persisted balance configuration: it has no LoadBalanceTaskPolicy() method, "+ + "so the compiled-in defaults are used and any saved balance settings are ignored", configPersistence) } glog.V(1).Infof("Using default balance configuration") diff --git a/weed/worker/tasks/base/generic_components.go b/weed/worker/tasks/base/generic_components.go index 0a41bbd76..259a5f777 100644 --- a/weed/worker/tasks/base/generic_components.go +++ b/weed/worker/tasks/base/generic_components.go @@ -42,6 +42,18 @@ func (d *GenericDetector) IsEnabled() bool { return d.taskDef.Config.IsEnabled() } +// SetEnabled turns detection for this task type on or off. +// +// The admin maintenance policy is applied to detectors through an +// interface{ SetEnabled(bool) } type assertion (see +// MaintenanceIntegration.configureDetectorFromPolicy). Every registered task is +// backed by this generic detector, so without this method that assertion failed +// for every task and the policy never reached the flag that +// ScanWithTaskDetectors actually gates on. See issue #10874. +func (d *GenericDetector) SetEnabled(enabled bool) { + d.taskDef.Config.SetEnabled(enabled) +} + // GenericScheduler implements TaskScheduler using function-based logic type GenericScheduler struct { taskDef *TaskDefinition @@ -127,3 +139,22 @@ func (s *GenericScheduler) GetDefaultRepeatInterval() time.Duration { func (s *GenericScheduler) IsEnabled() bool { return s.taskDef.Config.IsEnabled() } + +// SetEnabled turns scheduling for this task type on or off. Detector and scheduler +// share one TaskDefinition, so this is the same flag GenericDetector.SetEnabled sets; +// both setters exist because the maintenance integration configures the two +// independently. See GenericDetector.SetEnabled and issue #10874. +func (s *GenericScheduler) SetEnabled(enabled bool) { + s.taskDef.Config.SetEnabled(enabled) +} + +// SetMaxConcurrent applies the policy's concurrency limit for this task type. It is +// the value GetMaxConcurrent returns, which the maintenance queue uses to decide +// whether another task of this type may start. Non-positive limits are ignored +// rather than turning into the implicit default of 1. +func (s *GenericScheduler) SetMaxConcurrent(maxConcurrent int) { + if maxConcurrent <= 0 { + return + } + s.taskDef.MaxConcurrent = maxConcurrent +} diff --git a/weed/worker/tasks/ec_balance/config.go b/weed/worker/tasks/ec_balance/config.go index cc3397228..c9e410d6e 100644 --- a/weed/worker/tasks/ec_balance/config.go +++ b/weed/worker/tasks/ec_balance/config.go @@ -241,12 +241,25 @@ func LoadConfigFromPersistence(configPersistence interface{}) *Config { if persistence, ok := configPersistence.(interface { LoadEcBalanceTaskPolicy() (*worker_pb.TaskPolicy, error) }); ok { - if policy, err := persistence.LoadEcBalanceTaskPolicy(); err == nil && policy != nil { - if err := cfg.FromTaskPolicy(policy); err == nil { + policy, err := persistence.LoadEcBalanceTaskPolicy() + switch { + case err != nil: + glog.Warningf("Could not read the persisted EC balance configuration, falling back to defaults: %v", err) + case policy == nil: + glog.V(1).Infof("No EC balance configuration persisted yet, using defaults") + default: + if err := cfg.FromTaskPolicy(policy); err != nil { + glog.Warningf("Could not apply the persisted EC balance configuration, falling back to defaults: %v", err) + } else { glog.V(1).Infof("Loaded EC balance configuration from persistence") return cfg } } + } else if configPersistence != nil { + // A store was handed in but does not expose the accessor, so the persisted + // settings are silently ignored - always a wiring bug, never a normal state. + glog.Warningf("%T cannot provide the persisted EC balance configuration: it has no LoadEcBalanceTaskPolicy() method, "+ + "so the compiled-in defaults are used and any saved EC balance settings are ignored", configPersistence) } glog.V(1).Infof("Using default EC balance configuration") diff --git a/weed/worker/tasks/erasure_coding/config.go b/weed/worker/tasks/erasure_coding/config.go index 48371bb07..7079d1640 100644 --- a/weed/worker/tasks/erasure_coding/config.go +++ b/weed/worker/tasks/erasure_coding/config.go @@ -229,12 +229,25 @@ func LoadConfigFromPersistence(configPersistence interface{}) *Config { if persistence, ok := configPersistence.(interface { LoadErasureCodingTaskPolicy() (*worker_pb.TaskPolicy, error) }); ok { - if policy, err := persistence.LoadErasureCodingTaskPolicy(); err == nil && policy != nil { - if err := config.FromTaskPolicy(policy); err == nil { + policy, err := persistence.LoadErasureCodingTaskPolicy() + switch { + case err != nil: + glog.Warningf("Could not read the persisted erasure coding configuration, falling back to defaults: %v", err) + case policy == nil: + glog.V(1).Infof("No erasure coding configuration persisted yet, using defaults") + default: + if err := config.FromTaskPolicy(policy); err != nil { + glog.Warningf("Could not apply the persisted erasure coding configuration, falling back to defaults: %v", err) + } else { glog.V(1).Infof("Loaded erasure coding configuration from persistence") return config } } + } else if configPersistence != nil { + // A store was handed in but does not expose the accessor, so the persisted + // settings are silently ignored - always a wiring bug, never a normal state. + glog.Warningf("%T cannot provide the persisted erasure coding configuration: it has no LoadErasureCodingTaskPolicy() method, "+ + "so the compiled-in defaults are used and any saved erasure coding settings are ignored", configPersistence) } glog.V(1).Infof("Using default erasure coding configuration") diff --git a/weed/worker/tasks/vacuum/config.go b/weed/worker/tasks/vacuum/config.go index 5d4469707..e97970b65 100644 --- a/weed/worker/tasks/vacuum/config.go +++ b/weed/worker/tasks/vacuum/config.go @@ -73,12 +73,25 @@ func LoadConfigFromPersistence(configPersistence interface{}) *Config { if persistence, ok := configPersistence.(interface { LoadVacuumTaskPolicy() (*worker_pb.TaskPolicy, error) }); ok { - if policy, err := persistence.LoadVacuumTaskPolicy(); err == nil && policy != nil { - if err := config.FromTaskPolicy(policy); err == nil { + policy, err := persistence.LoadVacuumTaskPolicy() + switch { + case err != nil: + glog.Warningf("Could not read the persisted vacuum configuration, falling back to defaults: %v", err) + case policy == nil: + glog.V(1).Infof("No vacuum configuration persisted yet, using defaults") + default: + if err := config.FromTaskPolicy(policy); err != nil { + glog.Warningf("Could not apply the persisted vacuum configuration, falling back to defaults: %v", err) + } else { glog.V(1).Infof("Loaded vacuum configuration from persistence") return config } } + } else if configPersistence != nil { + // A store was handed in but does not expose the accessor, so the persisted + // settings are silently ignored - always a wiring bug, never a normal state. + glog.Warningf("%T cannot provide the persisted vacuum configuration: it has no LoadVacuumTaskPolicy() method, "+ + "so the compiled-in defaults are used and any saved vacuum settings are ignored", configPersistence) } glog.V(1).Infof("Using default vacuum configuration") diff --git a/weed/worker/tasks/vacuum/config_persistence_test.go b/weed/worker/tasks/vacuum/config_persistence_test.go new file mode 100644 index 000000000..c05021109 --- /dev/null +++ b/weed/worker/tasks/vacuum/config_persistence_test.go @@ -0,0 +1,75 @@ +package vacuum + +import ( + "errors" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/worker_pb" +) + +type stubVacuumStore struct { + policy *worker_pb.TaskPolicy + err error +} + +func (s *stubVacuumStore) LoadVacuumTaskPolicy() (*worker_pb.TaskPolicy, error) { + return s.policy, s.err +} + +// wrongShapedStore is what the maintenance manager used to be handed: a non-nil value that +// does not satisfy the accessor the loader asserts on. It has to keep falling back to the +// defaults, but no longer silently - the reporter of issue #10874 had to read the source to +// find out why their disabled task kept running. +type wrongShapedStore struct{} + +func (wrongShapedStore) SomethingElse() {} + +func TestLoadConfigFromPersistenceUsesPersistedPolicy(t *testing.T) { + persisted := NewDefaultConfig() + persisted.Enabled = false + persisted.GarbageThreshold = 0.75 + persisted.MaxConcurrent = 5 + + loaded := LoadConfigFromPersistence(&stubVacuumStore{policy: persisted.ToTaskPolicy()}) + if loaded == nil { + t.Fatal("LoadConfigFromPersistence returned nil") + } + if loaded.Enabled { + t.Error("enabled = true, want the persisted false") + } + if loaded.GarbageThreshold != 0.75 { + t.Errorf("garbage threshold = %v, want the persisted 0.75", loaded.GarbageThreshold) + } + if loaded.MaxConcurrent != 5 { + t.Errorf("max concurrent = %d, want the persisted 5", loaded.MaxConcurrent) + } +} + +func TestLoadConfigFromPersistenceFallsBackToDefaults(t *testing.T) { + defaults := NewDefaultConfig() + + cases := map[string]interface{}{ + "no store configured": nil, + "store without the value": &stubVacuumStore{}, + "store that errors": &stubVacuumStore{err: errors.New("disk on fire")}, + "store of the wrong type": wrongShapedStore{}, + } + + for name, store := range cases { + t.Run(name, func(t *testing.T) { + loaded := LoadConfigFromPersistence(store) + if loaded == nil { + t.Fatal("LoadConfigFromPersistence returned nil") + } + if !loaded.Enabled { + t.Error("enabled = false, want the compiled-in default true") + } + if loaded.GarbageThreshold != defaults.GarbageThreshold { + t.Errorf("garbage threshold = %v, want the default %v", loaded.GarbageThreshold, defaults.GarbageThreshold) + } + if loaded.ScanIntervalSeconds != defaults.ScanIntervalSeconds { + t.Errorf("scan interval = %d, want the default %d", loaded.ScanIntervalSeconds, defaults.ScanIntervalSeconds) + } + }) + } +}