/* * Copyright 2026 Safronov Grigorii * * Licensed under the CDDL, Version 1.0 (the "License"); * you may not use this file except in compliance with the License. * * You may obtain a copy of the License at * https://opensource.org/licenses/CDDL-1.0 */ // Файл: internal/config/dynamic_config.go // Назначение: Централизованное управление конфигурацией через Raft package config import ( "encoding/json" "fmt" "io" "strings" "sync" "sync/atomic" "time" "github.com/hashicorp/raft" "futriis/internal/log" ) // ============================================================================= // ТИПЫ ДЛЯ ДИНАМИЧЕСКОЙ КОНФИГУРАЦИИ // ============================================================================= // DynamicConfigChange представляет изменение конфигурации type DynamicConfigChange struct { ID string `json:"id"` Version uint64 `json:"version"` Timestamp int64 `json:"timestamp"` ChangedBy string `json:"changed_by"` Description string `json:"description"` Changes map[string]interface{} `json:"changes"` } // DynamicConfigSnapshot представляет снэпшот конфигурации type DynamicConfigSnapshot struct { Version uint64 `json:"version"` UpdatedAt int64 `json:"updated_at"` Config *Config `json:"config"` ChangeHistory []*DynamicConfigChange `json:"change_history"` } // ConfigChangeListener представляет слушатель изменений конфигурации type ConfigChangeListener func(change *DynamicConfigChange) // ============================================================================= // DYNAMIC CONFIG MANAGER // ============================================================================= // DynamicConfigManager управляет динамической конфигурацией через Raft type DynamicConfigManager struct { config *Config logger *log.Logger raft *raft.Raft fsm *ConfigFSM mu sync.RWMutex version atomic.Uint64 changeHistory []*DynamicConfigChange listeners map[string]ConfigChangeListener listenerMu sync.RWMutex stopChan chan struct{} wg sync.WaitGroup isLeader atomic.Bool } // ConfigFSM реализует конечный автомат для конфигурации type ConfigFSM struct { config *DynamicConfigSnapshot mu sync.RWMutex logger *log.Logger } // ConfigSnapshot реализует raft.FSMSnapshot type ConfigSnapshot struct { config *DynamicConfigSnapshot } // NewDynamicConfigManager создаёт новый менеджер динамической конфигурации func NewDynamicConfigManager(cfg *Config, logger *log.Logger) *DynamicConfigManager { if cfg == nil { cfg = &Config{} } // Создаём начальный снэпшот initialSnapshot := &DynamicConfigSnapshot{ Version: 1, UpdatedAt: time.Now().UnixMilli(), Config: cfg, ChangeHistory: make([]*DynamicConfigChange, 0), } fsm := &ConfigFSM{ config: initialSnapshot, logger: logger, } dm := &DynamicConfigManager{ config: cfg, logger: logger, fsm: fsm, changeHistory: make([]*DynamicConfigChange, 0), listeners: make(map[string]ConfigChangeListener), stopChan: make(chan struct{}), } return dm } // SetRaft устанавливает Raft для управления конфигурацией func (dm *DynamicConfigManager) SetRaft(raft *raft.Raft) { dm.raft = raft } // GetFSM возвращает FSM для конфигурации func (dm *DynamicConfigManager) GetFSM() *ConfigFSM { return dm.fsm } // Start запускает менеджер динамической конфигурации func (dm *DynamicConfigManager) Start() { if dm.logger != nil { dm.logger.Info("Dynamic config manager started") } } // Stop останавливает менеджер динамической конфигурации func (dm *DynamicConfigManager) Stop() { close(dm.stopChan) dm.wg.Wait() if dm.logger != nil { dm.logger.Info("Dynamic config manager stopped") } } // GetConfig возвращает текущую конфигурацию func (dm *DynamicConfigManager) GetConfig() *Config { dm.fsm.mu.RLock() defer dm.fsm.mu.RUnlock() // Возвращаем копию конфигурации return deepCopyConfig(dm.fsm.config.Config) } // GetVersion возвращает текущую версию конфигурации func (dm *DynamicConfigManager) GetVersion() uint64 { return dm.version.Load() } // GetChangeHistory возвращает историю изменений func (dm *DynamicConfigManager) GetChangeHistory() []*DynamicConfigChange { dm.mu.RLock() defer dm.mu.RUnlock() history := make([]*DynamicConfigChange, len(dm.changeHistory)) copy(history, dm.changeHistory) return history } // ApplyChange применяет изменение конфигурации через Raft func (dm *DynamicConfigManager) ApplyChange(changes map[string]interface{}, description string, changedBy string) error { if dm.raft == nil { return fmt.Errorf("raft not initialized") } if !dm.isLeader.Load() { return fmt.Errorf("not the leader") } // Создаём команду изменения change := &DynamicConfigChange{ ID: fmt.Sprintf("config_%d_%d", dm.version.Load()+1, time.Now().UnixNano()), Version: dm.version.Load() + 1, Timestamp: time.Now().UnixMilli(), ChangedBy: changedBy, Description: description, Changes: changes, } // Сериализуем и применяем через Raft data, err := json.Marshal(change) if err != nil { return fmt.Errorf("failed to marshal config change: %v", err) } future := dm.raft.Apply(data, 10*time.Second) if err := future.Error(); err != nil { return fmt.Errorf("raft apply failed: %v", err) } if dm.logger != nil { dm.logger.Info(fmt.Sprintf("Applied config change: %s (version: %d)", description, change.Version)) } return nil } // UpdateConfig обновляет конфигурацию func (dm *DynamicConfigManager) UpdateConfig(updates map[string]interface{}, description string, changedBy string) error { return dm.ApplyChange(updates, description, changedBy) } // ApplyConfigChange применяет изменение к FSM (вызывается из Raft) func (fsm *ConfigFSM) ApplyConfigChange(change *DynamicConfigChange) error { fsm.mu.Lock() defer fsm.mu.Unlock() // Применяем изменения к конфигурации config := fsm.config.Config // Рекурсивно обновляем поля for path, value := range change.Changes { if err := setConfigField(config, path, value); err != nil { return err } } // Обновляем метаданные fsm.config.Version = change.Version fsm.config.UpdatedAt = time.Now().UnixMilli() fsm.config.ChangeHistory = append(fsm.config.ChangeHistory, change) if fsm.logger != nil { fsm.logger.Debug(fmt.Sprintf("Applied config change to FSM: %s", change.Description)) } return nil } // Apply применяет команду к FSM (реализация raft.FSM) func (fsm *ConfigFSM) Apply(log *raft.Log) interface{} { var change DynamicConfigChange if err := json.Unmarshal(log.Data, &change); err != nil { return fmt.Errorf("failed to unmarshal config change: %v", err) } if err := fsm.ApplyConfigChange(&change); err != nil { return err } return nil } // Snapshot создаёт снэпшот состояния FSM func (fsm *ConfigFSM) Snapshot() (raft.FSMSnapshot, error) { fsm.mu.RLock() defer fsm.mu.RUnlock() // Создаём копию снэпшота snapshot := &ConfigSnapshot{ config: &DynamicConfigSnapshot{ Version: fsm.config.Version, UpdatedAt: fsm.config.UpdatedAt, Config: deepCopyConfig(fsm.config.Config), ChangeHistory: make([]*DynamicConfigChange, len(fsm.config.ChangeHistory)), }, } copy(snapshot.config.ChangeHistory, fsm.config.ChangeHistory) return snapshot, nil } // Restore восстанавливает состояние FSM из снэпшота func (fsm *ConfigFSM) Restore(snapshot io.ReadCloser) error { defer snapshot.Close() var data DynamicConfigSnapshot decoder := json.NewDecoder(snapshot) if err := decoder.Decode(&data); err != nil { return err } fsm.mu.Lock() defer fsm.mu.Unlock() fsm.config = &data return nil } // Persist сохраняет снэпшот func (s *ConfigSnapshot) Persist(sink raft.SnapshotSink) error { data, err := json.Marshal(s.config) if err != nil { sink.Cancel() return err } if _, err := sink.Write(data); err != nil { sink.Cancel() return err } return sink.Close() } // Release освобождает ресурсы снэпшота func (s *ConfigSnapshot) Release() {} // RegisterListener регистрирует слушатель изменений конфигурации func (dm *DynamicConfigManager) RegisterListener(name string, listener ConfigChangeListener) { dm.listenerMu.Lock() defer dm.listenerMu.Unlock() dm.listeners[name] = listener } // UnregisterListener удаляет слушатель func (dm *DynamicConfigManager) UnregisterListener(name string) { dm.listenerMu.Lock() defer dm.listenerMu.Unlock() delete(dm.listeners, name) } // notifyListeners уведомляет слушателей об изменении func (dm *DynamicConfigManager) notifyListeners(change *DynamicConfigChange) { dm.listenerMu.RLock() defer dm.listenerMu.RUnlock() for _, listener := range dm.listeners { go listener(change) } } // ============================================================================= // ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ // ============================================================================= // deepCopyConfig создаёт глубокую копию конфигурации func deepCopyConfig(src *Config) *Config { if src == nil { return nil } // Используем JSON для глубокого копирования data, err := json.Marshal(src) if err != nil { return nil } var dst Config if err := json.Unmarshal(data, &dst); err != nil { return nil } return &dst } // setConfigField устанавливает значение поля конфигурации по пути func setConfigField(config *Config, path string, value interface{}) error { // Разбиваем путь на части parts := strings.Split(path, ".") if len(parts) == 0 { return fmt.Errorf("empty path") } // Простая реализация для основных путей // В реальной production-версии следует использовать рефлексию switch path { case "cluster.name": if v, ok := value.(string); ok { config.Cluster.Name = v } case "cluster.heartbeat_timeout_ms": if v, ok := value.(int); ok { config.Cluster.HeartbeatTimeoutMs = v } else if v, ok := value.(int64); ok { config.Cluster.HeartbeatTimeoutMs = int(v) } else if v, ok := value.(float64); ok { config.Cluster.HeartbeatTimeoutMs = int(v) } case "cluster.election_timeout_ms": if v, ok := value.(int); ok { config.Cluster.ElectionTimeoutMs = v } else if v, ok := value.(int64); ok { config.Cluster.ElectionTimeoutMs = int(v) } else if v, ok := value.(float64); ok { config.Cluster.ElectionTimeoutMs = int(v) } case "cluster.commit_timeout_ms": if v, ok := value.(int); ok { config.Cluster.CommitTimeoutMs = v } else if v, ok := value.(int64); ok { config.Cluster.CommitTimeoutMs = int(v) } else if v, ok := value.(float64); ok { config.Cluster.CommitTimeoutMs = int(v) } case "cluster.snapshot_interval_min": if v, ok := value.(int); ok { config.Cluster.SnapshotIntervalMin = v } else if v, ok := value.(int64); ok { config.Cluster.SnapshotIntervalMin = int(v) } else if v, ok := value.(float64); ok { config.Cluster.SnapshotIntervalMin = int(v) } case "cluster.snapshot_threshold": if v, ok := value.(int); ok { config.Cluster.SnapshotThreshold = v } else if v, ok := value.(int64); ok { config.Cluster.SnapshotThreshold = int(v) } else if v, ok := value.(float64); ok { config.Cluster.SnapshotThreshold = int(v) } case "cluster.recovery_timeout_sec": if v, ok := value.(int); ok { config.Cluster.RecoveryTimeoutSec = v } else if v, ok := value.(int64); ok { config.Cluster.RecoveryTimeoutSec = int(v) } else if v, ok := value.(float64); ok { config.Cluster.RecoveryTimeoutSec = int(v) } case "cluster.split_brain_prevention": if v, ok := value.(bool); ok { config.Cluster.SplitBrainPrevention = v } case "storage.page_size_mb": if v, ok := value.(int); ok { config.Storage.PageSizeMB = v } else if v, ok := value.(int64); ok { config.Storage.PageSizeMB = int(v) } else if v, ok := value.(float64); ok { config.Storage.PageSizeMB = int(v) } case "storage.max_collections": if v, ok := value.(int); ok { config.Storage.MaxCollections = v } else if v, ok := value.(int64); ok { config.Storage.MaxCollections = int(v) } else if v, ok := value.(float64); ok { config.Storage.MaxCollections = int(v) } case "storage.max_documents_per_collection": if v, ok := value.(int); ok { config.Storage.MaxDocumentsPerCollection = v } else if v, ok := value.(int64); ok { config.Storage.MaxDocumentsPerCollection = int(v) } else if v, ok := value.(float64); ok { config.Storage.MaxDocumentsPerCollection = int(v) } case "replication.enabled": if v, ok := value.(bool); ok { config.Replication.Enabled = v } case "replication.sync_replication": if v, ok := value.(bool); ok { config.Replication.SyncReplication = v } case "replication.replication_timeout_ms": if v, ok := value.(int); ok { config.Replication.ReplicationTimeoutMs = v } else if v, ok := value.(int64); ok { config.Replication.ReplicationTimeoutMs = int(v) } else if v, ok := value.(float64); ok { config.Replication.ReplicationTimeoutMs = int(v) } case "replication.max_replica_lag_ms": if v, ok := value.(int); ok { config.Replication.MaxReplicaLagMs = v } else if v, ok := value.(int64); ok { config.Replication.MaxReplicaLagMs = int(v) } else if v, ok := value.(float64); ok { config.Replication.MaxReplicaLagMs = int(v) } case "wal.segment_size_mb": if v, ok := value.(int); ok { config.WAL.SegmentSizeMB = v } else if v, ok := value.(int64); ok { config.WAL.SegmentSizeMB = int(v) } else if v, ok := value.(float64); ok { config.WAL.SegmentSizeMB = int(v) } case "wal.sync_interval_sec": if v, ok := value.(int); ok { config.WAL.SyncIntervalSec = v } else if v, ok := value.(int64); ok { config.WAL.SyncIntervalSec = int(v) } else if v, ok := value.(float64); ok { config.WAL.SyncIntervalSec = int(v) } case "wal.batch_size": if v, ok := value.(int); ok { config.WAL.BatchSize = v } else if v, ok := value.(int64); ok { config.WAL.BatchSize = int(v) } else if v, ok := value.(float64); ok { config.WAL.BatchSize = int(v) } case "wal.enabled": if v, ok := value.(bool); ok { config.WAL.Enabled = v } case "wal.async_recovery": if v, ok := value.(bool); ok { config.WAL.AsyncRecovery = v } case "mvcc.max_versions_per_doc": if v, ok := value.(int); ok { config.MVCC.MaxVersionsPerDoc = v } else if v, ok := value.(int64); ok { config.MVCC.MaxVersionsPerDoc = int(v) } else if v, ok := value.(float64); ok { config.MVCC.MaxVersionsPerDoc = int(v) } case "mvcc.retention_days": if v, ok := value.(int); ok { config.MVCC.RetentionDays = v } else if v, ok := value.(int64); ok { config.MVCC.RetentionDays = int(v) } else if v, ok := value.(float64); ok { config.MVCC.RetentionDays = int(v) } case "saga.enabled": if v, ok := value.(bool); ok { config.Saga.Enabled = v } case "saga.coordinator_count": if v, ok := value.(int); ok { config.Saga.CoordinatorCount = v } else if v, ok := value.(int64); ok { config.Saga.CoordinatorCount = int(v) } else if v, ok := value.(float64); ok { config.Saga.CoordinatorCount = int(v) } case "saga.max_retries": if v, ok := value.(int); ok { config.Saga.MaxRetries = v } else if v, ok := value.(int64); ok { config.Saga.MaxRetries = int(v) } else if v, ok := value.(float64); ok { config.Saga.MaxRetries = int(v) } case "saga.saga_timeout_sec": if v, ok := value.(int); ok { config.Saga.SagaTimeoutSec = v } else if v, ok := value.(int64); ok { config.Saga.SagaTimeoutSec = int(v) } else if v, ok := value.(float64); ok { config.Saga.SagaTimeoutSec = int(v) } case "backpressure.enabled": if v, ok := value.(bool); ok { config.Backpressure.Enabled = v } case "backpressure.cpu_threshold": if v, ok := value.(float64); ok { config.Backpressure.CPUThreshold = v } case "backpressure.memory_threshold": if v, ok := value.(float64); ok { config.Backpressure.MemoryThreshold = v } case "api.port": if v, ok := value.(int); ok { config.API.Port = v } else if v, ok := value.(int64); ok { config.API.Port = int(v) } else if v, ok := value.(float64); ok { config.API.Port = int(v) } case "log.log_level": if v, ok := value.(string); ok { config.Log.LogLevel = v } case "log.log_file": if v, ok := value.(string); ok { config.Log.LogFile = v } default: // Для неизвестных путей логируем предупреждение // В production-версии следует использовать рефлексию для полного покрытия return fmt.Errorf("unknown config path: %s", path) } return nil } // GetLeaderStatus возвращает статус лидера func (dm *DynamicConfigManager) GetLeaderStatus() bool { return dm.isLeader.Load() } // SetLeaderStatus устанавливает статус лидера func (dm *DynamicConfigManager) SetLeaderStatus(isLeader bool) { dm.isLeader.Store(isLeader) } // GetListeners возвращает список зарегистрированных слушателей func (dm *DynamicConfigManager) GetListeners() []string { dm.listenerMu.RLock() defer dm.listenerMu.RUnlock() names := make([]string, 0, len(dm.listeners)) for name := range dm.listeners { names = append(names, name) } return names }