1146 lines
34 KiB
Go
1146 lines
34 KiB
Go
/*
|
|
* 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"
|
|
)
|
|
|
|
// =============================================================================
|
|
// СОВМЕСТИМЫЕ АТОМАРНЫЕ ОБЁРТКИ (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) }
|
|
|
|
// =============================================================================
|
|
// ТИПЫ ДЛЯ ДИНАМИЧЕСКОЙ КОНФИГУРАЦИИ
|
|
// =============================================================================
|
|
|
|
// 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 atomicUint64
|
|
changeHistory []*DynamicConfigChange
|
|
listeners map[string]ConfigChangeListener
|
|
listenerMu sync.RWMutex
|
|
stopChan chan struct{}
|
|
stopOnce sync.Once
|
|
wg sync.WaitGroup
|
|
isLeader atomicBool
|
|
}
|
|
|
|
// ConfigFSM реализует конечный автомат для конфигурации.
|
|
type ConfigFSM struct {
|
|
config *DynamicConfigSnapshot
|
|
mu sync.RWMutex
|
|
logger *log.Logger
|
|
// onChange вызывается после успешного применения изменения.
|
|
// Может быть установлен менеджером для уведомления слушателей.
|
|
onChange func(change *DynamicConfigChange)
|
|
onChangeMu sync.RWMutex
|
|
}
|
|
|
|
// 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{}),
|
|
}
|
|
dm.version.Store(1)
|
|
|
|
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(r *raft.Raft) {
|
|
dm.mu.Lock()
|
|
defer dm.mu.Unlock()
|
|
dm.raft = r
|
|
}
|
|
|
|
// 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() {
|
|
dm.stopOnce.Do(func() {
|
|
close(dm.stopChan)
|
|
})
|
|
dm.wg.Wait()
|
|
if dm.logger != nil {
|
|
dm.logger.Info("Dynamic config manager stopped")
|
|
}
|
|
}
|
|
|
|
// 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 {
|
|
dm.fsm.mu.RLock()
|
|
defer dm.fsm.mu.RUnlock()
|
|
if dm.fsm.config != nil {
|
|
return dm.fsm.config.Version
|
|
}
|
|
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 {
|
|
dm.mu.RLock()
|
|
r := dm.raft
|
|
dm.mu.RUnlock()
|
|
|
|
if r == nil {
|
|
return fmt.Errorf("raft not initialized")
|
|
}
|
|
|
|
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", currentVersion+1, time.Now().UnixNano()),
|
|
Version: currentVersion + 1,
|
|
Timestamp: time.Now().UnixMilli(),
|
|
ChangedBy: changedBy,
|
|
Description: description,
|
|
Changes: changes,
|
|
}
|
|
|
|
data, err := json.Marshal(change)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to marshal config change: %v", err)
|
|
}
|
|
|
|
future := r.Apply(data, 10*time.Second)
|
|
if err := future.Error(); err != nil {
|
|
return fmt.Errorf("raft apply failed: %v", err)
|
|
}
|
|
|
|
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))
|
|
}
|
|
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 {
|
|
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
|
|
}
|
|
}
|
|
|
|
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))
|
|
}
|
|
|
|
fsm.onChangeMu.RLock()
|
|
cb := fsm.onChange
|
|
fsm.onChangeMu.RUnlock()
|
|
if cb != nil {
|
|
cb(change)
|
|
}
|
|
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()
|
|
|
|
if fsm.config == nil {
|
|
return &ConfigSnapshot{config: &DynamicConfigSnapshot{
|
|
Version: 1,
|
|
UpdatedAt: time.Now().UnixMilli(),
|
|
Config: &Config{},
|
|
ChangeHistory: make([]*DynamicConfigChange, 0),
|
|
}}, nil
|
|
}
|
|
|
|
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 &ConfigSnapshot{config: &snapshotData}, 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 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()
|
|
fsm.config = &data
|
|
fsm.mu.Unlock()
|
|
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 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)
|
|
}
|
|
}
|
|
|
|
// =============================================================================
|
|
// ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ
|
|
// =============================================================================
|
|
|
|
// deepCopyConfig создаёт глубокую копию конфигурации через JSON.
|
|
func deepCopyConfig(src *Config) *Config {
|
|
if src == nil {
|
|
return &Config{}
|
|
}
|
|
data, err := json.Marshal(src)
|
|
if err != nil {
|
|
return &Config{}
|
|
}
|
|
var dst Config
|
|
if err := json.Unmarshal(data, &dst); err != nil {
|
|
return &Config{}
|
|
}
|
|
return &dst
|
|
}
|
|
|
|
// coerceInt пытается привести значение к int.
|
|
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 устанавливает значение поля конфигурации по пути.
|
|
// Используется FSM при применении изменений из Raft.
|
|
func setConfigField(config *Config, path string, value interface{}) error {
|
|
if path == "" {
|
|
return fmt.Errorf("empty config path")
|
|
}
|
|
|
|
switch path {
|
|
// ===== cluster =====
|
|
case "cluster.name":
|
|
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 := coerceInt(value); ok {
|
|
config.Cluster.HeartbeatTimeoutMs = v
|
|
return nil
|
|
}
|
|
return fmt.Errorf("cluster.heartbeat_timeout_ms: expected int, got %T", value)
|
|
case "cluster.election_timeout_ms":
|
|
if v, ok := coerceInt(value); ok {
|
|
config.Cluster.ElectionTimeoutMs = v
|
|
return nil
|
|
}
|
|
return fmt.Errorf("cluster.election_timeout_ms: expected int, got %T", value)
|
|
case "cluster.commit_timeout_ms":
|
|
if v, ok := coerceInt(value); ok {
|
|
config.Cluster.CommitTimeoutMs = v
|
|
return nil
|
|
}
|
|
return fmt.Errorf("cluster.commit_timeout_ms: expected int, got %T", value)
|
|
case "cluster.snapshot_interval_min":
|
|
if v, ok := coerceInt(value); ok {
|
|
config.Cluster.SnapshotIntervalMin = v
|
|
return nil
|
|
}
|
|
return fmt.Errorf("cluster.snapshot_interval_min: expected int, got %T", value)
|
|
case "cluster.snapshot_threshold":
|
|
if v, ok := coerceInt(value); ok {
|
|
config.Cluster.SnapshotThreshold = v
|
|
return nil
|
|
}
|
|
return fmt.Errorf("cluster.snapshot_threshold: expected int, got %T", value)
|
|
case "cluster.recovery_timeout_sec":
|
|
if v, ok := coerceInt(value); ok {
|
|
config.Cluster.RecoveryTimeoutSec = v
|
|
return nil
|
|
}
|
|
return fmt.Errorf("cluster.recovery_timeout_sec: expected int, got %T", value)
|
|
case "cluster.split_brain_prevention":
|
|
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 := coerceInt(value); ok {
|
|
config.Storage.PageSizeMB = v
|
|
return nil
|
|
}
|
|
return fmt.Errorf("storage.page_size_mb: expected int, got %T", value)
|
|
case "storage.max_collections":
|
|
if v, ok := coerceInt(value); ok {
|
|
config.Storage.MaxCollections = v
|
|
return nil
|
|
}
|
|
return fmt.Errorf("storage.max_collections: expected int, got %T", value)
|
|
case "storage.max_documents_per_collection":
|
|
if v, ok := coerceInt(value); ok {
|
|
config.Storage.MaxDocumentsPerCollection = 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 := 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 := 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 := coerceInt(value); ok {
|
|
config.Replication.ReplicationTimeoutMs = v
|
|
return nil
|
|
}
|
|
return fmt.Errorf("replication.replication_timeout_ms: expected int, got %T", value)
|
|
case "replication.max_replica_lag_ms":
|
|
if v, ok := coerceInt(value); ok {
|
|
config.Replication.MaxReplicaLagMs = 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 := coerceInt(value); ok {
|
|
config.WAL.SegmentSizeMB = v
|
|
return nil
|
|
}
|
|
return fmt.Errorf("wal.segment_size_mb: expected int, got %T", value)
|
|
case "wal.sync_interval_sec":
|
|
if v, ok := coerceInt(value); ok {
|
|
config.WAL.SyncIntervalSec = v
|
|
return nil
|
|
}
|
|
return fmt.Errorf("wal.sync_interval_sec: expected int, got %T", value)
|
|
case "wal.batch_size":
|
|
if v, ok := coerceInt(value); ok {
|
|
config.WAL.BatchSize = v
|
|
return nil
|
|
}
|
|
return fmt.Errorf("wal.batch_size: expected int, got %T", value)
|
|
case "wal.enabled":
|
|
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 := 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 := coerceInt(value); ok {
|
|
config.MVCC.MaxVersionsPerDoc = 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 := coerceInt(value); ok {
|
|
config.MVCC.RetentionDays = 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 := 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 := coerceInt(value); ok {
|
|
config.Saga.CoordinatorCount = v
|
|
return nil
|
|
}
|
|
return fmt.Errorf("saga.coordinator_count: expected int, got %T", value)
|
|
case "saga.max_retries":
|
|
if v, ok := coerceInt(value); ok {
|
|
config.Saga.MaxRetries = v
|
|
return nil
|
|
}
|
|
return fmt.Errorf("saga.max_retries: expected int, got %T", value)
|
|
case "saga.saga_timeout_sec":
|
|
if v, ok := coerceInt(value); ok {
|
|
config.Saga.SagaTimeoutSec = 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 := 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 := 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 := 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 := coerceInt(value); ok {
|
|
config.API.Port = v
|
|
return nil
|
|
}
|
|
return fmt.Errorf("api.port: expected int, got %T", value)
|
|
|
|
// ===== log =====
|
|
case "log.log_level":
|
|
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 := 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)
|
|
|
|
// ===== compression =====
|
|
case "compression.enabled":
|
|
if v, ok := coerceBool(value); ok {
|
|
config.Compression.Enabled = v
|
|
return nil
|
|
}
|
|
return fmt.Errorf("compression.enabled: expected bool, got %T", value)
|
|
case "compression.algorithm":
|
|
if v, ok := coerceString(value); ok {
|
|
config.Compression.Algorithm = v
|
|
return nil
|
|
}
|
|
return fmt.Errorf("compression.algorithm: expected string, got %T", value)
|
|
case "compression.level":
|
|
if v, ok := coerceInt(value); ok {
|
|
config.Compression.Level = v
|
|
return nil
|
|
}
|
|
return fmt.Errorf("compression.level: expected int, got %T", value)
|
|
case "compression.min_size":
|
|
if v, ok := coerceInt(value); ok {
|
|
config.Compression.MinSize = v
|
|
return nil
|
|
}
|
|
return fmt.Errorf("compression.min_size: expected int, got %T", value)
|
|
|
|
// ===== performance =====
|
|
case "performance.enable_pipeline":
|
|
if v, ok := coerceBool(value); ok {
|
|
config.Performance.EnablePipeline = v
|
|
return nil
|
|
}
|
|
return fmt.Errorf("performance.enable_pipeline: expected bool, got %T", value)
|
|
case "performance.batch_size":
|
|
if v, ok := coerceInt(value); ok {
|
|
config.Performance.BatchSize = v
|
|
return nil
|
|
}
|
|
return fmt.Errorf("performance.batch_size: expected int, got %T", value)
|
|
case "performance.max_connections":
|
|
if v, ok := coerceInt(value); ok {
|
|
config.Performance.MaxConnections = v
|
|
return nil
|
|
}
|
|
return fmt.Errorf("performance.max_connections: expected int, got %T", value)
|
|
case "performance.read_from_follower":
|
|
if v, ok := coerceBool(value); ok {
|
|
config.Performance.ReadFromFollower = v
|
|
return nil
|
|
}
|
|
return fmt.Errorf("performance.read_from_follower: expected bool, got %T", value)
|
|
|
|
default:
|
|
return fmt.Errorf("unknown config path: %s", path)
|
|
}
|
|
}
|
|
|
|
// 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 устанавливает статус лидера.
|
|
// Оставлено для совместимости; предпочтительнее использовать Raft.State().
|
|
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
|
|
} |