Upload files to "internal/cluster"
This commit is contained in:
516
internal/cluster/leader_fallback.go
Normal file
516
internal/cluster/leader_fallback.go
Normal file
@@ -0,0 +1,516 @@
|
||||
/*
|
||||
* 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(),
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user