From 9c6bda7de415b81da028bd48665bb9c19453e291 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: Fri, 31 Jul 2026 11:55:05 +0000 Subject: [PATCH] Upload files to "internal/cluster" --- internal/cluster/self_healing.go | 729 +++++++++++++++++++++++++++++++ 1 file changed, 729 insertions(+) create mode 100644 internal/cluster/self_healing.go diff --git a/internal/cluster/self_healing.go b/internal/cluster/self_healing.go new file mode 100644 index 0000000..c9d24f8 --- /dev/null +++ b/internal/cluster/self_healing.go @@ -0,0 +1,729 @@ +/* + * 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/cluster/self_healing.go +// Назначение: Механизмы самоисцеления для автоматического восстановления узлов + +package cluster + +import ( + "context" + "fmt" + "sync" + "sync/atomic" + "time" + + "futriis/internal/log" + "futriis/internal/storage" +) + +// ============================================================================= +// ТИПЫ ДЛЯ SELF-HEALING +// ============================================================================= + +// HealingAction представляет действие по восстановлению +type HealingAction int + +const ( + HealingActionNone HealingAction = iota + HealingActionRestartNode // Перезапуск узла + HealingActionRejoinCluster // Переподключение к кластеру + HealingActionResharding // Перебалансировка шардов + HealingActionDataRecovery // Восстановление данных + HealingActionStateRestore // Восстановление состояния из снэпшота +) + +// HealingState представляет состояние процесса восстановления +type HealingState string + +const ( + HealingStateIdle HealingState = "idle" + HealingStateDetecting HealingState = "detecting" + HealingStateAnalyzing HealingState = "analyzing" + HealingStateHealing HealingState = "healing" + HealingStateVerifying HealingState = "verifying" + HealingStateComplete HealingState = "complete" + HealingStateFailed HealingState = "failed" +) + +// HealingEvent представляет событие восстановления +type HealingEvent struct { + ID string `json:"id"` + NodeID string `json:"node_id"` + Action HealingAction `json:"action"` + State HealingState `json:"state"` + Timestamp int64 `json:"timestamp"` + Description string `json:"description"` + Error string `json:"error,omitempty"` + Data map[string]interface{} `json:"data,omitempty"` +} + +// HealingConfig содержит настройки самоисцеления +type HealingConfig struct { + // Включено ли самоисцеление + Enabled bool `json:"enabled"` + // Интервал проверки состояния (сек) + CheckIntervalSec int `json:"check_interval_sec"` + // Количество последовательных сбоев до активации + FailureThreshold int `json:"failure_threshold"` + // Задержка перед началом восстановления (сек) + HealingDelaySec int `json:"healing_delay_sec"` + // Максимальное время восстановления (сек) + HealingTimeoutSec int `json:"healing_timeout_sec"` + // Автоматический перезапуск узла + AutoRestart bool `json:"auto_restart"` + // Автоматическая перебалансировка + AutoResharding bool `json:"auto_resharding"` + // Автоматическое восстановление данных + AutoDataRecovery bool `json:"auto_data_recovery"` + // Максимальное количество одновременных восстановлений + MaxConcurrentHealings int `json:"max_concurrent_healings"` +} + +// DefaultHealingConfig возвращает конфигурацию по умолчанию +func DefaultHealingConfig() *HealingConfig { + return &HealingConfig{ + Enabled: true, + CheckIntervalSec: 10, + FailureThreshold: 3, + HealingDelaySec: 5, + HealingTimeoutSec: 300, + AutoRestart: true, + AutoResharding: true, + AutoDataRecovery: true, + MaxConcurrentHealings: 3, + } +} + +// ============================================================================= +// SELF-HEALING MANAGER +// ============================================================================= + +// SelfHealingManager управляет механизмами самоисцеления +type SelfHealingManager struct { + config *HealingConfig + logger *log.Logger + coordinator *RaftCoordinator + gossipManager *GossipManager + store *storage.Storage + healingEvents sync.Map // map[string]*HealingEvent + activeHealings atomic.Int32 + stopChan chan struct{} + wg sync.WaitGroup + mu sync.RWMutex + nodeFailures map[string]int + healingHistory []*HealingEvent + metrics *HealingMetrics + eventHandlers map[string]func(*HealingEvent) + handlerMu sync.RWMutex +} + +// HealingMetrics хранит метрики восстановления +type HealingMetrics struct { + TotalHealings atomic.Uint64 + SuccessfulHealings atomic.Uint64 + FailedHealings atomic.Uint64 + ReshardingsTriggered atomic.Uint64 + DataRecoveries atomic.Uint64 + RestartsTriggered atomic.Uint64 + AvgHealingDuration atomic.Int64 +} + +// NewSelfHealingManager создаёт новый менеджер самоисцеления +func NewSelfHealingManager( + config *HealingConfig, + coordinator *RaftCoordinator, + gossipManager *GossipManager, + store *storage.Storage, + logger *log.Logger, +) *SelfHealingManager { + if config == nil { + config = DefaultHealingConfig() + } + + shm := &SelfHealingManager{ + config: config, + logger: logger, + coordinator: coordinator, + gossipManager: gossipManager, + store: store, + stopChan: make(chan struct{}), + nodeFailures: make(map[string]int), + healingHistory: make([]*HealingEvent, 0), + metrics: &HealingMetrics{}, + eventHandlers: make(map[string]func(*HealingEvent)), + } + + return shm +} + +// Start запускает менеджер самоисцеления +func (shm *SelfHealingManager) Start() { + if !shm.config.Enabled { + shm.logger.Info("Self-healing is disabled") + return + } + + shm.wg.Add(2) + go shm.monitorLoop() + go shm.healingLoop() + + shm.logger.Info("Self-healing manager started") +} + +// Stop останавливает менеджер самоисцеления +func (shm *SelfHealingManager) Stop() { + close(shm.stopChan) + shm.wg.Wait() + shm.logger.Info("Self-healing manager stopped") +} + +// monitorLoop периодически проверяет состояние узлов +func (shm *SelfHealingManager) monitorLoop() { + defer shm.wg.Done() + + ticker := time.NewTicker(time.Duration(shm.config.CheckIntervalSec) * time.Second) + defer ticker.Stop() + + for { + select { + case <-shm.stopChan: + return + case <-ticker.C: + shm.monitorNodes() + } + } +} + +// monitorNodes проверяет состояние всех узлов в кластере +func (shm *SelfHealingManager) monitorNodes() { + // Получаем информацию о узлах из gossip + membership := shm.gossipManager.GetMembership() + + for _, node := range membership { + if node.ID == shm.coordinator.localNodeInfo.ID { + continue + } + + // Проверяем состояние узла + if !node.IsAlive && !node.IsSuspect { + shm.handleNodeFailure(node) + } + } + + // Проверяем также через координатор + nodes := shm.coordinator.GetAllNodes() + for _, node := range nodes { + if node.ID == shm.coordinator.localNodeInfo.ID { + continue + } + + // Проверяем, не пропал ли узел из gossip + if gossipNode := shm.gossipManager.GetNodeByID(node.ID); gossipNode == nil { + // Узел известен координатору, но не в gossip - возможно проблема + if shm.shouldHealNode(node.ID) { + go shm.initiateHealing(node.ID, "Node missing from gossip") + } + } + } +} + +// handleNodeFailure обрабатывает сбой узла +func (shm *SelfHealingManager) handleNodeFailure(node *NodeState) { + shm.mu.Lock() + defer shm.mu.Unlock() + + // Увеличиваем счётчик сбоев + shm.nodeFailures[node.ID]++ + + if shm.logger != nil { + shm.logger.Warn(fmt.Sprintf("Node %s failure detected (count: %d)", node.ID, shm.nodeFailures[node.ID])) + } + + // Проверяем, превышен ли порог + if shm.nodeFailures[node.ID] >= shm.config.FailureThreshold { + if shm.shouldHealNode(node.ID) { + go shm.initiateHealing(node.ID, fmt.Sprintf("Node failure threshold exceeded: %d", shm.nodeFailures[node.ID])) + } + } +} + +// shouldHealNode проверяет, нужно ли восстанавливать узел +func (shm *SelfHealingManager) shouldHealNode(nodeID string) bool { + // Проверяем, не идёт ли уже восстановление + if _, ok := shm.healingEvents.Load(nodeID); ok { + return false + } + + // Проверяем количество активных восстановлений + if shm.activeHealings.Load() >= int32(shm.config.MaxConcurrentHealings) { + return false + } + + return true +} + +// initiateHealing запускает процесс восстановления узла +func (shm *SelfHealingManager) initiateHealing(nodeID string, reason string) { + // Проверяем, не идёт ли уже восстановление + if _, ok := shm.healingEvents.Load(nodeID); ok { + return + } + + // Увеличиваем счётчик активных восстановлений + if !shm.activeHealings.CompareAndSwap(shm.activeHealings.Load(), shm.activeHealings.Load()+1) { + return + } + defer shm.activeHealings.Add(-1) + + // Создаём событие восстановления + eventID := fmt.Sprintf("healing_%s_%d", nodeID, time.Now().UnixNano()) + event := &HealingEvent{ + ID: eventID, + NodeID: nodeID, + Action: HealingActionNone, + State: HealingStateDetecting, + Timestamp: time.Now().UnixMilli(), + Description: reason, + Data: make(map[string]interface{}), + } + + shm.healingEvents.Store(nodeID, event) + shm.metrics.TotalHealings.Add(1) + + if shm.logger != nil { + shm.logger.Info(fmt.Sprintf("Initiating healing for node %s: %s", nodeID, reason)) + } + + // Выполняем восстановление + shm.executeHealing(nodeID, event) +} + +// executeHealing выполняет процесс восстановления +func (shm *SelfHealingManager) executeHealing(nodeID string, event *HealingEvent) { + // Создаём контекст с таймаутом для всего процесса восстановления + healingTimeout := time.Duration(shm.config.HealingTimeoutSec) * time.Second + ctx, cancel := context.WithTimeout(context.Background(), healingTimeout) + defer cancel() + + // Фаза 1: Анализ состояния + event.State = HealingStateAnalyzing + event.Timestamp = time.Now().UnixMilli() + shm.healingEvents.Store(nodeID, event) + + if shm.logger != nil { + shm.logger.Info(fmt.Sprintf("Analyzing node %s...", nodeID)) + } + + // Проверяем контекст + select { + case <-ctx.Done(): + event.State = HealingStateFailed + event.Error = "healing timeout during analysis" + shm.healingEvents.Store(nodeID, event) + shm.failHealing(nodeID) + return + default: + } + + // Проверяем, жив ли узел + nodeInfo := shm.coordinator.GetNodeByID(nodeID) + if nodeInfo == nil { + // Узел не найден - возможно, он уже удалён + event.State = HealingStateComplete + event.Description = "Node not found, assuming removed" + shm.healingEvents.Store(nodeID, event) + shm.completeHealing(nodeID) + return + } + + // Проверяем, не восстановился ли узел сам + if shm.isNodeAlive(nodeID) { + event.State = HealingStateComplete + event.Description = "Node recovered automatically" + shm.healingEvents.Store(nodeID, event) + shm.completeHealing(nodeID) + return + } + + // Проверяем контекст перед выполнением действия + select { + case <-ctx.Done(): + event.State = HealingStateFailed + event.Error = "healing timeout before action" + shm.healingEvents.Store(nodeID, event) + shm.failHealing(nodeID) + return + default: + } + + // Фаза 2: Выбор действия + action := shm.determineHealingAction(nodeID, nodeInfo) + event.Action = action + event.State = HealingStateHealing + event.Timestamp = time.Now().UnixMilli() + shm.healingEvents.Store(nodeID, event) + + if shm.logger != nil { + shm.logger.Info(fmt.Sprintf("Selected healing action %v for node %s", action, nodeID)) + } + + // Фаза 3: Выполнение действия с учётом контекста + var err error + done := make(chan struct{}) + go func() { + defer close(done) + switch action { + case HealingActionRestartNode: + err = shm.restartNode(nodeID, event) + case HealingActionRejoinCluster: + err = shm.rejoinCluster(nodeID, event) + case HealingActionResharding: + err = shm.triggerResharding(nodeID, event) + case HealingActionDataRecovery: + err = shm.recoverData(nodeID, event) + case HealingActionStateRestore: + err = shm.restoreState(nodeID, event) + default: + err = fmt.Errorf("no suitable healing action found") + } + }() + + select { + case <-ctx.Done(): + event.State = HealingStateFailed + event.Error = "healing timeout during action execution" + shm.healingEvents.Store(nodeID, event) + shm.failHealing(nodeID) + return + case <-done: + // Действие выполнено, продолжаем + } + + // Фаза 4: Проверка результата + event.State = HealingStateVerifying + event.Timestamp = time.Now().UnixMilli() + shm.healingEvents.Store(nodeID, event) + + if err != nil { + event.State = HealingStateFailed + event.Error = err.Error() + shm.healingEvents.Store(nodeID, event) + shm.failHealing(nodeID) + if shm.logger != nil { + shm.logger.Error(fmt.Sprintf("Healing for node %s failed: %v", nodeID, err)) + } + return + } + + // Проверяем контекст перед финальной проверкой + select { + case <-ctx.Done(): + event.State = HealingStateFailed + event.Error = "healing timeout during verification" + shm.healingEvents.Store(nodeID, event) + shm.failHealing(nodeID) + return + default: + } + + // Проверяем, восстановился ли узел + if shm.isNodeAlive(nodeID) { + event.State = HealingStateComplete + event.Description = fmt.Sprintf("Healing completed successfully with action %v", action) + shm.healingEvents.Store(nodeID, event) + shm.completeHealing(nodeID) + shm.metrics.SuccessfulHealings.Add(1) + + if shm.logger != nil { + shm.logger.Info(fmt.Sprintf("Healing for node %s completed successfully", nodeID)) + } + } else { + event.State = HealingStateFailed + event.Error = "Node did not recover after healing action" + shm.healingEvents.Store(nodeID, event) + shm.failHealing(nodeID) + shm.metrics.FailedHealings.Add(1) + } +} + +// determineHealingAction определяет действие для восстановления +func (shm *SelfHealingManager) determineHealingAction(nodeID string, nodeInfo *NodeInfo) HealingAction { + // Проверяем статус узла + switch nodeInfo.Status { + case "offline": + if shm.config.AutoRestart { + return HealingActionRestartNode + } + return HealingActionRejoinCluster + case "failed": + if shm.config.AutoDataRecovery { + return HealingActionDataRecovery + } + return HealingActionRestartNode + case "syncing": + return HealingActionStateRestore + default: + return HealingActionRejoinCluster + } +} + +// restartNode перезапускает узел +func (shm *SelfHealingManager) restartNode(nodeID string, event *HealingEvent) error { + shm.metrics.RestartsTriggered.Add(1) + + if shm.logger != nil { + shm.logger.Info(fmt.Sprintf("Attempting to restart node %s", nodeID)) + } + + // Получаем информацию об узле + nodeInfo := shm.coordinator.GetNodeByID(nodeID) + if nodeInfo == nil { + return fmt.Errorf("node %s not found", nodeID) + } + + // Пробуем переподключиться к узлу + address := fmt.Sprintf("%s:%d", nodeInfo.IP, nodeInfo.Port) + if err := shm.coordinator.UpdateNodeStatus(nodeID, StatusActive); err != nil { + return fmt.Errorf("failed to update node status: %v", err) + } + + event.Data["address"] = address + event.Data["method"] = "restart" + + return nil +} + +// rejoinCluster переподключает узел к кластеру +func (shm *SelfHealingManager) rejoinCluster(nodeID string, event *HealingEvent) error { + if shm.logger != nil { + shm.logger.Info(fmt.Sprintf("Attempting to rejoin node %s to cluster", nodeID)) + } + + // Пробуем восстановить соединение + if err := shm.coordinator.UpdateNodeStatus(nodeID, StatusActive); err != nil { + return fmt.Errorf("failed to update node status: %v", err) + } + + event.Data["method"] = "rejoin" + + return nil +} + +// triggerResharding запускает перебалансировку шардов +func (shm *SelfHealingManager) triggerResharding(nodeID string, event *HealingEvent) error { + if !shm.config.AutoResharding { + return fmt.Errorf("auto resharding is disabled") + } + + shm.metrics.ReshardingsTriggered.Add(1) + + if shm.logger != nil { + shm.logger.Info(fmt.Sprintf("Triggering resharding due to node failure: %s", nodeID)) + } + + // Запускаем перебалансировку через координатор + if err := shm.coordinator.TriggerResharding("self_healing_node_failure"); err != nil { + return fmt.Errorf("failed to trigger resharding: %v", err) + } + + event.Data["method"] = "resharding" + + return nil +} + +// recoverData восстанавливает данные узла +func (shm *SelfHealingManager) recoverData(nodeID string, event *HealingEvent) error { + if !shm.config.AutoDataRecovery { + return fmt.Errorf("auto data recovery is disabled") + } + + shm.metrics.DataRecoveries.Add(1) + + if shm.logger != nil { + shm.logger.Info(fmt.Sprintf("Initiating data recovery for node %s", nodeID)) + } + + // Проверяем целостность данных + if err := shm.verifyDataIntegrity(); err != nil { + return fmt.Errorf("data integrity check failed: %v", err) + } + + // Восстанавливаем данные из бэкапа если необходимо + // TODO: Реализовать восстановление из бэкапа + + event.Data["method"] = "data_recovery" + + return nil +} + +// restoreState восстанавливает состояние узла +func (shm *SelfHealingManager) restoreState(nodeID string, event *HealingEvent) error { + if shm.logger != nil { + shm.logger.Info(fmt.Sprintf("Restoring state for node %s", nodeID)) + } + + // Восстанавливаем состояние из снэпшота Raft + // TODO: Реализовать восстановление состояния + + event.Data["method"] = "state_restore" + + return nil +} + +// verifyDataIntegrity проверяет целостность данных +func (shm *SelfHealingManager) verifyDataIntegrity() error { + // Проверяем целостность хранилища + stats := shm.store.GetStats() + if stats == nil { + return fmt.Errorf("failed to get storage stats") + } + + // Проверяем, что данные не повреждены + // TODO: Реализовать полноценную проверку целостности + + return nil +} + +// isNodeAlive проверяет, жив ли узел +func (shm *SelfHealingManager) isNodeAlive(nodeID string) bool { + // Проверяем через gossip + gossipNode := shm.gossipManager.GetNodeByID(nodeID) + if gossipNode != nil && gossipNode.IsAlive { + return true + } + + // Проверяем через координатор + nodeInfo := shm.coordinator.GetNodeByID(nodeID) + if nodeInfo != nil && nodeInfo.Status == "active" { + return true + } + + return false +} + +// completeHealing завершает процесс восстановления +func (shm *SelfHealingManager) completeHealing(nodeID string) { + shm.mu.Lock() + defer shm.mu.Unlock() + + // Удаляем из списка сбоев + delete(shm.nodeFailures, nodeID) + + // Удаляем событие + if event, ok := shm.healingEvents.Load(nodeID); ok { + event.(*HealingEvent).State = HealingStateComplete + shm.healingHistory = append(shm.healingHistory, event.(*HealingEvent)) + shm.healingEvents.Delete(nodeID) + } +} + +// failHealing отмечает неудачное восстановление +func (shm *SelfHealingManager) failHealing(nodeID string) { + shm.mu.Lock() + defer shm.mu.Unlock() + + if event, ok := shm.healingEvents.Load(nodeID); ok { + shm.healingHistory = append(shm.healingHistory, event.(*HealingEvent)) + shm.healingEvents.Delete(nodeID) + } +} + +// healingLoop периодически проверяет зависшие восстановления +func (shm *SelfHealingManager) healingLoop() { + defer shm.wg.Done() + + ticker := time.NewTicker(30 * time.Second) + defer ticker.Stop() + + for { + select { + case <-shm.stopChan: + return + case <-ticker.C: + shm.checkStalledHealings() + } + } +} + +// checkStalledHealings проверяет зависшие восстановления +func (shm *SelfHealingManager) checkStalledHealings() { + now := time.Now().UnixMilli() + + shm.healingEvents.Range(func(key, value interface{}) bool { + nodeID := key.(string) + event := value.(*HealingEvent) + + if event.State == HealingStateHealing || event.State == HealingStateVerifying { + if now-event.Timestamp > int64(shm.config.HealingTimeoutSec*1000) { + // Восстановление зависло + event.State = HealingStateFailed + event.Error = "Healing stalled" + shm.healingEvents.Store(nodeID, event) + shm.failHealing(nodeID) + + if shm.logger != nil { + shm.logger.Warn(fmt.Sprintf("Healing for node %s stalled, marking as failed", nodeID)) + } + } + } + + return true + }) +} + +// GetHealingStatus возвращает статус восстановления для узла +func (shm *SelfHealingManager) GetHealingStatus(nodeID string) *HealingEvent { + if event, ok := shm.healingEvents.Load(nodeID); ok { + return event.(*HealingEvent) + } + return nil +} + +// GetHealingHistory возвращает историю восстановлений +func (shm *SelfHealingManager) GetHealingHistory() []*HealingEvent { + shm.mu.RLock() + defer shm.mu.RUnlock() + + history := make([]*HealingEvent, len(shm.healingHistory)) + copy(history, shm.healingHistory) + return history +} + +// GetMetrics возвращает метрики восстановления +func (shm *SelfHealingManager) GetMetrics() map[string]interface{} { + return map[string]interface{}{ + "total_healings": shm.metrics.TotalHealings.Load(), + "successful_healings": shm.metrics.SuccessfulHealings.Load(), + "failed_healings": shm.metrics.FailedHealings.Load(), + "reshardings_triggered": shm.metrics.ReshardingsTriggered.Load(), + "data_recoveries": shm.metrics.DataRecoveries.Load(), + "restarts_triggered": shm.metrics.RestartsTriggered.Load(), + "active_healings": shm.activeHealings.Load(), + "history_count": len(shm.healingHistory), + } +} + +// RegisterEventHandler регистрирует обработчик событий +func (shm *SelfHealingManager) RegisterEventHandler(name string, handler func(*HealingEvent)) { + shm.handlerMu.Lock() + defer shm.handlerMu.Unlock() + shm.eventHandlers[name] = handler +} + +// emitEvent отправляет событие всем обработчикам +func (shm *SelfHealingManager) emitEvent(event *HealingEvent) { + shm.handlerMu.RLock() + defer shm.handlerMu.RUnlock() + + for _, handler := range shm.eventHandlers { + go handler(event) + } +}