You can not select more than 25 topics
			Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
		
		
		
		
		
			
		
			
				
					
					
						
							690 lines
						
					
					
						
							23 KiB
						
					
					
				
			
		
		
		
			
			
			
		
		
	
	
							690 lines
						
					
					
						
							23 KiB
						
					
					
				| package dash | |
| 
 | |
| import ( | |
| 	"fmt" | |
| 	"os" | |
| 	"path/filepath" | |
| 	"time" | |
| 
 | |
| 	"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/erasure_coding" | |
| 	"github.com/seaweedfs/seaweedfs/weed/worker/tasks/vacuum" | |
| 	"google.golang.org/protobuf/encoding/protojson" | |
| 	"google.golang.org/protobuf/proto" | |
| ) | |
| 
 | |
| const ( | |
| 	// Configuration subdirectory | |
| 	ConfigSubdir = "conf" | |
| 
 | |
| 	// Configuration file names (protobuf binary) | |
| 	MaintenanceConfigFile     = "maintenance.pb" | |
| 	VacuumTaskConfigFile      = "task_vacuum.pb" | |
| 	ECTaskConfigFile          = "task_erasure_coding.pb" | |
| 	BalanceTaskConfigFile     = "task_balance.pb" | |
| 	ReplicationTaskConfigFile = "task_replication.pb" | |
| 
 | |
| 	// JSON reference files | |
| 	MaintenanceConfigJSONFile     = "maintenance.json" | |
| 	VacuumTaskConfigJSONFile      = "task_vacuum.json" | |
| 	ECTaskConfigJSONFile          = "task_erasure_coding.json" | |
| 	BalanceTaskConfigJSONFile     = "task_balance.json" | |
| 	ReplicationTaskConfigJSONFile = "task_replication.json" | |
| 
 | |
| 	ConfigDirPermissions  = 0755 | |
| 	ConfigFilePermissions = 0644 | |
| ) | |
| 
 | |
| // Task configuration types | |
| type ( | |
| 	VacuumTaskConfig        = worker_pb.VacuumTaskConfig | |
| 	ErasureCodingTaskConfig = worker_pb.ErasureCodingTaskConfig | |
| 	BalanceTaskConfig       = worker_pb.BalanceTaskConfig | |
| 	ReplicationTaskConfig   = worker_pb.ReplicationTaskConfig | |
| ) | |
| 
 | |
| // ConfigPersistence handles saving and loading configuration files | |
| type ConfigPersistence struct { | |
| 	dataDir string | |
| } | |
| 
 | |
| // NewConfigPersistence creates a new configuration persistence manager | |
| func NewConfigPersistence(dataDir string) *ConfigPersistence { | |
| 	return &ConfigPersistence{ | |
| 		dataDir: dataDir, | |
| 	} | |
| } | |
| 
 | |
| // SaveMaintenanceConfig saves maintenance configuration to protobuf file and JSON reference | |
| func (cp *ConfigPersistence) SaveMaintenanceConfig(config *MaintenanceConfig) error { | |
| 	if cp.dataDir == "" { | |
| 		return fmt.Errorf("no data directory specified, cannot save configuration") | |
| 	} | |
| 
 | |
| 	confDir := filepath.Join(cp.dataDir, ConfigSubdir) | |
| 	if err := os.MkdirAll(confDir, ConfigDirPermissions); err != nil { | |
| 		return fmt.Errorf("failed to create config directory: %w", err) | |
| 	} | |
| 
 | |
| 	// Save as protobuf (primary format) | |
| 	pbConfigPath := filepath.Join(confDir, MaintenanceConfigFile) | |
| 	pbData, err := proto.Marshal(config) | |
| 	if err != nil { | |
| 		return fmt.Errorf("failed to marshal maintenance config to protobuf: %w", err) | |
| 	} | |
| 
 | |
| 	if err := os.WriteFile(pbConfigPath, pbData, ConfigFilePermissions); err != nil { | |
| 		return fmt.Errorf("failed to write protobuf config file: %w", err) | |
| 	} | |
| 
 | |
| 	// Save JSON reference copy for debugging | |
| 	jsonConfigPath := filepath.Join(confDir, MaintenanceConfigJSONFile) | |
| 	jsonData, err := protojson.MarshalOptions{ | |
| 		Multiline:       true, | |
| 		Indent:          "  ", | |
| 		EmitUnpopulated: true, | |
| 	}.Marshal(config) | |
| 	if err != nil { | |
| 		return fmt.Errorf("failed to marshal maintenance config to JSON: %w", err) | |
| 	} | |
| 
 | |
| 	if err := os.WriteFile(jsonConfigPath, jsonData, ConfigFilePermissions); err != nil { | |
| 		return fmt.Errorf("failed to write JSON reference file: %w", err) | |
| 	} | |
| 
 | |
| 	return nil | |
| } | |
| 
 | |
| // LoadMaintenanceConfig loads maintenance configuration from protobuf file | |
| func (cp *ConfigPersistence) LoadMaintenanceConfig() (*MaintenanceConfig, error) { | |
| 	if cp.dataDir == "" { | |
| 		return DefaultMaintenanceConfig(), nil | |
| 	} | |
| 
 | |
| 	confDir := filepath.Join(cp.dataDir, ConfigSubdir) | |
| 	configPath := filepath.Join(confDir, MaintenanceConfigFile) | |
| 
 | |
| 	// Try to load from protobuf file | |
| 	if configData, err := os.ReadFile(configPath); err == nil { | |
| 		var config MaintenanceConfig | |
| 		if err := proto.Unmarshal(configData, &config); err == nil { | |
| 			// Always populate policy from separate task configuration files | |
| 			config.Policy = buildPolicyFromTaskConfigs() | |
| 			return &config, nil | |
| 		} | |
| 	} | |
| 
 | |
| 	// File doesn't exist or failed to load, use defaults | |
| 	return DefaultMaintenanceConfig(), nil | |
| } | |
| 
 | |
| // GetConfigPath returns the path to a configuration file | |
| func (cp *ConfigPersistence) GetConfigPath(filename string) string { | |
| 	if cp.dataDir == "" { | |
| 		return "" | |
| 	} | |
| 
 | |
| 	// All configs go in conf subdirectory | |
| 	confDir := filepath.Join(cp.dataDir, ConfigSubdir) | |
| 	return filepath.Join(confDir, filename) | |
| } | |
| 
 | |
| // ListConfigFiles returns all configuration files in the conf subdirectory | |
| func (cp *ConfigPersistence) ListConfigFiles() ([]string, error) { | |
| 	if cp.dataDir == "" { | |
| 		return nil, fmt.Errorf("no data directory specified") | |
| 	} | |
| 
 | |
| 	confDir := filepath.Join(cp.dataDir, ConfigSubdir) | |
| 	files, err := os.ReadDir(confDir) | |
| 	if err != nil { | |
| 		// If conf directory doesn't exist, return empty list | |
| 		if os.IsNotExist(err) { | |
| 			return []string{}, nil | |
| 		} | |
| 		return nil, fmt.Errorf("failed to read config directory: %w", err) | |
| 	} | |
| 
 | |
| 	var configFiles []string | |
| 	for _, file := range files { | |
| 		if !file.IsDir() { | |
| 			ext := filepath.Ext(file.Name()) | |
| 			if ext == ".json" || ext == ".pb" { | |
| 				configFiles = append(configFiles, file.Name()) | |
| 			} | |
| 		} | |
| 	} | |
| 
 | |
| 	return configFiles, nil | |
| } | |
| 
 | |
| // BackupConfig creates a backup of a configuration file | |
| func (cp *ConfigPersistence) BackupConfig(filename string) error { | |
| 	if cp.dataDir == "" { | |
| 		return fmt.Errorf("no data directory specified") | |
| 	} | |
| 
 | |
| 	configPath := cp.GetConfigPath(filename) | |
| 	if _, err := os.Stat(configPath); os.IsNotExist(err) { | |
| 		return fmt.Errorf("config file does not exist: %s", filename) | |
| 	} | |
| 
 | |
| 	// Create backup filename with timestamp | |
| 	timestamp := time.Now().Format("2006-01-02_15-04-05") | |
| 	backupName := fmt.Sprintf("%s.backup_%s", filename, timestamp) | |
| 
 | |
| 	// Determine backup directory (conf subdirectory) | |
| 	confDir := filepath.Join(cp.dataDir, ConfigSubdir) | |
| 	backupPath := filepath.Join(confDir, backupName) | |
| 
 | |
| 	// Copy file | |
| 	configData, err := os.ReadFile(configPath) | |
| 	if err != nil { | |
| 		return fmt.Errorf("failed to read config file: %w", err) | |
| 	} | |
| 
 | |
| 	if err := os.WriteFile(backupPath, configData, ConfigFilePermissions); err != nil { | |
| 		return fmt.Errorf("failed to create backup: %w", err) | |
| 	} | |
| 
 | |
| 	glog.V(1).Infof("Created backup of %s as %s", filename, backupName) | |
| 	return nil | |
| } | |
| 
 | |
| // RestoreConfig restores a configuration file from a backup | |
| func (cp *ConfigPersistence) RestoreConfig(filename, backupName string) error { | |
| 	if cp.dataDir == "" { | |
| 		return fmt.Errorf("no data directory specified") | |
| 	} | |
| 
 | |
| 	// Determine backup path (conf subdirectory) | |
| 	confDir := filepath.Join(cp.dataDir, ConfigSubdir) | |
| 	backupPath := filepath.Join(confDir, backupName) | |
| 
 | |
| 	if _, err := os.Stat(backupPath); os.IsNotExist(err) { | |
| 		return fmt.Errorf("backup file does not exist: %s", backupName) | |
| 	} | |
| 
 | |
| 	// Read backup file | |
| 	backupData, err := os.ReadFile(backupPath) | |
| 	if err != nil { | |
| 		return fmt.Errorf("failed to read backup file: %w", err) | |
| 	} | |
| 
 | |
| 	// Write to config file | |
| 	configPath := cp.GetConfigPath(filename) | |
| 	if err := os.WriteFile(configPath, backupData, ConfigFilePermissions); err != nil { | |
| 		return fmt.Errorf("failed to restore config: %w", err) | |
| 	} | |
| 
 | |
| 	glog.V(1).Infof("Restored %s from backup %s", filename, backupName) | |
| 	return nil | |
| } | |
| 
 | |
| // SaveVacuumTaskConfig saves vacuum task configuration to protobuf file | |
| func (cp *ConfigPersistence) SaveVacuumTaskConfig(config *VacuumTaskConfig) error { | |
| 	return cp.saveTaskConfig(VacuumTaskConfigFile, config) | |
| } | |
| 
 | |
| // SaveVacuumTaskPolicy saves complete vacuum task policy to protobuf file | |
| func (cp *ConfigPersistence) SaveVacuumTaskPolicy(policy *worker_pb.TaskPolicy) error { | |
| 	return cp.saveTaskConfig(VacuumTaskConfigFile, policy) | |
| } | |
| 
 | |
| // LoadVacuumTaskConfig loads vacuum task configuration from protobuf file | |
| func (cp *ConfigPersistence) LoadVacuumTaskConfig() (*VacuumTaskConfig, error) { | |
| 	// Load as TaskPolicy and extract vacuum config | |
| 	if taskPolicy, err := cp.LoadVacuumTaskPolicy(); err == nil && taskPolicy != nil { | |
| 		if vacuumConfig := taskPolicy.GetVacuumConfig(); vacuumConfig != nil { | |
| 			return vacuumConfig, nil | |
| 		} | |
| 	} | |
| 
 | |
| 	// Return default config if no valid config found | |
| 	return &VacuumTaskConfig{ | |
| 		GarbageThreshold:   0.3, | |
| 		MinVolumeAgeHours:  24, | |
| 		MinIntervalSeconds: 7 * 24 * 60 * 60, // 7 days | |
| 	}, 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, | |
| 					MinIntervalSeconds: 7 * 24 * 60 * 60, // 7 days | |
| 				}, | |
| 			}, | |
| 		}, nil | |
| 	} | |
| 
 | |
| 	confDir := filepath.Join(cp.dataDir, ConfigSubdir) | |
| 	configPath := filepath.Join(confDir, VacuumTaskConfigFile) | |
| 
 | |
| 	// 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, | |
| 					MinIntervalSeconds: 7 * 24 * 60 * 60, // 7 days | |
| 				}, | |
| 			}, | |
| 		}, nil | |
| 	} | |
| 
 | |
| 	// Read file | |
| 	configData, err := os.ReadFile(configPath) | |
| 	if err != nil { | |
| 		return nil, fmt.Errorf("failed to read vacuum 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 vacuum config | |
| 		if policy.GetVacuumConfig() != nil { | |
| 			glog.V(1).Infof("Loaded vacuum task policy from %s", configPath) | |
| 			return &policy, nil | |
| 		} | |
| 	} | |
| 
 | |
| 	return nil, fmt.Errorf("failed to unmarshal vacuum task configuration") | |
| } | |
| 
 | |
| // SaveErasureCodingTaskConfig saves EC task configuration to protobuf file | |
| func (cp *ConfigPersistence) SaveErasureCodingTaskConfig(config *ErasureCodingTaskConfig) error { | |
| 	return cp.saveTaskConfig(ECTaskConfigFile, config) | |
| } | |
| 
 | |
| // SaveErasureCodingTaskPolicy saves complete EC task policy to protobuf file | |
| func (cp *ConfigPersistence) SaveErasureCodingTaskPolicy(policy *worker_pb.TaskPolicy) error { | |
| 	return cp.saveTaskConfig(ECTaskConfigFile, policy) | |
| } | |
| 
 | |
| // LoadErasureCodingTaskConfig loads EC task configuration from protobuf file | |
| func (cp *ConfigPersistence) LoadErasureCodingTaskConfig() (*ErasureCodingTaskConfig, error) { | |
| 	// Load as TaskPolicy and extract EC config | |
| 	if taskPolicy, err := cp.LoadErasureCodingTaskPolicy(); err == nil && taskPolicy != nil { | |
| 		if ecConfig := taskPolicy.GetErasureCodingConfig(); ecConfig != nil { | |
| 			return ecConfig, nil | |
| 		} | |
| 	} | |
| 
 | |
| 	// Return default config if no valid config found | |
| 	return &ErasureCodingTaskConfig{ | |
| 		FullnessRatio:    0.9, | |
| 		QuietForSeconds:  3600, | |
| 		MinVolumeSizeMb:  1024, | |
| 		CollectionFilter: "", | |
| 	}, 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 | |
| 	} | |
| 
 | |
| 	confDir := filepath.Join(cp.dataDir, ConfigSubdir) | |
| 	configPath := filepath.Join(confDir, ECTaskConfigFile) | |
| 
 | |
| 	// 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 | |
| 	} | |
| 
 | |
| 	// Read file | |
| 	configData, err := os.ReadFile(configPath) | |
| 	if err != nil { | |
| 		return nil, fmt.Errorf("failed to read EC 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 config | |
| 		if policy.GetErasureCodingConfig() != nil { | |
| 			glog.V(1).Infof("Loaded EC task policy from %s", configPath) | |
| 			return &policy, nil | |
| 		} | |
| 	} | |
| 
 | |
| 	return nil, fmt.Errorf("failed to unmarshal EC task configuration") | |
| } | |
| 
 | |
| // SaveBalanceTaskConfig saves balance task configuration to protobuf file | |
| func (cp *ConfigPersistence) SaveBalanceTaskConfig(config *BalanceTaskConfig) error { | |
| 	return cp.saveTaskConfig(BalanceTaskConfigFile, config) | |
| } | |
| 
 | |
| // SaveBalanceTaskPolicy saves complete balance task policy to protobuf file | |
| func (cp *ConfigPersistence) SaveBalanceTaskPolicy(policy *worker_pb.TaskPolicy) error { | |
| 	return cp.saveTaskConfig(BalanceTaskConfigFile, policy) | |
| } | |
| 
 | |
| // LoadBalanceTaskConfig loads balance task configuration from protobuf file | |
| func (cp *ConfigPersistence) LoadBalanceTaskConfig() (*BalanceTaskConfig, error) { | |
| 	// Load as TaskPolicy and extract balance config | |
| 	if taskPolicy, err := cp.LoadBalanceTaskPolicy(); err == nil && taskPolicy != nil { | |
| 		if balanceConfig := taskPolicy.GetBalanceConfig(); balanceConfig != nil { | |
| 			return balanceConfig, nil | |
| 		} | |
| 	} | |
| 
 | |
| 	// Return default config if no valid config found | |
| 	return &BalanceTaskConfig{ | |
| 		ImbalanceThreshold: 0.1, | |
| 		MinServerCount:     2, | |
| 	}, 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 | |
| 	} | |
| 
 | |
| 	confDir := filepath.Join(cp.dataDir, ConfigSubdir) | |
| 	configPath := filepath.Join(confDir, BalanceTaskConfigFile) | |
| 
 | |
| 	// 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 | |
| 	} | |
| 
 | |
| 	// Read file | |
| 	configData, err := os.ReadFile(configPath) | |
| 	if err != nil { | |
| 		return nil, fmt.Errorf("failed to read 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 balance config | |
| 		if policy.GetBalanceConfig() != nil { | |
| 			glog.V(1).Infof("Loaded balance task policy from %s", configPath) | |
| 			return &policy, nil | |
| 		} | |
| 	} | |
| 
 | |
| 	return nil, fmt.Errorf("failed to unmarshal balance task configuration") | |
| } | |
| 
 | |
| // SaveReplicationTaskConfig saves replication task configuration to protobuf file | |
| func (cp *ConfigPersistence) SaveReplicationTaskConfig(config *ReplicationTaskConfig) error { | |
| 	return cp.saveTaskConfig(ReplicationTaskConfigFile, config) | |
| } | |
| 
 | |
| // LoadReplicationTaskConfig loads replication task configuration from protobuf file | |
| func (cp *ConfigPersistence) LoadReplicationTaskConfig() (*ReplicationTaskConfig, error) { | |
| 	var config ReplicationTaskConfig | |
| 	err := cp.loadTaskConfig(ReplicationTaskConfigFile, &config) | |
| 	if err != nil { | |
| 		// Return default config if file doesn't exist | |
| 		if os.IsNotExist(err) { | |
| 			return &ReplicationTaskConfig{ | |
| 				TargetReplicaCount: 1, | |
| 			}, nil | |
| 		} | |
| 		return nil, err | |
| 	} | |
| 	return &config, nil | |
| } | |
| 
 | |
| // saveTaskConfig is a generic helper for saving task configurations with both protobuf and JSON reference | |
| func (cp *ConfigPersistence) saveTaskConfig(filename string, config proto.Message) error { | |
| 	if cp.dataDir == "" { | |
| 		return fmt.Errorf("no data directory specified, cannot save task configuration") | |
| 	} | |
| 
 | |
| 	// Create conf subdirectory path | |
| 	confDir := filepath.Join(cp.dataDir, ConfigSubdir) | |
| 	configPath := filepath.Join(confDir, filename) | |
| 
 | |
| 	// Generate JSON reference filename | |
| 	jsonFilename := filename[:len(filename)-3] + ".json" // Replace .pb with .json | |
| 	jsonPath := filepath.Join(confDir, jsonFilename) | |
| 
 | |
| 	// Create conf directory if it doesn't exist | |
| 	if err := os.MkdirAll(confDir, ConfigDirPermissions); err != nil { | |
| 		return fmt.Errorf("failed to create config directory: %w", err) | |
| 	} | |
| 
 | |
| 	// Marshal configuration to protobuf binary format | |
| 	configData, err := proto.Marshal(config) | |
| 	if err != nil { | |
| 		return fmt.Errorf("failed to marshal task config: %w", err) | |
| 	} | |
| 
 | |
| 	// Write protobuf file | |
| 	if err := os.WriteFile(configPath, configData, ConfigFilePermissions); err != nil { | |
| 		return fmt.Errorf("failed to write task config file: %w", err) | |
| 	} | |
| 
 | |
| 	// Marshal configuration to JSON for reference | |
| 	marshaler := protojson.MarshalOptions{ | |
| 		Multiline:       true, | |
| 		Indent:          "  ", | |
| 		EmitUnpopulated: true, | |
| 	} | |
| 	jsonData, err := marshaler.Marshal(config) | |
| 	if err != nil { | |
| 		glog.Warningf("Failed to marshal task config to JSON reference: %v", err) | |
| 	} else { | |
| 		// Write JSON reference file | |
| 		if err := os.WriteFile(jsonPath, jsonData, ConfigFilePermissions); err != nil { | |
| 			glog.Warningf("Failed to write task config JSON reference: %v", err) | |
| 		} | |
| 	} | |
| 
 | |
| 	glog.V(1).Infof("Saved task configuration to %s (with JSON reference)", configPath) | |
| 	return nil | |
| } | |
| 
 | |
| // loadTaskConfig is a generic helper for loading task configurations from conf subdirectory | |
| func (cp *ConfigPersistence) loadTaskConfig(filename string, config proto.Message) error { | |
| 	if cp.dataDir == "" { | |
| 		return os.ErrNotExist // Will trigger default config return | |
| 	} | |
| 
 | |
| 	confDir := filepath.Join(cp.dataDir, ConfigSubdir) | |
| 	configPath := filepath.Join(confDir, filename) | |
| 
 | |
| 	// Check if file exists | |
| 	if _, err := os.Stat(configPath); os.IsNotExist(err) { | |
| 		return err // Will trigger default config return | |
| 	} | |
| 
 | |
| 	// Read file | |
| 	configData, err := os.ReadFile(configPath) | |
| 	if err != nil { | |
| 		return fmt.Errorf("failed to read task config file: %w", err) | |
| 	} | |
| 
 | |
| 	// Unmarshal protobuf binary data | |
| 	if err := proto.Unmarshal(configData, config); err != nil { | |
| 		return fmt.Errorf("failed to unmarshal task config: %w", err) | |
| 	} | |
| 
 | |
| 	glog.V(1).Infof("Loaded task configuration from %s", configPath) | |
| 	return nil | |
| } | |
| 
 | |
| // GetDataDir returns the data directory path | |
| func (cp *ConfigPersistence) GetDataDir() string { | |
| 	return cp.dataDir | |
| } | |
| 
 | |
| // IsConfigured returns true if a data directory is configured | |
| func (cp *ConfigPersistence) IsConfigured() bool { | |
| 	return cp.dataDir != "" | |
| } | |
| 
 | |
| // GetConfigInfo returns information about the configuration storage | |
| func (cp *ConfigPersistence) GetConfigInfo() map[string]interface{} { | |
| 	info := map[string]interface{}{ | |
| 		"data_dir_configured": cp.IsConfigured(), | |
| 		"data_dir":            cp.dataDir, | |
| 		"config_subdir":       ConfigSubdir, | |
| 	} | |
| 
 | |
| 	if cp.IsConfigured() { | |
| 		// Check if data directory exists | |
| 		if _, err := os.Stat(cp.dataDir); err == nil { | |
| 			info["data_dir_exists"] = true | |
| 
 | |
| 			// Check if conf subdirectory exists | |
| 			confDir := filepath.Join(cp.dataDir, ConfigSubdir) | |
| 			if _, err := os.Stat(confDir); err == nil { | |
| 				info["conf_dir_exists"] = true | |
| 
 | |
| 				// List config files | |
| 				configFiles, err := cp.ListConfigFiles() | |
| 				if err == nil { | |
| 					info["config_files"] = configFiles | |
| 				} | |
| 			} else { | |
| 				info["conf_dir_exists"] = false | |
| 			} | |
| 		} else { | |
| 			info["data_dir_exists"] = false | |
| 		} | |
| 	} | |
| 
 | |
| 	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 / 3600), // Convert seconds to hours | |
| 					MinIntervalSeconds: int32(vacuumConfig.MinIntervalSeconds), | |
| 				}, | |
| 			}, | |
| 		} | |
| 	} | |
| 
 | |
| 	// 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 | |
| }
 |