mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-11 17:15:51 +00:00
Compare commits
7
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f1ffea7522 | ||
|
|
389b99f8e8 | ||
|
|
654502ce8f | ||
|
|
1b5e98143e | ||
|
|
d86c692a39 | ||
|
|
e0f07c762a | ||
|
|
2d5ffecc1f |
@@ -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() {
|
||||
@@ -1779,7 +1784,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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -0,0 +1,73 @@
|
||||
package dash
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/worker/tasks/balance"
|
||||
"github.com/seaweedfs/seaweedfs/weed/worker/tasks/vacuum"
|
||||
)
|
||||
|
||||
// TestLoadMaintenanceConfigHonoursPersistedTaskConfigs guards against a regression: 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")
|
||||
}
|
||||
}
|
||||
@@ -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 reported bug 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())
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,184 @@
|
||||
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, and
|
||||
// has to write the same file the dedicated loader reads.
|
||||
dispatched := NewConfigPersistence(t.TempDir())
|
||||
if err := dispatched.SaveTaskPolicy("ec_balance", saved.ToTaskPolicy()); err != nil {
|
||||
t.Fatalf("SaveTaskPolicy(ec_balance): %v", err)
|
||||
}
|
||||
roundTripped := ec_balance.LoadConfigFromPersistence(dispatched)
|
||||
if roundTripped == nil || roundTripped.MinServerCount != 9 {
|
||||
t.Errorf("SaveTaskPolicy(ec_balance) did not round-trip through the ec_balance store: %+v", roundTripped)
|
||||
}
|
||||
}
|
||||
|
||||
// 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 || got[0] != "ssd" || got[1] != "archive" {
|
||||
t.Errorf("erasure coding preferred tags = %v, want the persisted [ssd archive]", 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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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,53 @@ 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.
|
||||
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 +223,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 +261,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 +278,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 +547,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 +587,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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -0,0 +1,133 @@
|
||||
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 a reported bug: 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")
|
||||
}
|
||||
}
|
||||
@@ -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 the scan-flood bug
|
||||
// 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")
|
||||
}
|
||||
}
|
||||
@@ -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()
|
||||
|
||||
@@ -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. 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)
|
||||
callsInWindow := client.callCount() - before
|
||||
|
||||
// One scan makes two master-client calls. 3s cadence: at most 2 scans (4 calls) in 5s,
|
||||
// even if a stall shifts the window. 1s cadence: about 5 scans (10 calls).
|
||||
if callsInWindow > 4 {
|
||||
t.Errorf("scan loop made %d master-client calls in %v after recovering from a failed scan; "+
|
||||
"the ticker was left at the %v backoff instead of returning to the configured %v",
|
||||
callsInWindow, window, time.Second, baseInterval)
|
||||
}
|
||||
if callsInWindow == 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")
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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.
|
||||
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.
|
||||
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
|
||||
}
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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: a fallback the operator cannot see reads as a disabled task
|
||||
// that keeps 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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user