From fb017a93519dc158a10d572a6917f08fe74162aa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=93=D1=80=D0=B8=D0=B3=D0=BE=D1=80=D0=B8=D0=B9=20=D0=A1?= =?UTF-8?q?=D0=B0=D1=84=D1=80=D0=BE=D0=BD=D0=BE=D0=B2?= Date: Wed, 16 Sep 2026 22:11:10 +0000 Subject: [PATCH] Update internal/config/dynamic_config.go --- internal/config/dynamic_config.go | 859 ++++++++++++++++++++++++------ 1 file changed, 689 insertions(+), 170 deletions(-) diff --git a/internal/config/dynamic_config.go b/internal/config/dynamic_config.go index 587f2e6..6ada71e 100644 --- a/internal/config/dynamic_config.go +++ b/internal/config/dynamic_config.go @@ -26,6 +26,31 @@ import ( "futriis/internal/log" ) +// ============================================================================= +// СОВМЕСТИМЫЕ АТОМАРНЫЕ ОБЁРТКИ (Go 1.13+) +// ============================================================================= + +type atomicBool struct{ v int32 } + +func (b *atomicBool) Load() bool { return atomic.LoadInt32(&b.v) != 0 } +func (b *atomicBool) Store(v bool) { atomic.StoreInt32(&b.v, boolToInt32(v)) } +func (b *atomicBool) CAS(old, new bool) bool { + return atomic.CompareAndSwapInt32(&b.v, boolToInt32(old), boolToInt32(new)) +} + +func boolToInt32(b bool) int32 { + if b { + return 1 + } + return 0 +} + +type atomicUint64 struct{ v uint64 } + +func (a *atomicUint64) Load() uint64 { return atomic.LoadUint64(&a.v) } +func (a *atomicUint64) Store(v uint64) { atomic.StoreUint64(&a.v, v) } +func (a *atomicUint64) Add(d uint64) uint64 { return atomic.AddUint64(&a.v, d) } + // ============================================================================= // ТИПЫ ДЛЯ ДИНАМИЧЕСКОЙ КОНФИГУРАЦИИ // ============================================================================= @@ -42,9 +67,9 @@ type DynamicConfigChange struct { // DynamicConfigSnapshot представляет снэпшот конфигурации type DynamicConfigSnapshot struct { - Version uint64 `json:"version"` - UpdatedAt int64 `json:"updated_at"` - Config *Config `json:"config"` + Version uint64 `json:"version"` + UpdatedAt int64 `json:"updated_at"` + Config *Config `json:"config"` ChangeHistory []*DynamicConfigChange `json:"change_history"` } @@ -62,13 +87,14 @@ type DynamicConfigManager struct { raft *raft.Raft fsm *ConfigFSM mu sync.RWMutex - version atomic.Uint64 + version atomicUint64 changeHistory []*DynamicConfigChange listeners map[string]ConfigChangeListener listenerMu sync.RWMutex stopChan chan struct{} + stopOnce sync.Once wg sync.WaitGroup - isLeader atomic.Bool + isLeader atomicBool } // ConfigFSM реализует конечный автомат для конфигурации @@ -76,6 +102,10 @@ type ConfigFSM struct { config *DynamicConfigSnapshot mu sync.RWMutex logger *log.Logger + // onChange вызывается после успешного применения изменения. + // Может быть установлен менеджером для уведомления слушателей. + onChange func(change *DynamicConfigChange) + onChangeMu sync.RWMutex } // ConfigSnapshot реализует raft.FSMSnapshot @@ -91,9 +121,9 @@ func NewDynamicConfigManager(cfg *Config, logger *log.Logger) *DynamicConfigMana // Создаём начальный снэпшот initialSnapshot := &DynamicConfigSnapshot{ - Version: 1, - UpdatedAt: time.Now().UnixMilli(), - Config: cfg, + Version: 1, + UpdatedAt: time.Now().UnixMilli(), + Config: cfg, ChangeHistory: make([]*DynamicConfigChange, 0), } @@ -110,13 +140,31 @@ func NewDynamicConfigManager(cfg *Config, logger *log.Logger) *DynamicConfigMana listeners: make(map[string]ConfigChangeListener), stopChan: make(chan struct{}), } + dm.version.Store(1) + + // Подключаем FSM к менеджеру для уведомления слушателей + fsm.onChangeMu.Lock() + fsm.onChange = func(change *DynamicConfigChange) { + dm.version.Store(change.Version) + dm.mu.Lock() + dm.changeHistory = append(dm.changeHistory, change) + // Ограничиваем историю + if len(dm.changeHistory) > 1000 { + dm.changeHistory = dm.changeHistory[len(dm.changeHistory)-1000:] + } + dm.mu.Unlock() + dm.notifyListeners(change) + } + fsm.onChangeMu.Unlock() return dm } // SetRaft устанавливает Raft для управления конфигурацией -func (dm *DynamicConfigManager) SetRaft(raft *raft.Raft) { - dm.raft = raft +func (dm *DynamicConfigManager) SetRaft(r *raft.Raft) { + dm.mu.Lock() + defer dm.mu.Unlock() + dm.raft = r } // GetFSM возвращает FSM для конфигурации @@ -131,26 +179,38 @@ func (dm *DynamicConfigManager) Start() { } } -// Stop останавливает менеджер динамической конфигурации +// Stop останавливает менеджер динамической конфигурации. +// Безопасен для многократного вызова. func (dm *DynamicConfigManager) Stop() { - close(dm.stopChan) + dm.stopOnce.Do(func() { + close(dm.stopChan) + }) dm.wg.Wait() if dm.logger != nil { dm.logger.Info("Dynamic config manager stopped") } } -// GetConfig возвращает текущую конфигурацию +// GetConfig возвращает текущую конфигурацию. +// Возвращает копию, чтобы вызывающий не мог случайно мутировать состояние FSM. func (dm *DynamicConfigManager) GetConfig() *Config { dm.fsm.mu.RLock() defer dm.fsm.mu.RUnlock() - // Возвращаем копию конфигурации + if dm.fsm.config == nil || dm.fsm.config.Config == nil { + return nil + } return deepCopyConfig(dm.fsm.config.Config) } // GetVersion возвращает текущую версию конфигурации func (dm *DynamicConfigManager) GetVersion() uint64 { + // Синхронизируем с FSM, чтобы значение было актуальным + dm.fsm.mu.RLock() + defer dm.fsm.mu.RUnlock() + if dm.fsm.config != nil { + return dm.fsm.config.Version + } return dm.version.Load() } @@ -166,18 +226,25 @@ func (dm *DynamicConfigManager) GetChangeHistory() []*DynamicConfigChange { // ApplyChange применяет изменение конфигурации через Raft func (dm *DynamicConfigManager) ApplyChange(changes map[string]interface{}, description string, changedBy string) error { - if dm.raft == nil { + dm.mu.RLock() + r := dm.raft + dm.mu.RUnlock() + + if r == nil { return fmt.Errorf("raft not initialized") } - if !dm.isLeader.Load() { - return fmt.Errorf("not the leader") + // Проверяем, что мы лидер — через сам Raft, а не через внешний флаг + if r.State() != raft.Leader { + return fmt.Errorf("not the leader (current state: %s)", r.State()) } + currentVersion := dm.GetVersion() + // Создаём команду изменения change := &DynamicConfigChange{ - ID: fmt.Sprintf("config_%d_%d", dm.version.Load()+1, time.Now().UnixNano()), - Version: dm.version.Load() + 1, + ID: fmt.Sprintf("config_%d_%d", currentVersion+1, time.Now().UnixNano()), + Version: currentVersion + 1, Timestamp: time.Now().UnixMilli(), ChangedBy: changedBy, Description: description, @@ -190,11 +257,18 @@ func (dm *DynamicConfigManager) ApplyChange(changes map[string]interface{}, desc return fmt.Errorf("failed to marshal config change: %v", err) } - future := dm.raft.Apply(data, 10*time.Second) + future := r.Apply(data, 10*time.Second) if err := future.Error(); err != nil { return fmt.Errorf("raft apply failed: %v", err) } + // Проверяем ошибку из FSM.Apply + if resp := future.Response(); resp != nil { + if err, ok := resp.(error); ok { + return fmt.Errorf("fsm apply failed: %v", err) + } + } + if dm.logger != nil { dm.logger.Info(fmt.Sprintf("Applied config change: %s (version: %d)", description, change.Version)) } @@ -209,15 +283,30 @@ func (dm *DynamicConfigManager) UpdateConfig(updates map[string]interface{}, des // ApplyConfigChange применяет изменение к FSM (вызывается из Raft) func (fsm *ConfigFSM) ApplyConfigChange(change *DynamicConfigChange) error { - fsm.mu.Lock() - defer fsm.mu.Unlock() + if change == nil { + return fmt.Errorf("nil config change") + } + + fsm.mu.Lock() + + if fsm.config == nil { + fsm.config = &DynamicConfigSnapshot{ + Version: 1, + UpdatedAt: time.Now().UnixMilli(), + Config: &Config{}, + ChangeHistory: make([]*DynamicConfigChange, 0), + } + } + if fsm.config.Config == nil { + fsm.config.Config = &Config{} + } - // Применяем изменения к конфигурации config := fsm.config.Config - // Рекурсивно обновляем поля + // Применяем изменения к конфигурации for path, value := range change.Changes { if err := setConfigField(config, path, value); err != nil { + fsm.mu.Unlock() return err } } @@ -226,11 +315,25 @@ func (fsm *ConfigFSM) ApplyConfigChange(change *DynamicConfigChange) error { fsm.config.Version = change.Version fsm.config.UpdatedAt = time.Now().UnixMilli() fsm.config.ChangeHistory = append(fsm.config.ChangeHistory, change) + // Ограничиваем историю + if len(fsm.config.ChangeHistory) > 1000 { + fsm.config.ChangeHistory = fsm.config.ChangeHistory[len(fsm.config.ChangeHistory)-1000:] + } + + fsm.mu.Unlock() if fsm.logger != nil { fsm.logger.Debug(fmt.Sprintf("Applied config change to FSM: %s", change.Description)) } + // Уведомляем слушателей (через callback, установленный менеджером) + fsm.onChangeMu.RLock() + cb := fsm.onChange + fsm.onChangeMu.RUnlock() + if cb != nil { + cb(change) + } + return nil } @@ -253,19 +356,32 @@ 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)), - }, + if fsm.config == nil { + return &ConfigSnapshot{config: &DynamicConfigSnapshot{ + Version: 1, + UpdatedAt: time.Now().UnixMilli(), + Config: &Config{}, + ChangeHistory: make([]*DynamicConfigChange, 0), + }}, nil } - copy(snapshot.config.ChangeHistory, fsm.config.ChangeHistory) + // Глубокая копия через JSON, чтобы снэпшот не зависел от мутаций FSM + data, err := json.Marshal(fsm.config) + if err != nil { + return nil, fmt.Errorf("failed to marshal snapshot: %v", err) + } + var snapshotData DynamicConfigSnapshot + if err := json.Unmarshal(data, &snapshotData); err != nil { + return nil, fmt.Errorf("failed to unmarshal snapshot: %v", err) + } + if snapshotData.ChangeHistory == nil { + snapshotData.ChangeHistory = make([]*DynamicConfigChange, 0) + } + if snapshotData.Config == nil { + snapshotData.Config = &Config{} + } - return snapshot, nil + return &ConfigSnapshot{config: &snapshotData}, nil } // Restore восстанавливает состояние FSM из снэпшота @@ -275,13 +391,22 @@ func (fsm *ConfigFSM) Restore(snapshot io.ReadCloser) error { var data DynamicConfigSnapshot decoder := json.NewDecoder(snapshot) if err := decoder.Decode(&data); err != nil { - return err + return fmt.Errorf("failed to decode snapshot: %v", err) + } + + if data.Config == nil { + data.Config = &Config{} + } + if data.ChangeHistory == nil { + data.ChangeHistory = make([]*DynamicConfigChange, 0) + } + if data.Version == 0 { + data.Version = 1 } fsm.mu.Lock() - defer fsm.mu.Unlock() - fsm.config = &data + fsm.mu.Unlock() return nil } @@ -319,13 +444,22 @@ func (dm *DynamicConfigManager) UnregisterListener(name string) { delete(dm.listeners, name) } -// notifyListeners уведомляет слушателей об изменении +// notifyListeners уведомляет слушателей об изменении. +// Каждый слушатель вызывается в отдельной горутине; паника в слушателе +// не должна ронять процесс. func (dm *DynamicConfigManager) notifyListeners(change *DynamicConfigChange) { dm.listenerMu.RLock() defer dm.listenerMu.RUnlock() - for _, listener := range dm.listeners { - go listener(change) + for name, listener := range dm.listeners { + go func(n string, l ConfigChangeListener) { + defer func() { + if r := recover(); r != nil && dm.logger != nil { + dm.logger.Error(fmt.Sprintf("panic in config listener %s: %v", n, r)) + } + }() + l(change) + }(name, listener) } } @@ -333,260 +467,645 @@ func (dm *DynamicConfigManager) notifyListeners(change *DynamicConfigChange) { // ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ // ============================================================================= -// deepCopyConfig создаёт глубокую копию конфигурации +// deepCopyConfig создаёт глубокую копию конфигурации через JSON. +// Если src == nil, возвращает &Config{} (не nil), чтобы избежать паник. func deepCopyConfig(src *Config) *Config { if src == nil { - return nil + return &Config{} } - // Используем JSON для глубокого копирования data, err := json.Marshal(src) if err != nil { - return nil + // Возвращаем пустую копию, а не nil, чтобы не сломать вызывающий код + return &Config{} } var dst Config if err := json.Unmarshal(data, &dst); err != nil { - return nil + return &Config{} } return &dst } -// setConfigField устанавливает значение поля конфигурации по пути +// coerceInt пытается привести значение к int. +// Поддерживает int, int32, int64, uint, uint32, uint64, float32, float64, json.Number. +func coerceInt(value interface{}) (int, bool) { + switch v := value.(type) { + case int: + return v, true + case int8: + return int(v), true + case int16: + return int(v), true + case int32: + return int(v), true + case int64: + return int(v), true + case uint: + return int(v), true + case uint8: + return int(v), true + case uint16: + return int(v), true + case uint32: + return int(v), true + case uint64: + return int(v), true + case float32: + return int(v), true + case float64: + return int(v), true + case json.Number: + if i, err := v.Int64(); err == nil { + return int(i), true + } + if f, err := v.Float64(); err == nil { + return int(f), true + } + } + return 0, false +} + +// coerceInt64 пытается привести значение к int64. +func coerceInt64(value interface{}) (int64, bool) { + switch v := value.(type) { + case int: + return int64(v), true + case int8: + return int64(v), true + case int16: + return int64(v), true + case int32: + return int64(v), true + case int64: + return v, true + case uint: + return int64(v), true + case uint8: + return int64(v), true + case uint16: + return int64(v), true + case uint32: + return int64(v), true + case uint64: + return int64(v), true + case float32: + return int64(v), true + case float64: + return int64(v), true + case json.Number: + if i, err := v.Int64(); err == nil { + return i, true + } + if f, err := v.Float64(); err == nil { + return int64(f), true + } + } + return 0, false +} + +// coerceFloat64 пытается привести значение к float64. +func coerceFloat64(value interface{}) (float64, bool) { + switch v := value.(type) { + case float64: + return v, true + case float32: + return float64(v), true + case int: + return float64(v), true + case int8: + return float64(v), true + case int16: + return float64(v), true + case int32: + return float64(v), true + case int64: + return float64(v), true + case uint: + return float64(v), true + case uint8: + return float64(v), true + case uint16: + return float64(v), true + case uint32: + return float64(v), true + case uint64: + return float64(v), true + case json.Number: + if f, err := v.Float64(); err == nil { + return f, true + } + } + return 0, false +} + +// coerceBool пытается привести значение к bool. +func coerceBool(value interface{}) (bool, bool) { + switch v := value.(type) { + case bool: + return v, true + case string: + s := strings.ToLower(strings.TrimSpace(v)) + return s == "true" || s == "1" || s == "yes" || s == "on", true + case int: + return v != 0, true + case int64: + return v != 0, true + case float64: + return v != 0, true + } + return false, false +} + +// coerceString пытается привести значение к string. +func coerceString(value interface{}) (string, bool) { + if s, ok := value.(string); ok { + return s, true + } + return "", false +} + +// setConfigField устанавливает значение поля конфигурации по пути. +// Поддерживает основные секции: cluster, storage, replication, wal, mvcc, +// saga, backpressure, api, log, plugins, autoscaling, monitoring, transactions. func setConfigField(config *Config, path string, value interface{}) error { - // Разбиваем путь на части - parts := strings.Split(path, ".") - if len(parts) == 0 { - return fmt.Errorf("empty path") + if path == "" { + return fmt.Errorf("empty config path") } - // Простая реализация для основных путей - // В реальной production-версии следует использовать рефлексию switch path { + // ===== cluster ===== case "cluster.name": - if v, ok := value.(string); ok { + if v, ok := coerceString(value); ok { config.Cluster.Name = v + return nil } + return fmt.Errorf("cluster.name: expected string, got %T", value) + case "cluster.node_ip": + if v, ok := coerceString(value); ok { + config.Cluster.NodeIP = v + return nil + } + return fmt.Errorf("cluster.node_ip: expected string, got %T", value) + case "cluster.node_port": + if v, ok := coerceInt(value); ok { + config.Cluster.NodePort = v + return nil + } + return fmt.Errorf("cluster.node_port: expected int, got %T", value) + case "cluster.raft_port": + if v, ok := coerceInt(value); ok { + config.Cluster.RaftPort = v + return nil + } + return fmt.Errorf("cluster.raft_port: expected int, got %T", value) case "cluster.heartbeat_timeout_ms": - if v, ok := value.(int); ok { + if v, ok := coerceInt(value); 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) + return nil } + return fmt.Errorf("cluster.heartbeat_timeout_ms: expected int, got %T", value) case "cluster.election_timeout_ms": - if v, ok := value.(int); ok { + if v, ok := coerceInt(value); 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) + return nil } + return fmt.Errorf("cluster.election_timeout_ms: expected int, got %T", value) case "cluster.commit_timeout_ms": - if v, ok := value.(int); ok { + if v, ok := coerceInt(value); 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) + return nil } + return fmt.Errorf("cluster.commit_timeout_ms: expected int, got %T", value) case "cluster.snapshot_interval_min": - if v, ok := value.(int); ok { + if v, ok := coerceInt(value); 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) + return nil } + return fmt.Errorf("cluster.snapshot_interval_min: expected int, got %T", value) case "cluster.snapshot_threshold": - if v, ok := value.(int); ok { + if v, ok := coerceInt(value); 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) + return nil } + return fmt.Errorf("cluster.snapshot_threshold: expected int, got %T", value) case "cluster.recovery_timeout_sec": - if v, ok := value.(int); ok { + if v, ok := coerceInt(value); 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) + return nil } + return fmt.Errorf("cluster.recovery_timeout_sec: expected int, got %T", value) case "cluster.split_brain_prevention": - if v, ok := value.(bool); ok { + if v, ok := coerceBool(value); ok { config.Cluster.SplitBrainPrevention = v + return nil } + return fmt.Errorf("cluster.split_brain_prevention: expected bool, got %T", value) + case "cluster.region": + if v, ok := coerceString(value); ok { + config.Cluster.Region = v + return nil + } + return fmt.Errorf("cluster.region: expected string, got %T", value) + case "cluster.priority_zone": + if v, ok := coerceInt(value); ok { + config.Cluster.PriorityZone = v + return nil + } + return fmt.Errorf("cluster.priority_zone: expected int, got %T", value) + + // ===== storage ===== case "storage.page_size_mb": - if v, ok := value.(int); ok { + if v, ok := coerceInt(value); 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) + return nil } + return fmt.Errorf("storage.page_size_mb: expected int, got %T", value) case "storage.max_collections": - if v, ok := value.(int); ok { + if v, ok := coerceInt(value); 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) + return nil } + return fmt.Errorf("storage.max_collections: expected int, got %T", value) case "storage.max_documents_per_collection": - if v, ok := value.(int); ok { + if v, ok := coerceInt(value); 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) + return nil } + return fmt.Errorf("storage.max_documents_per_collection: expected int, got %T", value) + case "storage.default_engine": + if v, ok := coerceString(value); ok { + config.Storage.DefaultEngine = v + return nil + } + return fmt.Errorf("storage.default_engine: expected string, got %T", value) + case "storage.enable_custom_engines": + if v, ok := coerceBool(value); ok { + config.Storage.EnableCustomEngines = v + return nil + } + return fmt.Errorf("storage.enable_custom_engines: expected bool, got %T", value) + + // ===== replication ===== case "replication.enabled": - if v, ok := value.(bool); ok { + if v, ok := coerceBool(value); ok { config.Replication.Enabled = v + return nil } + return fmt.Errorf("replication.enabled: expected bool, got %T", value) case "replication.sync_replication": - if v, ok := value.(bool); ok { + if v, ok := coerceBool(value); ok { config.Replication.SyncReplication = v + return nil } + return fmt.Errorf("replication.sync_replication: expected bool, got %T", value) case "replication.replication_timeout_ms": - if v, ok := value.(int); ok { + if v, ok := coerceInt(value); 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) + return nil } + return fmt.Errorf("replication.replication_timeout_ms: expected int, got %T", value) case "replication.max_replica_lag_ms": - if v, ok := value.(int); ok { + if v, ok := coerceInt(value); 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) + return nil } + return fmt.Errorf("replication.max_replica_lag_ms: expected int, got %T", value) + + // ===== wal ===== case "wal.segment_size_mb": - if v, ok := value.(int); ok { + if v, ok := coerceInt(value); 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) + return nil } + return fmt.Errorf("wal.segment_size_mb: expected int, got %T", value) case "wal.sync_interval_sec": - if v, ok := value.(int); ok { + if v, ok := coerceInt(value); 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) + return nil } + return fmt.Errorf("wal.sync_interval_sec: expected int, got %T", value) case "wal.batch_size": - if v, ok := value.(int); ok { + if v, ok := coerceInt(value); 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) + return nil } + return fmt.Errorf("wal.batch_size: expected int, got %T", value) case "wal.enabled": - if v, ok := value.(bool); ok { + if v, ok := coerceBool(value); ok { config.WAL.Enabled = v + return nil } + return fmt.Errorf("wal.enabled: expected bool, got %T", value) case "wal.async_recovery": - if v, ok := value.(bool); ok { + if v, ok := coerceBool(value); ok { config.WAL.AsyncRecovery = v + return nil } + return fmt.Errorf("wal.async_recovery: expected bool, got %T", value) + case "wal.recovery_workers": + if v, ok := coerceInt(value); ok { + config.WAL.RecoveryWorkers = v + return nil + } + return fmt.Errorf("wal.recovery_workers: expected int, got %T", value) + case "wal.async_recovery_workers": + if v, ok := coerceInt(value); ok { + config.WAL.AsyncRecoveryWorkers = v + return nil + } + return fmt.Errorf("wal.async_recovery_workers: expected int, got %T", value) + case "wal.async_recovery_buffer": + if v, ok := coerceInt(value); ok { + config.WAL.AsyncRecoveryBuffer = v + return nil + } + return fmt.Errorf("wal.async_recovery_buffer: expected int, got %T", value) + + // ===== mvcc ===== case "mvcc.max_versions_per_doc": - if v, ok := value.(int); ok { + if v, ok := coerceInt(value); 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) + return nil } + return fmt.Errorf("mvcc.max_versions_per_doc: expected int, got %T", value) + case "mvcc.visibility_map_size": + if v, ok := coerceInt(value); ok { + config.MVCC.VisibilityMapSize = v + return nil + } + return fmt.Errorf("mvcc.visibility_map_size: expected int, got %T", value) + case "mvcc.prune_interval_min": + if v, ok := coerceInt(value); ok { + config.MVCC.PruneIntervalMin = v + return nil + } + return fmt.Errorf("mvcc.prune_interval_min: expected int, got %T", value) case "mvcc.retention_days": - if v, ok := value.(int); ok { + if v, ok := coerceInt(value); 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) + return nil } + return fmt.Errorf("mvcc.retention_days: expected int, got %T", value) + case "mvcc.read_cache_size": + if v, ok := coerceInt(value); ok { + config.MVCC.ReadCacheSize = v + return nil + } + return fmt.Errorf("mvcc.read_cache_size: expected int, got %T", value) + case "mvcc.read_cache_ttl_sec": + if v, ok := coerceInt(value); ok { + config.MVCC.ReadCacheTTLSec = v + return nil + } + return fmt.Errorf("mvcc.read_cache_ttl_sec: expected int, got %T", value) + + // ===== saga ===== case "saga.enabled": - if v, ok := value.(bool); ok { + if v, ok := coerceBool(value); ok { config.Saga.Enabled = v + return nil } + return fmt.Errorf("saga.enabled: expected bool, got %T", value) case "saga.coordinator_count": - if v, ok := value.(int); ok { + if v, ok := coerceInt(value); 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) + return nil } + return fmt.Errorf("saga.coordinator_count: expected int, got %T", value) case "saga.max_retries": - if v, ok := value.(int); ok { + if v, ok := coerceInt(value); 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) + return nil } + return fmt.Errorf("saga.max_retries: expected int, got %T", value) case "saga.saga_timeout_sec": - if v, ok := value.(int); ok { + if v, ok := coerceInt(value); 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) + return nil } + return fmt.Errorf("saga.saga_timeout_sec: expected int, got %T", value) + case "saga.retry_backoff_ms": + if v, ok := coerceInt(value); ok { + config.Saga.RetryBackoffMs = v + return nil + } + return fmt.Errorf("saga.retry_backoff_ms: expected int, got %T", value) + case "saga.stuck_check_interval_sec": + if v, ok := coerceInt(value); ok { + config.Saga.StuckCheckIntervalSec = v + return nil + } + return fmt.Errorf("saga.stuck_check_interval_sec: expected int, got %T", value) + + // ===== backpressure ===== case "backpressure.enabled": - if v, ok := value.(bool); ok { + if v, ok := coerceBool(value); ok { config.Backpressure.Enabled = v + return nil } + return fmt.Errorf("backpressure.enabled: expected bool, got %T", value) case "backpressure.cpu_threshold": - if v, ok := value.(float64); ok { + if v, ok := coerceFloat64(value); ok { config.Backpressure.CPUThreshold = v + return nil } + return fmt.Errorf("backpressure.cpu_threshold: expected float64, got %T", value) case "backpressure.memory_threshold": - if v, ok := value.(float64); ok { + if v, ok := coerceFloat64(value); ok { config.Backpressure.MemoryThreshold = v + return nil } + return fmt.Errorf("backpressure.memory_threshold: expected float64, got %T", value) + case "backpressure.queue_size_threshold": + if v, ok := coerceInt(value); ok { + config.Backpressure.QueueSizeThreshold = v + return nil + } + return fmt.Errorf("backpressure.queue_size_threshold: expected int, got %T", value) + case "backpressure.connection_threshold": + if v, ok := coerceInt(value); ok { + config.Backpressure.ConnectionThreshold = v + return nil + } + return fmt.Errorf("backpressure.connection_threshold: expected int, got %T", value) + + // ===== api ===== case "api.port": - if v, ok := value.(int); ok { + if v, ok := coerceInt(value); 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) + return nil } + return fmt.Errorf("api.port: expected int, got %T", value) + + // ===== log ===== case "log.log_level": - if v, ok := value.(string); ok { + if v, ok := coerceString(value); ok { config.Log.LogLevel = v + return nil } + return fmt.Errorf("log.log_level: expected string, got %T", value) case "log.log_file": - if v, ok := value.(string); ok { + if v, ok := coerceString(value); ok { config.Log.LogFile = v + return nil } + return fmt.Errorf("log.log_file: expected string, got %T", value) + + // ===== plugins ===== + case "plugins.enabled": + if v, ok := coerceBool(value); ok { + config.Plugins.Enabled = v + return nil + } + return fmt.Errorf("plugins.enabled: expected bool, got %T", value) + case "plugins.max_cpu_time_ms": + if v, ok := coerceInt(value); ok { + config.Plugins.MaxCPUTimeMs = v + return nil + } + return fmt.Errorf("plugins.max_cpu_time_ms: expected int, got %T", value) + case "plugins.max_memory_mb": + if v, ok := coerceInt(value); ok { + config.Plugins.MaxMemoryMB = v + return nil + } + return fmt.Errorf("plugins.max_memory_mb: expected int, got %T", value) + case "plugins.hot_reload_interval_sec": + if v, ok := coerceInt(value); ok { + config.Plugins.HotReloadIntervalSec = v + return nil + } + return fmt.Errorf("plugins.hot_reload_interval_sec: expected int, got %T", value) + case "plugins.max_lua_states": + if v, ok := coerceInt(value); ok { + config.Plugins.MaxLuaStates = v + return nil + } + return fmt.Errorf("plugins.max_lua_states: expected int, got %T", value) + + // ===== autoscaling ===== + case "autoscaling.enabled": + if v, ok := coerceBool(value); ok { + config.Autoscaling.Enabled = v + return nil + } + return fmt.Errorf("autoscaling.enabled: expected bool, got %T", value) + case "autoscaling.min_nodes": + if v, ok := coerceInt(value); ok { + config.Autoscaling.MinNodes = v + return nil + } + return fmt.Errorf("autoscaling.min_nodes: expected int, got %T", value) + case "autoscaling.max_nodes": + if v, ok := coerceInt(value); ok { + config.Autoscaling.MaxNodes = v + return nil + } + return fmt.Errorf("autoscaling.max_nodes: expected int, got %T", value) + case "autoscaling.scale_up_threshold": + if v, ok := coerceFloat64(value); ok { + config.Autoscaling.ScaleUpThreshold = v + return nil + } + return fmt.Errorf("autoscaling.scale_up_threshold: expected float64, got %T", value) + case "autoscaling.scale_down_threshold": + if v, ok := coerceFloat64(value); ok { + config.Autoscaling.ScaleDownThreshold = v + return nil + } + return fmt.Errorf("autoscaling.scale_down_threshold: expected float64, got %T", value) + case "autoscaling.evaluation_interval_sec": + if v, ok := coerceInt(value); ok { + config.Autoscaling.EvaluationIntervalSec = v + return nil + } + return fmt.Errorf("autoscaling.evaluation_interval_sec: expected int, got %T", value) + case "autoscaling.scale_up_cooldown_sec": + if v, ok := coerceInt(value); ok { + config.Autoscaling.ScaleUpCooldownSec = v + return nil + } + return fmt.Errorf("autoscaling.scale_up_cooldown_sec: expected int, got %T", value) + case "autoscaling.scale_down_cooldown_sec": + if v, ok := coerceInt(value); ok { + config.Autoscaling.ScaleDownCooldownSec = v + return nil + } + return fmt.Errorf("autoscaling.scale_down_cooldown_sec: expected int, got %T", value) + + // ===== monitoring ===== + case "monitoring.enable_metrics": + if v, ok := coerceBool(value); ok { + config.Monitoring.EnableMetrics = v + return nil + } + return fmt.Errorf("monitoring.enable_metrics: expected bool, got %T", value) + case "monitoring.metrics_port": + if v, ok := coerceInt(value); ok { + config.Monitoring.MetricsPort = v + return nil + } + return fmt.Errorf("monitoring.metrics_port: expected int, got %T", value) + case "monitoring.enable_tracing": + if v, ok := coerceBool(value); ok { + config.Monitoring.EnableTracing = v + return nil + } + return fmt.Errorf("monitoring.enable_tracing: expected bool, got %T", value) + case "monitoring.trace_sample_rate": + if v, ok := coerceFloat64(value); ok { + config.Monitoring.TraceSampleRate = v + return nil + } + return fmt.Errorf("monitoring.trace_sample_rate: expected float64, got %T", value) + + // ===== transactions ===== + case "transactions.default_timeout_sec": + if v, ok := coerceInt(value); ok { + config.Transactions.DefaultTimeoutSec = v + return nil + } + return fmt.Errorf("transactions.default_timeout_sec: expected int, got %T", value) + case "transactions.enabled": + if v, ok := coerceBool(value); ok { + config.Transactions.Enabled = v + return nil + } + return fmt.Errorf("transactions.enabled: expected bool, got %T", value) + case "transactions.deadlock_check_interval_sec": + if v, ok := coerceInt(value); ok { + config.Transactions.DeadlockCheckIntervalSec = v + return nil + } + return fmt.Errorf("transactions.deadlock_check_interval_sec: expected int, got %T", value) + case "transactions.max_savepoints_per_tx": + if v, ok := coerceInt(value); ok { + config.Transactions.MaxSavepointsPerTx = v + return nil + } + return fmt.Errorf("transactions.max_savepoints_per_tx: expected int, got %T", value) + default: - // Для неизвестных путей логируем предупреждение - // В production-версии следует использовать рефлексию для полного покрытия return fmt.Errorf("unknown config path: %s", path) } - - return nil } // GetLeaderStatus возвращает статус лидера func (dm *DynamicConfigManager) GetLeaderStatus() bool { + dm.mu.RLock() + r := dm.raft + dm.mu.RUnlock() + + if r != nil { + return r.State() == raft.Leader + } return dm.isLeader.Load() } -// SetLeaderStatus устанавливает статус лидера +// SetLeaderStatus устанавливает статус лидера. +// Оставлено для совместимости; предпочтительнее использовать Raft.State(). func (dm *DynamicConfigManager) SetLeaderStatus(isLeader bool) { dm.isLeader.Store(isLeader) } @@ -601,4 +1120,4 @@ func (dm *DynamicConfigManager) GetListeners() []string { names = append(names, name) } return names -} +} \ No newline at end of file