290 lines
8.6 KiB
Go
290 lines
8.6 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/cluster/recovery.go
|
||
// Назначение: Автоматическое восстановление узлов после сбоя
|
||
|
||
package cluster
|
||
|
||
import (
|
||
"fmt"
|
||
"sync"
|
||
"sync/atomic"
|
||
"time"
|
||
)
|
||
|
||
// NodeState представляет состояние узла для восстановления
|
||
type NodeState struct {
|
||
NodeID string `json:"node_id"`
|
||
LastSeen time.Time `json:"last_seen"`
|
||
LastLogIndex uint64 `json:"last_log_index"`
|
||
FailureCount int `json:"failure_count"`
|
||
IsRecovering bool `json:"is_recovering"`
|
||
}
|
||
|
||
// RecoveryManager управляет восстановлением узлов
|
||
type RecoveryManager struct {
|
||
coordinator *RaftCoordinator
|
||
logger LoggerInterface
|
||
states sync.Map
|
||
maxFailures int
|
||
recoveryDelay time.Duration
|
||
stopChan chan struct{}
|
||
wg sync.WaitGroup
|
||
isActive atomic.Bool
|
||
}
|
||
|
||
// NewRecoveryManager создаёт новый менеджер восстановления
|
||
func NewRecoveryManager(coordinator *RaftCoordinator, logger LoggerInterface) *RecoveryManager {
|
||
rm := &RecoveryManager{
|
||
coordinator: coordinator,
|
||
logger: logger,
|
||
maxFailures: 3,
|
||
recoveryDelay: 30 * time.Second,
|
||
stopChan: make(chan struct{}),
|
||
}
|
||
|
||
return rm
|
||
}
|
||
|
||
// Start запускает мониторинг восстановления
|
||
func (rm *RecoveryManager) Start() {
|
||
rm.isActive.Store(true)
|
||
rm.wg.Add(2)
|
||
go rm.monitorLoop()
|
||
go rm.recoveryLoop()
|
||
|
||
if rm.logger != nil {
|
||
rm.logger.Info("Recovery manager started")
|
||
}
|
||
}
|
||
|
||
// Stop останавливает менеджер восстановления
|
||
func (rm *RecoveryManager) Stop() {
|
||
rm.isActive.Store(false)
|
||
close(rm.stopChan)
|
||
rm.wg.Wait()
|
||
|
||
if rm.logger != nil {
|
||
rm.logger.Info("Recovery manager stopped")
|
||
}
|
||
}
|
||
|
||
// monitorLoop отслеживает состояние узлов
|
||
func (rm *RecoveryManager) monitorLoop() {
|
||
defer rm.wg.Done()
|
||
|
||
ticker := time.NewTicker(10 * time.Second)
|
||
defer ticker.Stop()
|
||
|
||
for {
|
||
select {
|
||
case <-rm.stopChan:
|
||
return
|
||
case <-ticker.C:
|
||
rm.checkNodesHealth()
|
||
}
|
||
}
|
||
}
|
||
|
||
// checkNodesHealth проверяет здоровье узлов
|
||
func (rm *RecoveryManager) checkNodesHealth() {
|
||
if rm.coordinator == nil {
|
||
return
|
||
}
|
||
|
||
nodes := rm.coordinator.GetAllNodes()
|
||
now := time.Now()
|
||
|
||
for _, node := range nodes {
|
||
stateVal, ok := rm.states.Load(node.ID)
|
||
var state *NodeState
|
||
if ok {
|
||
state = stateVal.(*NodeState)
|
||
} else {
|
||
state = &NodeState{
|
||
NodeID: node.ID,
|
||
LastSeen: now,
|
||
LastLogIndex: 0,
|
||
FailureCount: 0,
|
||
IsRecovering: false,
|
||
}
|
||
rm.states.Store(node.ID, state)
|
||
}
|
||
|
||
// Проверяем, был ли недавно heartbeat
|
||
lastSeen := time.UnixMilli(node.LastSeen)
|
||
if now.Sub(lastSeen) > 30*time.Second {
|
||
state.FailureCount++
|
||
if rm.logger != nil {
|
||
rm.logger.Warn(fmt.Sprintf("Node %s appears unhealthy, failure count: %d", node.ID, state.FailureCount))
|
||
}
|
||
|
||
if state.FailureCount >= rm.maxFailures && !state.IsRecovering {
|
||
rm.triggerRecovery(node.ID)
|
||
}
|
||
} else {
|
||
if state.FailureCount > 0 {
|
||
state.FailureCount = 0
|
||
if rm.logger != nil {
|
||
rm.logger.Info(fmt.Sprintf("Node %s recovered", node.ID))
|
||
}
|
||
}
|
||
}
|
||
|
||
state.LastSeen = now
|
||
rm.states.Store(node.ID, state)
|
||
}
|
||
}
|
||
|
||
// triggerRecovery запускает восстановление узла
|
||
func (rm *RecoveryManager) triggerRecovery(nodeID string) {
|
||
stateVal, ok := rm.states.Load(nodeID)
|
||
if !ok {
|
||
return
|
||
}
|
||
|
||
state := stateVal.(*NodeState)
|
||
if state.IsRecovering {
|
||
return
|
||
}
|
||
|
||
state.IsRecovering = true
|
||
rm.states.Store(nodeID, state)
|
||
|
||
if rm.logger != nil {
|
||
rm.logger.Info(fmt.Sprintf("Triggering recovery for node %s", nodeID))
|
||
}
|
||
|
||
// Запускаем асинхронное восстановление
|
||
go rm.recoverNode(nodeID)
|
||
}
|
||
|
||
// recoverNode восстанавливает узел
|
||
func (rm *RecoveryManager) recoverNode(nodeID string) {
|
||
defer func() {
|
||
if r := recover(); r != nil {
|
||
if rm.logger != nil {
|
||
rm.logger.Error(fmt.Sprintf("Recovery for node %s panicked: %v", nodeID, r))
|
||
}
|
||
}
|
||
|
||
// Сбрасываем флаг восстановления
|
||
if stateVal, ok := rm.states.Load(nodeID); ok {
|
||
state := stateVal.(*NodeState)
|
||
state.IsRecovering = false
|
||
rm.states.Store(nodeID, state)
|
||
}
|
||
}()
|
||
|
||
// Ждём перед попыткой восстановления
|
||
time.Sleep(rm.recoveryDelay)
|
||
|
||
if rm.coordinator == nil {
|
||
return
|
||
}
|
||
|
||
// Проверяем, не восстановился ли узел самостоятельно
|
||
node := rm.coordinator.GetNodeByID(nodeID)
|
||
if node != nil && time.Now().UnixMilli()-node.LastSeen < 30000 {
|
||
if rm.logger != nil {
|
||
rm.logger.Info(fmt.Sprintf("Node %s recovered on its own", nodeID))
|
||
}
|
||
return
|
||
}
|
||
|
||
// Пытаемся переподключить узел
|
||
if rm.logger != nil {
|
||
rm.logger.Info(fmt.Sprintf("Attempting to reconnect node %s", nodeID))
|
||
}
|
||
|
||
// Обновляем статус узла
|
||
if err := rm.coordinator.UpdateNodeStatus(nodeID, StatusActive); err != nil {
|
||
if rm.logger != nil {
|
||
rm.logger.Error(fmt.Sprintf("Failed to update node %s status: %v", nodeID, err))
|
||
}
|
||
}
|
||
|
||
// Запускаем синхронизацию данных
|
||
if err := rm.syncNodeData(nodeID); err != nil {
|
||
if rm.logger != nil {
|
||
rm.logger.Error(fmt.Sprintf("Failed to sync node %s data: %v", nodeID, err))
|
||
}
|
||
}
|
||
|
||
if rm.logger != nil {
|
||
rm.logger.Info(fmt.Sprintf("Recovery completed for node %s", nodeID))
|
||
}
|
||
}
|
||
|
||
// syncNodeData синхронизирует данные с узлом
|
||
func (rm *RecoveryManager) syncNodeData(nodeID string) error {
|
||
// Получаем информацию об узле
|
||
node := rm.coordinator.GetNodeByID(nodeID)
|
||
if node == nil {
|
||
return fmt.Errorf("node not found: %s", nodeID)
|
||
}
|
||
|
||
if rm.logger != nil {
|
||
rm.logger.Debug(fmt.Sprintf("Syncing data with node %s at %s:%d", nodeID, node.IP, node.Port))
|
||
}
|
||
|
||
// В реальной реализации здесь была бы синхронизация данных
|
||
// Например, отправка всех изменений с момента последнего успешного heartbeat
|
||
|
||
return nil
|
||
}
|
||
|
||
// recoveryLoop периодически проверяет и восстанавливает узлы
|
||
func (rm *RecoveryManager) recoveryLoop() {
|
||
defer rm.wg.Done()
|
||
|
||
ticker := time.NewTicker(1 * time.Minute)
|
||
defer ticker.Stop()
|
||
|
||
for {
|
||
select {
|
||
case <-rm.stopChan:
|
||
return
|
||
case <-ticker.C:
|
||
rm.attemptRecoveryAll()
|
||
}
|
||
}
|
||
}
|
||
|
||
// attemptRecoveryAll пытается восстановить все проблемные узлы
|
||
func (rm *RecoveryManager) attemptRecoveryAll() {
|
||
rm.states.Range(func(key, value interface{}) bool {
|
||
state := value.(*NodeState)
|
||
if state.FailureCount >= rm.maxFailures && !state.IsRecovering {
|
||
go rm.recoverNode(state.NodeID)
|
||
}
|
||
return true
|
||
})
|
||
}
|
||
|
||
// GetNodeState возвращает состояние узла
|
||
func (rm *RecoveryManager) GetNodeState(nodeID string) *NodeState {
|
||
if val, ok := rm.states.Load(nodeID); ok {
|
||
return val.(*NodeState)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// GetAllStates возвращает все состояния узлов
|
||
func (rm *RecoveryManager) GetAllStates() map[string]*NodeState {
|
||
states := make(map[string]*NodeState)
|
||
rm.states.Range(func(key, value interface{}) bool {
|
||
states[key.(string)] = value.(*NodeState)
|
||
return true
|
||
})
|
||
return states
|
||
}
|