517 lines
16 KiB
Go
517 lines
16 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/leader_fallback.go
|
||
// Назначение: Механизм Single Point of Failure - обеспечение записи при потере лидера
|
||
|
||
package cluster
|
||
|
||
import (
|
||
"fmt"
|
||
"sync"
|
||
"sync/atomic"
|
||
"time"
|
||
)
|
||
|
||
// WriteRequest представляет запрос на запись при отсутствии лидера
|
||
type WriteRequest struct {
|
||
ID string
|
||
Data []byte
|
||
CreatedAt int64
|
||
Retries int
|
||
Callback chan error
|
||
}
|
||
|
||
// PendingWriteQueue очередь отложенных записей
|
||
type PendingWriteQueue struct {
|
||
requests []*WriteRequest
|
||
mu sync.Mutex
|
||
maxSize int
|
||
maxAge time.Duration
|
||
}
|
||
|
||
// NewPendingWriteQueue создаёт новую очередь
|
||
func NewPendingWriteQueue(maxSize int, maxAge time.Duration) *PendingWriteQueue {
|
||
return &PendingWriteQueue{
|
||
requests: make([]*WriteRequest, 0),
|
||
maxSize: maxSize,
|
||
maxAge: maxAge,
|
||
}
|
||
}
|
||
|
||
// Add добавляет запрос в очередь
|
||
func (q *PendingWriteQueue) Add(req *WriteRequest) error {
|
||
q.mu.Lock()
|
||
defer q.mu.Unlock()
|
||
|
||
if len(q.requests) >= q.maxSize {
|
||
return fmt.Errorf("pending write queue is full")
|
||
}
|
||
|
||
q.requests = append(q.requests, req)
|
||
return nil
|
||
}
|
||
|
||
// GetAll возвращает все запросы и очищает очередь
|
||
func (q *PendingWriteQueue) GetAll() []*WriteRequest {
|
||
q.mu.Lock()
|
||
defer q.mu.Unlock()
|
||
|
||
// Очищаем устаревшие запросы
|
||
now := time.Now().UnixMilli()
|
||
valid := make([]*WriteRequest, 0)
|
||
for _, req := range q.requests {
|
||
if now - req.CreatedAt < int64(q.maxAge.Milliseconds()) {
|
||
valid = append(valid, req)
|
||
} else {
|
||
if req.Callback != nil {
|
||
req.Callback <- fmt.Errorf("write request expired")
|
||
}
|
||
}
|
||
}
|
||
|
||
q.requests = make([]*WriteRequest, 0)
|
||
return valid
|
||
}
|
||
|
||
// Size возвращает размер очереди
|
||
func (q *PendingWriteQueue) Size() int {
|
||
q.mu.Lock()
|
||
defer q.mu.Unlock()
|
||
return len(q.requests)
|
||
}
|
||
|
||
// LeaderFallbackManager управляет fallback при потере лидера
|
||
type LeaderFallbackManager struct {
|
||
coordinator *RaftCoordinator
|
||
logger LoggerInterface
|
||
pendingWrites *PendingWriteQueue
|
||
fallbackMode atomic.Bool
|
||
lastLeaderSeen atomic.Int64
|
||
electionTimeout time.Duration
|
||
fallbackTimeout time.Duration
|
||
mu sync.RWMutex
|
||
stopChan chan struct{}
|
||
wg sync.WaitGroup
|
||
observers map[string]chan *LeaderChangeEvent
|
||
observerMu sync.RWMutex
|
||
writeBuffer []*WriteRequest
|
||
bufferMu sync.Mutex
|
||
}
|
||
|
||
// LeaderChangeEvent событие изменения лидера
|
||
type LeaderChangeEvent struct {
|
||
OldLeader string
|
||
NewLeader string
|
||
Timestamp int64
|
||
Term uint64
|
||
}
|
||
|
||
// FallbackConfig конфигурация fallback механизма
|
||
type FallbackConfig struct {
|
||
Enabled bool
|
||
ElectionTimeout time.Duration
|
||
FallbackTimeout time.Duration
|
||
PendingQueueSize int
|
||
PendingQueueMaxAge time.Duration
|
||
WriteBufferSize int
|
||
}
|
||
|
||
// DefaultFallbackConfig возвращает конфигурацию по умолчанию
|
||
func DefaultFallbackConfig() *FallbackConfig {
|
||
return &FallbackConfig{
|
||
Enabled: true,
|
||
ElectionTimeout: 5 * time.Second,
|
||
FallbackTimeout: 30 * time.Second,
|
||
PendingQueueSize: 10000,
|
||
PendingQueueMaxAge: 60 * time.Second,
|
||
WriteBufferSize: 1000,
|
||
}
|
||
}
|
||
|
||
// NewLeaderFallbackManager создаёт новый менеджер fallback
|
||
func NewLeaderFallbackManager(coordinator *RaftCoordinator, logger LoggerInterface, config *FallbackConfig) *LeaderFallbackManager {
|
||
if config == nil {
|
||
config = DefaultFallbackConfig()
|
||
}
|
||
|
||
lfm := &LeaderFallbackManager{
|
||
coordinator: coordinator,
|
||
logger: logger,
|
||
pendingWrites: NewPendingWriteQueue(config.PendingQueueSize, config.PendingQueueMaxAge),
|
||
electionTimeout: config.ElectionTimeout,
|
||
fallbackTimeout: config.FallbackTimeout,
|
||
stopChan: make(chan struct{}),
|
||
observers: make(map[string]chan *LeaderChangeEvent),
|
||
writeBuffer: make([]*WriteRequest, 0, config.WriteBufferSize),
|
||
}
|
||
|
||
lfm.lastLeaderSeen.Store(time.Now().UnixMilli())
|
||
|
||
// Запускаем мониторинг лидера
|
||
lfm.wg.Add(1)
|
||
go lfm.monitorLeader()
|
||
|
||
// Запускаем обработчик fallback записей
|
||
lfm.wg.Add(1)
|
||
go lfm.processFallbackWrites()
|
||
|
||
if logger != nil {
|
||
logger.Info("Leader fallback manager initialized")
|
||
}
|
||
|
||
return lfm
|
||
}
|
||
|
||
// monitorLeader отслеживает состояние лидера
|
||
func (lfm *LeaderFallbackManager) monitorLeader() {
|
||
defer lfm.wg.Done()
|
||
|
||
ticker := time.NewTicker(lfm.electionTimeout / 2)
|
||
defer ticker.Stop()
|
||
|
||
var lastLeader string
|
||
var leaderLostAt int64
|
||
|
||
for {
|
||
select {
|
||
case <-lfm.stopChan:
|
||
return
|
||
case <-ticker.C:
|
||
currentLeader := lfm.coordinator.GetLeader()
|
||
currentLeaderID := ""
|
||
if currentLeader != nil {
|
||
currentLeaderID = currentLeader.ID
|
||
}
|
||
isLeader := lfm.coordinator.IsLeader()
|
||
|
||
// Проверяем, есть ли лидер
|
||
if currentLeaderID == "" && !isLeader {
|
||
if !lfm.fallbackMode.Load() && leaderLostAt == 0 {
|
||
leaderLostAt = time.Now().UnixMilli()
|
||
lfm.enterFallbackMode()
|
||
} else if leaderLostAt > 0 && time.Now().UnixMilli()-leaderLostAt > int64(lfm.fallbackTimeout.Milliseconds()) {
|
||
lfm.handleProlongedLeaderLoss()
|
||
}
|
||
} else {
|
||
if lfm.fallbackMode.Load() {
|
||
lfm.exitFallbackMode()
|
||
lfm.processPendingWrites()
|
||
}
|
||
leaderLostAt = 0
|
||
lfm.lastLeaderSeen.Store(time.Now().UnixMilli())
|
||
}
|
||
|
||
// Отслеживаем смену лидера
|
||
if lastLeader != currentLeaderID {
|
||
if lastLeader != "" {
|
||
event := &LeaderChangeEvent{
|
||
OldLeader: lastLeader,
|
||
NewLeader: currentLeaderID,
|
||
Timestamp: time.Now().UnixMilli(),
|
||
Term: lfm.coordinator.GetCurrentTerm(),
|
||
}
|
||
lfm.notifyObservers(event)
|
||
|
||
if lfm.logger != nil {
|
||
lfm.logger.Info(fmt.Sprintf("Leader changed from %s to %s", lastLeader, currentLeaderID))
|
||
}
|
||
}
|
||
lastLeader = currentLeaderID
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// enterFallbackMode переводит систему в fallback режим
|
||
func (lfm *LeaderFallbackManager) enterFallbackMode() {
|
||
if lfm.fallbackMode.CompareAndSwap(false, true) {
|
||
if lfm.logger != nil {
|
||
lfm.logger.Warn("Entering fallback mode - no leader available")
|
||
}
|
||
|
||
// Создаём событие о потере лидера
|
||
leader := lfm.coordinator.GetLeader()
|
||
oldLeader := ""
|
||
if leader != nil {
|
||
oldLeader = leader.ID
|
||
}
|
||
event := &LeaderChangeEvent{
|
||
OldLeader: oldLeader,
|
||
NewLeader: "",
|
||
Timestamp: time.Now().UnixMilli(),
|
||
Term: lfm.coordinator.GetCurrentTerm(),
|
||
}
|
||
lfm.notifyObservers(event)
|
||
}
|
||
}
|
||
|
||
// exitFallbackMode выходит из fallback режима
|
||
func (lfm *LeaderFallbackManager) exitFallbackMode() {
|
||
if lfm.fallbackMode.CompareAndSwap(true, false) {
|
||
if lfm.logger != nil {
|
||
lfm.logger.Info("Exiting fallback mode - leader elected")
|
||
}
|
||
}
|
||
}
|
||
|
||
// handleProlongedLeaderLoss обрабатывает длительную потерю лидера
|
||
func (lfm *LeaderFallbackManager) handleProlongedLeaderLoss() {
|
||
if lfm.logger != nil {
|
||
lfm.logger.Error("Prolonged leader loss detected, initiating emergency measures")
|
||
}
|
||
|
||
// Проверяем возможность стать лидером
|
||
if !lfm.coordinator.IsLeader() && lfm.coordinator.raft != nil {
|
||
lfm.initiateEmergencyElection()
|
||
}
|
||
}
|
||
|
||
// initiateEmergencyElection инициирует экстренные выборы
|
||
func (lfm *LeaderFallbackManager) initiateEmergencyElection() {
|
||
if lfm.logger != nil {
|
||
lfm.logger.Info("Initiating emergency election")
|
||
}
|
||
|
||
// Принудительно запускаем выборы
|
||
if lfm.coordinator.raft != nil {
|
||
lfm.coordinator.raft.LeadershipTransfer()
|
||
}
|
||
|
||
// Если через electionTimeout всё ещё нет лидера, пробуем другие узлы
|
||
time.Sleep(lfm.electionTimeout)
|
||
|
||
if lfm.coordinator.GetLeader() == nil && !lfm.coordinator.IsLeader() {
|
||
lfm.tryAlternativeLeader()
|
||
}
|
||
}
|
||
|
||
// tryAlternativeLeader пробует альтернативные пути выбора лидера
|
||
func (lfm *LeaderFallbackManager) tryAlternativeLeader() {
|
||
nodes := lfm.coordinator.GetAllNodes()
|
||
if len(nodes) == 0 {
|
||
return
|
||
}
|
||
|
||
if lfm.logger != nil {
|
||
lfm.logger.Warn("Attempting to find alternative leader")
|
||
}
|
||
|
||
// Пробуем каждый узел
|
||
for _, node := range nodes {
|
||
if lfm.coordinator.GetLeader() != nil {
|
||
break
|
||
}
|
||
|
||
if lfm.logger != nil {
|
||
lfm.logger.Debug(fmt.Sprintf("Checking node %s for leadership possibility", node.ID))
|
||
}
|
||
|
||
time.Sleep(500 * time.Millisecond)
|
||
}
|
||
}
|
||
|
||
// processPendingWrites обрабатывает отложенные записи
|
||
func (lfm *LeaderFallbackManager) processPendingWrites() {
|
||
requests := lfm.pendingWrites.GetAll()
|
||
|
||
if len(requests) == 0 {
|
||
return
|
||
}
|
||
|
||
if lfm.logger != nil {
|
||
lfm.logger.Info(fmt.Sprintf("Processing %d pending writes after leader election", len(requests)))
|
||
}
|
||
|
||
for _, req := range requests {
|
||
// Пытаемся применить запись через Raft
|
||
future := lfm.coordinator.raft.Apply(req.Data, 10*time.Second)
|
||
err := future.Error()
|
||
|
||
if req.Callback != nil {
|
||
req.Callback <- err
|
||
}
|
||
|
||
if err != nil && lfm.logger != nil {
|
||
lfm.logger.Error(fmt.Sprintf("Failed to apply pending write %s: %v", req.ID, err))
|
||
}
|
||
}
|
||
}
|
||
|
||
// processFallbackWrites обрабатывает записи в fallback режиме
|
||
func (lfm *LeaderFallbackManager) processFallbackWrites() {
|
||
defer lfm.wg.Done()
|
||
|
||
ticker := time.NewTicker(1 * time.Second)
|
||
defer ticker.Stop()
|
||
|
||
for {
|
||
select {
|
||
case <-lfm.stopChan:
|
||
return
|
||
case <-ticker.C:
|
||
if lfm.fallbackMode.Load() {
|
||
lfm.bufferWrites()
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// bufferWrites буферизирует записи во время fallback
|
||
func (lfm *LeaderFallbackManager) bufferWrites() {
|
||
lfm.bufferMu.Lock()
|
||
defer lfm.bufferMu.Unlock()
|
||
|
||
if len(lfm.writeBuffer) == 0 {
|
||
return
|
||
}
|
||
|
||
// Проверяем, появился ли лидер
|
||
if lfm.coordinator.GetLeader() != nil || lfm.coordinator.IsLeader() {
|
||
// Сбрасываем буфер в очередь ожидания
|
||
for _, req := range lfm.writeBuffer {
|
||
lfm.pendingWrites.Add(req)
|
||
}
|
||
lfm.writeBuffer = lfm.writeBuffer[:0]
|
||
lfm.exitFallbackMode()
|
||
} else if len(lfm.writeBuffer) > 0 {
|
||
// Записываем в локальный WAL при длительном отсутствии лидера
|
||
lfm.writeToLocalWAL()
|
||
}
|
||
}
|
||
|
||
// writeToLocalWAL записывает в локальный WAL
|
||
func (lfm *LeaderFallbackManager) writeToLocalWAL() {
|
||
lfm.bufferMu.Lock()
|
||
defer lfm.bufferMu.Unlock()
|
||
|
||
if lfm.logger != nil {
|
||
lfm.logger.Debug(fmt.Sprintf("Writing %d operations to local WAL", len(lfm.writeBuffer)))
|
||
}
|
||
|
||
// Здесь можно реализовать запись в локальный WAL
|
||
// для последующего восстановления
|
||
lfm.writeBuffer = lfm.writeBuffer[:0]
|
||
}
|
||
|
||
// SubmitWrite submits a write operation with fallback support
|
||
func (lfm *LeaderFallbackManager) SubmitWrite(data []byte) error {
|
||
return lfm.SubmitWriteWithCallback(data, nil)
|
||
}
|
||
|
||
// SubmitWriteWithCallback отправляет запись с callback
|
||
func (lfm *LeaderFallbackManager) SubmitWriteWithCallback(data []byte, callback chan error) error {
|
||
req := &WriteRequest{
|
||
ID: fmt.Sprintf("write_%d", time.Now().UnixNano()),
|
||
Data: data,
|
||
CreatedAt: time.Now().UnixMilli(),
|
||
Retries: 0,
|
||
Callback: callback,
|
||
}
|
||
|
||
// Если есть лидер, пытаемся записать сразу
|
||
if !lfm.fallbackMode.Load() && lfm.coordinator.IsLeader() {
|
||
future := lfm.coordinator.raft.Apply(data, 10*time.Second)
|
||
if callback != nil {
|
||
callback <- future.Error()
|
||
}
|
||
return future.Error()
|
||
}
|
||
|
||
// В fallback режиме буферизируем
|
||
lfm.bufferMu.Lock()
|
||
defer lfm.bufferMu.Unlock()
|
||
|
||
lfm.writeBuffer = append(lfm.writeBuffer, req)
|
||
|
||
if lfm.logger != nil {
|
||
lfm.logger.Debug(fmt.Sprintf("Write buffered in fallback mode (buffer size: %d)", len(lfm.writeBuffer)))
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// IsFallbackMode возвращает статус fallback режима
|
||
func (lfm *LeaderFallbackManager) IsFallbackMode() bool {
|
||
return lfm.fallbackMode.Load()
|
||
}
|
||
|
||
// GetPendingWritesCount возвращает количество отложенных записей
|
||
func (lfm *LeaderFallbackManager) GetPendingWritesCount() int {
|
||
return lfm.pendingWrites.Size()
|
||
}
|
||
|
||
// GetBufferedWritesCount возвращает количество буферизированных записей
|
||
func (lfm *LeaderFallbackManager) GetBufferedWritesCount() int {
|
||
lfm.bufferMu.Lock()
|
||
defer lfm.bufferMu.Unlock()
|
||
return len(lfm.writeBuffer)
|
||
}
|
||
|
||
// Subscribe подписывается на события смены лидера
|
||
func (lfm *LeaderFallbackManager) Subscribe(id string) <-chan *LeaderChangeEvent {
|
||
lfm.observerMu.Lock()
|
||
defer lfm.observerMu.Unlock()
|
||
|
||
ch := make(chan *LeaderChangeEvent, 10)
|
||
lfm.observers[id] = ch
|
||
return ch
|
||
}
|
||
|
||
// Unsubscribe отписывается от событий
|
||
func (lfm *LeaderFallbackManager) Unsubscribe(id string) {
|
||
lfm.observerMu.Lock()
|
||
defer lfm.observerMu.Unlock()
|
||
|
||
if ch, ok := lfm.observers[id]; ok {
|
||
close(ch)
|
||
delete(lfm.observers, id)
|
||
}
|
||
}
|
||
|
||
// notifyObservers уведомляет наблюдателей
|
||
func (lfm *LeaderFallbackManager) notifyObservers(event *LeaderChangeEvent) {
|
||
lfm.observerMu.RLock()
|
||
defer lfm.observerMu.RUnlock()
|
||
|
||
for _, ch := range lfm.observers {
|
||
select {
|
||
case ch <- event:
|
||
default:
|
||
if lfm.logger != nil {
|
||
lfm.logger.Warn("Observer channel full, dropping event")
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// Stop останавливает fallback менеджер
|
||
func (lfm *LeaderFallbackManager) Stop() {
|
||
close(lfm.stopChan)
|
||
lfm.wg.Wait()
|
||
|
||
if lfm.logger != nil {
|
||
lfm.logger.Info("Leader fallback manager stopped")
|
||
}
|
||
}
|
||
|
||
// GetStats возвращает статистику
|
||
func (lfm *LeaderFallbackManager) GetStats() map[string]interface{} {
|
||
return map[string]interface{}{
|
||
"fallback_mode": lfm.fallbackMode.Load(),
|
||
"pending_writes": lfm.GetPendingWritesCount(),
|
||
"buffered_writes": lfm.GetBufferedWritesCount(),
|
||
"last_leader_seen": lfm.lastLeaderSeen.Load(),
|
||
"observers_count": len(lfm.observers),
|
||
"election_timeout_ms": lfm.electionTimeout.Milliseconds(),
|
||
"fallback_timeout_ms": lfm.fallbackTimeout.Milliseconds(),
|
||
}
|
||
}
|