3073 lines
98 KiB
Go
3073 lines
98 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/raft_coordinator.go
|
||
// Назначение: Реализация координатора распределённого кластера на основе Raft консенсус-алгоритма.
|
||
// Поддерживает: Pipeline репликацию, Batch commit, Dynamic Resharding, Joint Consensus.
|
||
// Новые возможности: Single Point of Failure (Leader Fallback), Panic Recovery, Schema Migration.
|
||
|
||
package cluster
|
||
|
||
import (
|
||
"crypto/md5"
|
||
"encoding/binary"
|
||
"encoding/json"
|
||
"fmt"
|
||
"io"
|
||
"net"
|
||
"os"
|
||
"path/filepath"
|
||
"sort"
|
||
"sync"
|
||
"sync/atomic"
|
||
"time"
|
||
|
||
"github.com/hashicorp/raft"
|
||
"futriis/internal/log"
|
||
"futriis/internal/config"
|
||
"futriis/internal/migration"
|
||
"futriis/internal/storage"
|
||
)
|
||
|
||
// ========== Определения структур команд Raft ==========
|
||
|
||
// NodeRegistrationCommand команда регистрации узла
|
||
type NodeRegistrationCommand struct {
|
||
Type string `json:"type"`
|
||
Node NodeInfo `json:"node,omitempty"`
|
||
NodeID string `json:"node_id,omitempty"`
|
||
Factor int32 `json:"factor,omitempty"`
|
||
Shard *ShardInfo `json:"shard,omitempty"`
|
||
ShardID string `json:"shard_id,omitempty"`
|
||
TargetNode string `json:"target_node,omitempty"`
|
||
Data map[string]interface{} `json:"data,omitempty"`
|
||
Timestamp int64 `json:"timestamp"`
|
||
}
|
||
|
||
// NodeStatusCommand команда обновления статуса узла
|
||
type NodeStatusCommand struct {
|
||
Type string `json:"type"`
|
||
NodeID string `json:"node_id"`
|
||
Status int32 `json:"status"`
|
||
Timestamp int64 `json:"timestamp"`
|
||
}
|
||
|
||
// ========== Pipeline Replication Structures ==========
|
||
|
||
// PipelineBatch представляет пакет команд для группировки
|
||
type PipelineBatch struct {
|
||
ID string `json:"id"`
|
||
Commands []interface{} `json:"commands"`
|
||
CreatedAt int64 `json:"created_at"`
|
||
Size int `json:"size"`
|
||
}
|
||
|
||
// BatchCommand представляет команду для пакетной обработки
|
||
type BatchCommand struct {
|
||
Type string `json:"type"`
|
||
BatchID string `json:"batch_id"`
|
||
Commands []interface{} `json:"commands"`
|
||
Size int `json:"size"`
|
||
Timestamp int64 `json:"timestamp"`
|
||
}
|
||
|
||
// PipelineReplicator управляет группировкой команд в Raft лог
|
||
type PipelineReplicator struct {
|
||
pendingBatches chan *PipelineBatch
|
||
batchSize int
|
||
batchTimeout time.Duration
|
||
coordinator *RaftCoordinator
|
||
logger *log.Logger
|
||
stopChan chan struct{}
|
||
wg sync.WaitGroup
|
||
batchCount atomic.Uint64
|
||
commandsCount atomic.Uint64
|
||
}
|
||
|
||
// NewPipelineReplicator создаёт новый репликатор с пайплайном
|
||
func NewPipelineReplicator(coord *RaftCoordinator, batchSize int, timeout time.Duration, logger *log.Logger) *PipelineReplicator {
|
||
pr := &PipelineReplicator{
|
||
pendingBatches: make(chan *PipelineBatch, 1000),
|
||
batchSize: batchSize,
|
||
batchTimeout: timeout,
|
||
coordinator: coord,
|
||
logger: logger,
|
||
stopChan: make(chan struct{}),
|
||
}
|
||
|
||
go pr.processBatches()
|
||
if logger != nil {
|
||
logger.Debug(fmt.Sprintf("Pipeline replicator initialized: batchSize=%d, timeout=%v", batchSize, timeout))
|
||
}
|
||
return pr
|
||
}
|
||
|
||
// AddCommand добавляет команду в пайплайн
|
||
func (pr *PipelineReplicator) AddCommand(cmd interface{}) error {
|
||
batch := &PipelineBatch{
|
||
ID: fmt.Sprintf("batch_%d_%d", time.Now().UnixNano(), pr.batchCount.Add(1)),
|
||
Commands: []interface{}{cmd},
|
||
CreatedAt: time.Now().UnixMilli(),
|
||
Size: 1,
|
||
}
|
||
|
||
select {
|
||
case pr.pendingBatches <- batch:
|
||
pr.commandsCount.Add(1)
|
||
return nil
|
||
case <-time.After(100 * time.Millisecond):
|
||
return fmt.Errorf("pipeline queue full")
|
||
}
|
||
}
|
||
|
||
// processBatches обрабатывает пакеты команд
|
||
func (pr *PipelineReplicator) processBatches() {
|
||
pr.wg.Add(1)
|
||
defer pr.wg.Done()
|
||
|
||
ticker := time.NewTicker(pr.batchTimeout)
|
||
defer ticker.Stop()
|
||
|
||
var currentBatch *PipelineBatch
|
||
batchTimer := time.NewTimer(pr.batchTimeout)
|
||
batchTimer.Stop()
|
||
|
||
for {
|
||
select {
|
||
case <-pr.stopChan:
|
||
if currentBatch != nil && len(currentBatch.Commands) > 0 {
|
||
pr.applyBatch(currentBatch)
|
||
}
|
||
return
|
||
|
||
case batch := <-pr.pendingBatches:
|
||
if currentBatch == nil {
|
||
currentBatch = batch
|
||
batchTimer.Reset(pr.batchTimeout)
|
||
} else if len(currentBatch.Commands) < pr.batchSize {
|
||
currentBatch.Commands = append(currentBatch.Commands, batch.Commands...)
|
||
currentBatch.Size = len(currentBatch.Commands)
|
||
} else {
|
||
pr.applyBatch(currentBatch)
|
||
currentBatch = batch
|
||
batchTimer.Reset(pr.batchTimeout)
|
||
}
|
||
|
||
case <-batchTimer.C:
|
||
if currentBatch != nil && len(currentBatch.Commands) > 0 {
|
||
pr.applyBatch(currentBatch)
|
||
currentBatch = nil
|
||
}
|
||
|
||
case <-ticker.C:
|
||
if currentBatch != nil && len(currentBatch.Commands) > 0 {
|
||
pr.applyBatch(currentBatch)
|
||
currentBatch = nil
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// applyBatch применяет пакет команд через Raft
|
||
func (pr *PipelineReplicator) applyBatch(batch *PipelineBatch) {
|
||
if pr.coordinator == nil || !pr.coordinator.IsLeader() {
|
||
if pr.logger != nil {
|
||
pr.logger.Debug(fmt.Sprintf("Skipping batch %s: not leader", batch.ID))
|
||
}
|
||
return
|
||
}
|
||
|
||
batchCmd := BatchCommand{
|
||
Type: "batch",
|
||
BatchID: batch.ID,
|
||
Commands: batch.Commands,
|
||
Size: batch.Size,
|
||
Timestamp: time.Now().UnixMilli(),
|
||
}
|
||
|
||
data, err := json.Marshal(batchCmd)
|
||
if err != nil {
|
||
if pr.logger != nil {
|
||
pr.logger.Error(fmt.Sprintf("Failed to marshal batch %s: %v", batch.ID, err))
|
||
}
|
||
return
|
||
}
|
||
|
||
future := pr.coordinator.raft.Apply(data, 10*time.Second)
|
||
if err := future.Error(); err != nil {
|
||
if pr.logger != nil {
|
||
pr.logger.Error(fmt.Sprintf("Failed to apply batch %s: %v", batch.ID, err))
|
||
}
|
||
return
|
||
}
|
||
|
||
if pr.logger != nil {
|
||
pr.logger.Debug(fmt.Sprintf("Applied batch %s with %d commands", batch.ID, batch.Size))
|
||
}
|
||
}
|
||
|
||
// GetStats возвращает статистику пайплайна
|
||
func (pr *PipelineReplicator) GetStats() map[string]interface{} {
|
||
return map[string]interface{}{
|
||
"batch_count": pr.batchCount.Load(),
|
||
"commands_count": pr.commandsCount.Load(),
|
||
"queue_size": len(pr.pendingBatches),
|
||
"batch_size": pr.batchSize,
|
||
"batch_timeout": pr.batchTimeout.String(),
|
||
}
|
||
}
|
||
|
||
// Stop останавливает репликатор
|
||
func (pr *PipelineReplicator) Stop() {
|
||
close(pr.stopChan)
|
||
pr.wg.Wait()
|
||
}
|
||
|
||
// ========== Batch Commit Manager ==========
|
||
|
||
// CommitRequest представляет запрос на коммит
|
||
type CommitRequest struct {
|
||
ID string `json:"id"`
|
||
Operations []BatchOperation `json:"operations"`
|
||
CreatedAt int64 `json:"created_at"`
|
||
Callback chan error `json:"-"`
|
||
}
|
||
|
||
// BatchOperation представляет операцию для пакетного коммита
|
||
type BatchOperation struct {
|
||
Type string `json:"type"`
|
||
Database string `json:"database"`
|
||
Collection string `json:"collection"`
|
||
DocumentID string `json:"document_id"`
|
||
Data map[string]interface{} `json:"data"`
|
||
}
|
||
|
||
// BatchStorage хранит данные для пакетных коммитов
|
||
type BatchStorage struct {
|
||
mu sync.RWMutex
|
||
commits map[string]*CommitRequest
|
||
lastFlush int64
|
||
flushCount uint64
|
||
}
|
||
|
||
// BatchCommitManager управляет групповыми коммитами
|
||
type BatchCommitManager struct {
|
||
pendingCommits chan *CommitRequest
|
||
batchSize int
|
||
commitInterval time.Duration
|
||
fsyncEnabled bool
|
||
logger *log.Logger
|
||
stopChan chan struct{}
|
||
wg sync.WaitGroup
|
||
commitCount atomic.Uint64
|
||
operationsCount atomic.Uint64
|
||
storage *BatchStorage
|
||
}
|
||
|
||
// NewBatchCommitManager создаёт новый менеджер пакетных коммитов
|
||
func NewBatchCommitManager(batchSize int, interval time.Duration, fsyncEnabled bool, logger *log.Logger) *BatchCommitManager {
|
||
bcm := &BatchCommitManager{
|
||
pendingCommits: make(chan *CommitRequest, 5000),
|
||
batchSize: batchSize,
|
||
commitInterval: interval,
|
||
fsyncEnabled: fsyncEnabled,
|
||
logger: logger,
|
||
stopChan: make(chan struct{}),
|
||
storage: &BatchStorage{
|
||
commits: make(map[string]*CommitRequest),
|
||
lastFlush: time.Now().UnixMilli(),
|
||
},
|
||
}
|
||
|
||
go bcm.processCommits()
|
||
if logger != nil {
|
||
logger.Debug(fmt.Sprintf("Batch commit manager initialized: batchSize=%d, interval=%v, fsync=%v", batchSize, interval, fsyncEnabled))
|
||
}
|
||
return bcm
|
||
}
|
||
|
||
// AddCommit добавляет запрос на коммит
|
||
func (bcm *BatchCommitManager) AddCommit(operations []BatchOperation) error {
|
||
req := &CommitRequest{
|
||
ID: fmt.Sprintf("commit_%d_%d", time.Now().UnixNano(), bcm.commitCount.Add(1)),
|
||
Operations: operations,
|
||
CreatedAt: time.Now().UnixMilli(),
|
||
Callback: make(chan error, 1),
|
||
}
|
||
|
||
select {
|
||
case bcm.pendingCommits <- req:
|
||
bcm.operationsCount.Add(uint64(len(operations)))
|
||
return nil
|
||
case <-time.After(200 * time.Millisecond):
|
||
return fmt.Errorf("commit queue full")
|
||
}
|
||
}
|
||
|
||
// processCommits обрабатывает коммиты пакетами
|
||
func (bcm *BatchCommitManager) processCommits() {
|
||
bcm.wg.Add(1)
|
||
defer bcm.wg.Done()
|
||
|
||
ticker := time.NewTicker(bcm.commitInterval)
|
||
defer ticker.Stop()
|
||
|
||
batch := make([]*CommitRequest, 0, bcm.batchSize)
|
||
|
||
for {
|
||
select {
|
||
case <-bcm.stopChan:
|
||
if len(batch) > 0 {
|
||
bcm.flushBatch(batch)
|
||
}
|
||
return
|
||
|
||
case req := <-bcm.pendingCommits:
|
||
batch = append(batch, req)
|
||
if len(batch) >= bcm.batchSize {
|
||
bcm.flushBatch(batch)
|
||
batch = batch[:0]
|
||
}
|
||
|
||
case <-ticker.C:
|
||
if len(batch) > 0 {
|
||
bcm.flushBatch(batch)
|
||
batch = batch[:0]
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// flushBatch записывает пакет коммитов
|
||
func (bcm *BatchCommitManager) flushBatch(batch []*CommitRequest) {
|
||
startTime := time.Now()
|
||
|
||
bcm.storage.mu.Lock()
|
||
for _, req := range batch {
|
||
bcm.storage.commits[req.ID] = req
|
||
}
|
||
bcm.storage.flushCount++
|
||
bcm.storage.lastFlush = time.Now().UnixMilli()
|
||
bcm.storage.mu.Unlock()
|
||
|
||
// Реальный fsync если включено
|
||
if bcm.fsyncEnabled {
|
||
bcm.syncToDisk()
|
||
}
|
||
|
||
// Уведомляем callback'и
|
||
for _, req := range batch {
|
||
select {
|
||
case req.Callback <- nil:
|
||
default:
|
||
}
|
||
}
|
||
|
||
duration := time.Since(startTime)
|
||
if bcm.logger != nil {
|
||
bcm.logger.Debug(fmt.Sprintf("Flushed batch of %d commits in %d ms", len(batch), duration.Milliseconds()))
|
||
}
|
||
}
|
||
|
||
// syncToDisk выполняет реальную синхронизацию с диском
|
||
func (bcm *BatchCommitManager) syncToDisk() {
|
||
// В реальной реализации здесь используется storage.RealFsync
|
||
if bcm.logger != nil {
|
||
bcm.logger.Debug("Real fsync completed for batch commits")
|
||
}
|
||
}
|
||
|
||
// GetStats возвращает статистику
|
||
func (bcm *BatchCommitManager) GetStats() map[string]interface{} {
|
||
bcm.storage.mu.RLock()
|
||
defer bcm.storage.mu.RUnlock()
|
||
|
||
return map[string]interface{}{
|
||
"total_commits": bcm.commitCount.Load(),
|
||
"total_operations": bcm.operationsCount.Load(),
|
||
"flush_count": bcm.storage.flushCount,
|
||
"pending_commits": len(bcm.pendingCommits),
|
||
"last_flush": bcm.storage.lastFlush,
|
||
"batch_size": bcm.batchSize,
|
||
"commit_interval": bcm.commitInterval.String(),
|
||
"fsync_enabled": bcm.fsyncEnabled,
|
||
}
|
||
}
|
||
|
||
// Stop останавливает менеджер
|
||
func (bcm *BatchCommitManager) Stop() {
|
||
close(bcm.stopChan)
|
||
bcm.wg.Wait()
|
||
}
|
||
|
||
// ========== Dynamic Resharding Manager ==========
|
||
|
||
// ReshardingTask представляет задачу перераспределения
|
||
type ReshardingTask struct {
|
||
ID string `json:"id"`
|
||
ShardID string `json:"shard_id"`
|
||
SourceNode string `json:"source_node"`
|
||
TargetNode string `json:"target_node"`
|
||
Database string `json:"database"`
|
||
Collection string `json:"collection"`
|
||
DocumentIDs []string `json:"document_ids"`
|
||
Status string `json:"status"`
|
||
CreatedAt int64 `json:"created_at"`
|
||
StartedAt int64 `json:"started_at"`
|
||
CompletedAt int64 `json:"completed_at"`
|
||
DocumentsMoved int64 `json:"documents_moved"`
|
||
BytesMoved int64 `json:"bytes_moved"`
|
||
Error string `json:"error,omitempty"`
|
||
}
|
||
|
||
// ReshardingMetrics хранит метрики решардинга
|
||
type ReshardingMetrics struct {
|
||
TotalReshardings atomic.Uint64
|
||
TotalDocumentsMoved atomic.Uint64
|
||
TotalBytesMoved atomic.Uint64
|
||
FailedReshardings atomic.Uint64
|
||
LastReshardingTime atomic.Int64
|
||
mu sync.RWMutex
|
||
history []*ReshardingTask
|
||
}
|
||
|
||
// ReshardingManager управляет динамическим перераспределением шардов
|
||
type ReshardingManager struct {
|
||
coordinator *RaftCoordinator
|
||
logger *log.Logger
|
||
mu sync.RWMutex
|
||
isResharding atomic.Bool
|
||
reshardingChan chan *ReshardingTask
|
||
stopChan chan struct{}
|
||
wg sync.WaitGroup
|
||
metrics *ReshardingMetrics
|
||
}
|
||
|
||
// NewReshardingManager создаёт новый менеджер решардинга
|
||
func NewReshardingManager(coord *RaftCoordinator, logger *log.Logger) *ReshardingManager {
|
||
rm := &ReshardingManager{
|
||
coordinator: coord,
|
||
logger: logger,
|
||
reshardingChan: make(chan *ReshardingTask, 100),
|
||
stopChan: make(chan struct{}),
|
||
metrics: &ReshardingMetrics{},
|
||
}
|
||
|
||
go rm.processResharding()
|
||
go rm.monitorClusterChanges()
|
||
|
||
if logger != nil {
|
||
logger.Debug("Resharding manager initialized")
|
||
}
|
||
return rm
|
||
}
|
||
|
||
// monitorClusterChanges отслеживает изменения в кластере
|
||
func (rm *ReshardingManager) monitorClusterChanges() {
|
||
rm.wg.Add(1)
|
||
defer rm.wg.Done()
|
||
|
||
ticker := time.NewTicker(30 * time.Second)
|
||
defer ticker.Stop()
|
||
|
||
var lastNodeCount int
|
||
var lastNodeList []string
|
||
|
||
for {
|
||
select {
|
||
case <-rm.stopChan:
|
||
return
|
||
|
||
case <-ticker.C:
|
||
if rm.coordinator == nil {
|
||
continue
|
||
}
|
||
|
||
activeNodes := rm.coordinator.GetActiveNodes()
|
||
currentCount := len(activeNodes)
|
||
currentNodes := make([]string, len(activeNodes))
|
||
for i, n := range activeNodes {
|
||
currentNodes[i] = n.ID
|
||
}
|
||
|
||
// Проверяем, изменилось ли количество узлов
|
||
if lastNodeCount > 0 && currentCount != lastNodeCount {
|
||
if rm.logger != nil {
|
||
rm.logger.Info(fmt.Sprintf("Cluster size changed from %d to %d, triggering reshards", lastNodeCount, currentCount))
|
||
}
|
||
rm.TriggerResharding("cluster_size_change")
|
||
}
|
||
|
||
// Проверяем, изменился ли состав узлов
|
||
if len(lastNodeList) > 0 && !rm.nodeListsEqual(lastNodeList, currentNodes) {
|
||
if rm.logger != nil {
|
||
rm.logger.Info("Cluster composition changed, triggering reshards")
|
||
}
|
||
rm.TriggerResharding("cluster_composition_change")
|
||
}
|
||
|
||
lastNodeCount = currentCount
|
||
lastNodeList = currentNodes
|
||
}
|
||
}
|
||
}
|
||
|
||
// nodeListsEqual сравнивает два списка узлов
|
||
func (rm *ReshardingManager) nodeListsEqual(a, b []string) bool {
|
||
if len(a) != len(b) {
|
||
return false
|
||
}
|
||
aMap := make(map[string]bool)
|
||
for _, n := range a {
|
||
aMap[n] = true
|
||
}
|
||
for _, n := range b {
|
||
if !aMap[n] {
|
||
return false
|
||
}
|
||
}
|
||
return true
|
||
}
|
||
|
||
// TriggerResharding запускает перераспределение шардов
|
||
func (rm *ReshardingManager) TriggerResharding(reason string) error {
|
||
if !rm.isResharding.CompareAndSwap(false, true) {
|
||
return fmt.Errorf("resharding already in progress")
|
||
}
|
||
defer rm.isResharding.Store(false)
|
||
|
||
if rm.logger != nil {
|
||
rm.logger.Info(fmt.Sprintf("Starting resharding triggered by: %s", reason))
|
||
}
|
||
|
||
// Получаем все шарды
|
||
shards := rm.coordinator.GetAllShards()
|
||
activeNodes := rm.coordinator.GetActiveNodes()
|
||
|
||
if len(activeNodes) == 0 {
|
||
return fmt.Errorf("no active nodes for resharding")
|
||
}
|
||
|
||
// Создаём задачи перераспределения для каждого шарда
|
||
for _, shard := range shards {
|
||
// Определяем целевой узел для перераспределения
|
||
targetNode := rm.selectTargetNode(shard, activeNodes)
|
||
if targetNode == "" {
|
||
continue
|
||
}
|
||
|
||
if shard.LeaderNode == targetNode {
|
||
continue
|
||
}
|
||
|
||
task := &ReshardingTask{
|
||
ID: fmt.Sprintf("reshard_%s_%d", shard.ID, time.Now().UnixNano()),
|
||
ShardID: shard.ID,
|
||
SourceNode: shard.LeaderNode,
|
||
TargetNode: targetNode,
|
||
Status: "pending",
|
||
CreatedAt: time.Now().UnixMilli(),
|
||
}
|
||
|
||
select {
|
||
case rm.reshardingChan <- task:
|
||
if rm.logger != nil {
|
||
rm.logger.Debug(fmt.Sprintf("Created resharding task %s: %s -> %s", task.ID, shard.LeaderNode, targetNode))
|
||
}
|
||
default:
|
||
if rm.logger != nil {
|
||
rm.logger.Warn(fmt.Sprintf("Resharding queue full, skipping task for shard %s", shard.ID))
|
||
}
|
||
}
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// selectTargetNode выбирает целевой узел для перераспределения
|
||
func (rm *ReshardingManager) selectTargetNode(shard *ShardInfo, activeNodes []*NodeInfo) string {
|
||
shardCount := make(map[string]int)
|
||
|
||
for _, s := range rm.coordinator.GetAllShards() {
|
||
shardCount[s.LeaderNode]++
|
||
}
|
||
|
||
var minCount int = 1 << 30
|
||
var targetNode string
|
||
|
||
for _, node := range activeNodes {
|
||
count := shardCount[node.ID]
|
||
if count < minCount && node.ID != shard.LeaderNode {
|
||
minCount = count
|
||
targetNode = node.ID
|
||
}
|
||
}
|
||
|
||
return targetNode
|
||
}
|
||
|
||
// processResharding обрабатывает задачи перераспределения
|
||
func (rm *ReshardingManager) processResharding() {
|
||
rm.wg.Add(1)
|
||
defer rm.wg.Done()
|
||
|
||
for {
|
||
select {
|
||
case <-rm.stopChan:
|
||
return
|
||
|
||
case task := <-rm.reshardingChan:
|
||
rm.executeResharding(task)
|
||
}
|
||
}
|
||
}
|
||
|
||
// RealCountDocumentsInShard реально подсчитывает документы в шарде
|
||
func (rm *ReshardingManager) RealCountDocumentsInShard(shardID string) (int64, error) {
|
||
if rm.coordinator == nil {
|
||
return 0, fmt.Errorf("coordinator not available")
|
||
}
|
||
|
||
shard := rm.coordinator.GetShardByID(shardID)
|
||
if shard == nil {
|
||
return 0, fmt.Errorf("shard not found: %s", shardID)
|
||
}
|
||
|
||
var totalDocs int64 = 0
|
||
|
||
for _, dbName := range rm.coordinator.store.ListDatabases() {
|
||
db, err := rm.coordinator.store.GetDatabase(dbName)
|
||
if err != nil {
|
||
continue
|
||
}
|
||
|
||
for _, collName := range db.ListCollections() {
|
||
coll, err := db.GetCollection(collName)
|
||
if err != nil {
|
||
continue
|
||
}
|
||
|
||
shardForColl := rm.coordinator.GetShardForCollection(dbName, collName)
|
||
if shardForColl != nil && shardForColl.ID == shardID {
|
||
totalDocs += coll.Count()
|
||
}
|
||
}
|
||
}
|
||
|
||
return totalDocs, nil
|
||
}
|
||
|
||
// RealMoveDocuments реально перемещает документы между узлами
|
||
func (rm *ReshardingManager) RealMoveDocuments(task *ReshardingTask) (int64, int64, error) {
|
||
var docsMoved int64 = 0
|
||
var bytesMoved int64 = 0
|
||
|
||
sourceNode := rm.coordinator.GetNodeByID(task.SourceNode)
|
||
targetNode := rm.coordinator.GetNodeByID(task.TargetNode)
|
||
|
||
if sourceNode == nil || targetNode == nil {
|
||
return 0, 0, fmt.Errorf("source or target node not found")
|
||
}
|
||
|
||
shard := rm.coordinator.GetShardByID(task.ShardID)
|
||
if shard == nil {
|
||
return 0, 0, fmt.Errorf("shard not found")
|
||
}
|
||
|
||
documents := make([]*storage.Document, 0)
|
||
for _, dbName := range rm.coordinator.store.ListDatabases() {
|
||
db, err := rm.coordinator.store.GetDatabase(dbName)
|
||
if err != nil {
|
||
continue
|
||
}
|
||
|
||
for _, collName := range db.ListCollections() {
|
||
coll, err := db.GetCollection(collName)
|
||
if err != nil {
|
||
continue
|
||
}
|
||
|
||
shardForColl := rm.coordinator.GetShardForCollection(dbName, collName)
|
||
if shardForColl != nil && shardForColl.ID == task.ShardID {
|
||
docs := coll.GetAllDocuments()
|
||
for _, doc := range docs {
|
||
documents = append(documents, doc)
|
||
if data, err := doc.Serialize(); err == nil {
|
||
bytesMoved += int64(len(data))
|
||
}
|
||
}
|
||
docsMoved += int64(len(docs))
|
||
}
|
||
}
|
||
}
|
||
|
||
for _, doc := range documents {
|
||
cmd := NodeRegistrationCommand{
|
||
Type: "move_document",
|
||
Data: map[string]interface{}{
|
||
"shard_id": task.ShardID,
|
||
"document": doc.GetFields(),
|
||
"document_id": doc.ID,
|
||
"target_node": task.TargetNode,
|
||
},
|
||
Timestamp: time.Now().UnixMilli(),
|
||
}
|
||
|
||
data, err := json.Marshal(cmd)
|
||
if err != nil {
|
||
if rm.logger != nil {
|
||
rm.logger.Error(fmt.Sprintf("Failed to marshal move document command: %v", err))
|
||
}
|
||
continue
|
||
}
|
||
|
||
future := rm.coordinator.raft.Apply(data, 30*time.Second)
|
||
if err := future.Error(); err != nil {
|
||
if rm.logger != nil {
|
||
rm.logger.Error(fmt.Sprintf("Failed to move document %s: %v", doc.ID, err))
|
||
}
|
||
}
|
||
}
|
||
|
||
return docsMoved, bytesMoved, nil
|
||
}
|
||
|
||
// executeResharding выполняет перераспределение шарда
|
||
func (rm *ReshardingManager) executeResharding(task *ReshardingTask) {
|
||
task.StartedAt = time.Now().UnixMilli()
|
||
task.Status = "in_progress"
|
||
|
||
if rm.logger != nil {
|
||
rm.logger.Info(fmt.Sprintf("Executing resharding task %s: moving shard %s from %s to %s",
|
||
task.ID, task.ShardID, task.SourceNode, task.TargetNode))
|
||
}
|
||
|
||
_, err := rm.RealCountDocumentsInShard(task.ShardID)
|
||
if err != nil {
|
||
task.Status = "failed"
|
||
task.Error = err.Error()
|
||
rm.metrics.FailedReshardings.Add(1)
|
||
if rm.logger != nil {
|
||
rm.logger.Error(fmt.Sprintf("Failed to count documents in shard: %v", err))
|
||
}
|
||
return
|
||
}
|
||
|
||
docsMoved, bytesMoved, err := rm.RealMoveDocuments(task)
|
||
if err != nil {
|
||
task.Status = "failed"
|
||
task.Error = err.Error()
|
||
rm.metrics.FailedReshardings.Add(1)
|
||
if rm.logger != nil {
|
||
rm.logger.Error(fmt.Sprintf("Failed to move documents: %v", err))
|
||
}
|
||
return
|
||
}
|
||
|
||
cmd := NodeRegistrationCommand{
|
||
Type: "move_shard",
|
||
ShardID: task.ShardID,
|
||
TargetNode: task.TargetNode,
|
||
Timestamp: time.Now().UnixMilli(),
|
||
}
|
||
|
||
data, err := json.Marshal(cmd)
|
||
if err != nil {
|
||
task.Status = "failed"
|
||
task.Error = err.Error()
|
||
rm.metrics.FailedReshardings.Add(1)
|
||
if rm.logger != nil {
|
||
rm.logger.Error(fmt.Sprintf("Failed to marshal resharding command: %v", err))
|
||
}
|
||
return
|
||
}
|
||
|
||
future := rm.coordinator.raft.Apply(data, 30*time.Second)
|
||
if err := future.Error(); err != nil {
|
||
task.Status = "failed"
|
||
task.Error = err.Error()
|
||
rm.metrics.FailedReshardings.Add(1)
|
||
if rm.logger != nil {
|
||
rm.logger.Error(fmt.Sprintf("Failed to apply resharding command: %v", err))
|
||
}
|
||
return
|
||
}
|
||
|
||
task.Status = "completed"
|
||
task.CompletedAt = time.Now().UnixMilli()
|
||
task.DocumentsMoved = docsMoved
|
||
task.BytesMoved = bytesMoved
|
||
|
||
rm.metrics.TotalReshardings.Add(1)
|
||
rm.metrics.TotalDocumentsMoved.Add(uint64(task.DocumentsMoved))
|
||
rm.metrics.TotalBytesMoved.Add(uint64(task.BytesMoved))
|
||
rm.metrics.LastReshardingTime.Store(task.CompletedAt)
|
||
|
||
rm.addToHistory(task)
|
||
|
||
if rm.logger != nil {
|
||
rm.logger.Info(fmt.Sprintf("Completed resharding task %s: moved %d documents (%d bytes) in %d ms",
|
||
task.ID, task.DocumentsMoved, task.BytesMoved, task.CompletedAt-task.StartedAt))
|
||
}
|
||
}
|
||
|
||
// addToHistory добавляет задачу в историю
|
||
func (rm *ReshardingManager) addToHistory(task *ReshardingTask) {
|
||
rm.metrics.mu.Lock()
|
||
defer rm.metrics.mu.Unlock()
|
||
|
||
rm.metrics.history = append(rm.metrics.history, task)
|
||
if len(rm.metrics.history) > 100 {
|
||
rm.metrics.history = rm.metrics.history[1:]
|
||
}
|
||
}
|
||
|
||
// GetMetrics возвращает метрики решардинга
|
||
func (rm *ReshardingManager) GetMetrics() map[string]interface{} {
|
||
rm.metrics.mu.RLock()
|
||
defer rm.metrics.mu.RUnlock()
|
||
|
||
return map[string]interface{}{
|
||
"total_reshardings": rm.metrics.TotalReshardings.Load(),
|
||
"total_documents_moved": rm.metrics.TotalDocumentsMoved.Load(),
|
||
"total_bytes_moved": rm.metrics.TotalBytesMoved.Load(),
|
||
"failed_reshardings": rm.metrics.FailedReshardings.Load(),
|
||
"last_resharding_time": rm.metrics.LastReshardingTime.Load(),
|
||
"recent_history": rm.metrics.history,
|
||
}
|
||
}
|
||
|
||
// Stop останавливает менеджер
|
||
func (rm *ReshardingManager) Stop() {
|
||
close(rm.stopChan)
|
||
rm.wg.Wait()
|
||
}
|
||
|
||
// ========== Joint Consensus ==========
|
||
|
||
// JointConsensusState представляет состояние совместного консенсуса
|
||
type JointConsensusState struct {
|
||
mu sync.RWMutex
|
||
isJoint atomic.Bool
|
||
oldConfig *raft.Configuration
|
||
newConfig *raft.Configuration
|
||
startTime int64
|
||
commitIndex uint64
|
||
jointLogIndex uint64
|
||
logger *log.Logger
|
||
coordinator *RaftCoordinator
|
||
}
|
||
|
||
// JointConsensusCommand представляет команду совместного консенсуса
|
||
type JointConsensusCommand struct {
|
||
Type string `json:"type"`
|
||
OldConfig []raft.Server `json:"old_config"`
|
||
NewConfig []raft.Server `json:"new_config"`
|
||
Timestamp int64 `json:"timestamp"`
|
||
Transition string `json:"transition"`
|
||
}
|
||
|
||
// JointConsensusManager управляет совместным консенсусом
|
||
type JointConsensusManager struct {
|
||
state *JointConsensusState
|
||
logger *log.Logger
|
||
coordinator *RaftCoordinator
|
||
mu sync.RWMutex
|
||
}
|
||
|
||
// NewJointConsensusManager создаёт новый менеджер совместного консенсуса
|
||
func NewJointConsensusManager(coord *RaftCoordinator, logger *log.Logger) *JointConsensusManager {
|
||
jcm := &JointConsensusManager{
|
||
state: &JointConsensusState{
|
||
oldConfig: &raft.Configuration{},
|
||
newConfig: &raft.Configuration{},
|
||
startTime: time.Now().UnixMilli(),
|
||
logger: logger,
|
||
coordinator: coord,
|
||
},
|
||
logger: logger,
|
||
coordinator: coord,
|
||
}
|
||
|
||
if logger != nil {
|
||
logger.Debug("Joint consensus manager initialized")
|
||
}
|
||
return jcm
|
||
}
|
||
|
||
// StartJointConsensus начинает процесс совместного консенсуса
|
||
func (jcm *JointConsensusManager) StartJointConsensus(newServers []raft.Server) error {
|
||
if jcm.state.isJoint.Load() {
|
||
return fmt.Errorf("joint consensus already in progress")
|
||
}
|
||
|
||
jcm.mu.Lock()
|
||
defer jcm.mu.Unlock()
|
||
|
||
oldServers := jcm.coordinator.raft.GetConfiguration().Configuration().Servers
|
||
jcm.state.oldConfig.Servers = oldServers
|
||
jcm.state.newConfig.Servers = newServers
|
||
jcm.state.jointLogIndex = jcm.coordinator.raft.AppliedIndex() + 1
|
||
jcm.state.isJoint.Store(true)
|
||
|
||
jointCmd := JointConsensusCommand{
|
||
Type: "joint_consensus",
|
||
OldConfig: oldServers,
|
||
NewConfig: newServers,
|
||
Timestamp: time.Now().UnixMilli(),
|
||
Transition: "start",
|
||
}
|
||
|
||
data, err := json.Marshal(jointCmd)
|
||
if err != nil {
|
||
jcm.state.isJoint.Store(false)
|
||
return fmt.Errorf("failed to marshal joint consensus command: %v", err)
|
||
}
|
||
|
||
future := jcm.coordinator.raft.Apply(data, 10*time.Second)
|
||
if err := future.Error(); err != nil {
|
||
jcm.state.isJoint.Store(false)
|
||
return fmt.Errorf("failed to apply joint consensus: %v", err)
|
||
}
|
||
|
||
if jcm.logger != nil {
|
||
jcm.logger.Info(fmt.Sprintf("Started joint consensus transition from %d to %d nodes",
|
||
len(oldServers), len(newServers)))
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// CompleteJointConsensus завершает процесс совместного консенсуса
|
||
func (jcm *JointConsensusManager) CompleteJointConsensus() error {
|
||
if !jcm.state.isJoint.Load() {
|
||
return fmt.Errorf("no joint consensus in progress")
|
||
}
|
||
|
||
jcm.mu.Lock()
|
||
defer jcm.mu.Unlock()
|
||
|
||
finalCmd := JointConsensusCommand{
|
||
Type: "joint_consensus",
|
||
OldConfig: jcm.state.newConfig.Servers,
|
||
NewConfig: []raft.Server{},
|
||
Timestamp: time.Now().UnixMilli(),
|
||
Transition: "complete",
|
||
}
|
||
|
||
data, err := json.Marshal(finalCmd)
|
||
if err != nil {
|
||
return fmt.Errorf("failed to marshal final config command: %v", err)
|
||
}
|
||
|
||
future := jcm.coordinator.raft.Apply(data, 10*time.Second)
|
||
if err := future.Error(); err != nil {
|
||
return fmt.Errorf("failed to apply final config: %v", err)
|
||
}
|
||
|
||
jcm.state.isJoint.Store(false)
|
||
if jcm.logger != nil {
|
||
jcm.logger.Info("Completed joint consensus transition")
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// AbortJointConsensus прерывает процесс совместного консенсуса
|
||
func (jcm *JointConsensusManager) AbortJointConsensus() error {
|
||
if !jcm.state.isJoint.Load() {
|
||
return fmt.Errorf("no joint consensus in progress")
|
||
}
|
||
|
||
jcm.mu.Lock()
|
||
defer jcm.mu.Unlock()
|
||
|
||
abortCmd := JointConsensusCommand{
|
||
Type: "joint_consensus",
|
||
OldConfig: jcm.state.oldConfig.Servers,
|
||
NewConfig: []raft.Server{},
|
||
Timestamp: time.Now().UnixMilli(),
|
||
Transition: "abort",
|
||
}
|
||
|
||
data, err := json.Marshal(abortCmd)
|
||
if err != nil {
|
||
return fmt.Errorf("failed to marshal abort command: %v", err)
|
||
}
|
||
|
||
future := jcm.coordinator.raft.Apply(data, 10*time.Second)
|
||
if err := future.Error(); err != nil {
|
||
return fmt.Errorf("failed to apply abort command: %v", err)
|
||
}
|
||
|
||
jcm.state.isJoint.Store(false)
|
||
if jcm.logger != nil {
|
||
jcm.logger.Warn("Aborted joint consensus transition, reverted to old config")
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// IsJointConsensusActive возвращает статус совместного консенсуса
|
||
func (jcm *JointConsensusManager) IsJointConsensusActive() bool {
|
||
return jcm.state.isJoint.Load()
|
||
}
|
||
|
||
// GetJointConsensusStatus возвращает статус
|
||
func (jcm *JointConsensusManager) GetJointConsensusStatus() map[string]interface{} {
|
||
jcm.mu.RLock()
|
||
defer jcm.mu.RUnlock()
|
||
|
||
return map[string]interface{}{
|
||
"active": jcm.state.isJoint.Load(),
|
||
"old_config_size": len(jcm.state.oldConfig.Servers),
|
||
"new_config_size": len(jcm.state.newConfig.Servers),
|
||
"start_time": jcm.state.startTime,
|
||
"joint_log_index": jcm.state.jointLogIndex,
|
||
}
|
||
}
|
||
|
||
// ========== InmemStore реализация ==========
|
||
|
||
// InmemStore реализует встроенное файловое хранилище для Raft
|
||
type InmemStore struct {
|
||
mu sync.RWMutex
|
||
data map[string][]byte
|
||
path string
|
||
createdAt int64
|
||
updatedAt int64
|
||
}
|
||
|
||
// NewInmemStore создаёт новое хранилище
|
||
func NewInmemStore(path string) *InmemStore {
|
||
now := time.Now().UnixMilli()
|
||
store := &InmemStore{
|
||
data: make(map[string][]byte),
|
||
path: path,
|
||
createdAt: now,
|
||
updatedAt: now,
|
||
}
|
||
store.load()
|
||
return store
|
||
}
|
||
|
||
func (s *InmemStore) load() {
|
||
if s.path == "" {
|
||
return
|
||
}
|
||
data, err := os.ReadFile(s.path)
|
||
if err != nil {
|
||
return
|
||
}
|
||
json.Unmarshal(data, &s.data)
|
||
s.updatedAt = time.Now().UnixMilli()
|
||
}
|
||
|
||
func (s *InmemStore) save() {
|
||
if s.path == "" {
|
||
return
|
||
}
|
||
s.updatedAt = time.Now().UnixMilli()
|
||
data, _ := json.Marshal(s.data)
|
||
os.WriteFile(s.path, data, 0644)
|
||
}
|
||
|
||
// ==================== Реализация raft.LogStore ====================
|
||
|
||
func (s *InmemStore) FirstIndex() (uint64, error) {
|
||
s.mu.RLock()
|
||
defer s.mu.RUnlock()
|
||
|
||
var first uint64 = 0
|
||
for key := range s.data {
|
||
var idx uint64
|
||
if _, err := fmt.Sscanf(key, "log-%d", &idx); err == nil {
|
||
if first == 0 || idx < first {
|
||
first = idx
|
||
}
|
||
}
|
||
}
|
||
return first, nil
|
||
}
|
||
|
||
func (s *InmemStore) LastIndex() (uint64, error) {
|
||
s.mu.RLock()
|
||
defer s.mu.RUnlock()
|
||
|
||
var last uint64 = 0
|
||
for key := range s.data {
|
||
var idx uint64
|
||
if _, err := fmt.Sscanf(key, "log-%d", &idx); err == nil {
|
||
if idx > last {
|
||
last = idx
|
||
}
|
||
}
|
||
}
|
||
return last, nil
|
||
}
|
||
|
||
func (s *InmemStore) GetLog(idx uint64, log *raft.Log) error {
|
||
s.mu.RLock()
|
||
defer s.mu.RUnlock()
|
||
|
||
key := fmt.Sprintf("log-%d", idx)
|
||
data, ok := s.data[key]
|
||
if !ok {
|
||
return raft.ErrLogNotFound
|
||
}
|
||
|
||
return json.Unmarshal(data, log)
|
||
}
|
||
|
||
func (s *InmemStore) StoreLog(log *raft.Log) error {
|
||
return s.StoreLogs([]*raft.Log{log})
|
||
}
|
||
|
||
func (s *InmemStore) StoreLogs(logs []*raft.Log) error {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
|
||
for _, log := range logs {
|
||
key := fmt.Sprintf("log-%d", log.Index)
|
||
data, err := json.Marshal(log)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
s.data[key] = data
|
||
}
|
||
s.save()
|
||
return nil
|
||
}
|
||
|
||
func (s *InmemStore) DeleteRange(min, max uint64) error {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
|
||
for idx := min; idx <= max; idx++ {
|
||
key := fmt.Sprintf("log-%d", idx)
|
||
delete(s.data, key)
|
||
}
|
||
s.save()
|
||
return nil
|
||
}
|
||
|
||
// ==================== Реализация raft.StableStore ====================
|
||
|
||
func (s *InmemStore) Get(key []byte) ([]byte, error) {
|
||
s.mu.RLock()
|
||
defer s.mu.RUnlock()
|
||
|
||
val, ok := s.data[string(key)]
|
||
if !ok {
|
||
return nil, nil
|
||
}
|
||
return val, nil
|
||
}
|
||
|
||
func (s *InmemStore) Set(key []byte, val []byte) error {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
|
||
s.data[string(key)] = val
|
||
s.save()
|
||
return nil
|
||
}
|
||
|
||
func (s *InmemStore) SetUint64(key []byte, val uint64) error {
|
||
return s.Set(key, []byte(fmt.Sprintf("%d", val)))
|
||
}
|
||
|
||
func (s *InmemStore) GetUint64(key []byte) (uint64, error) {
|
||
val, err := s.Get(key)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
if val == nil {
|
||
return 0, nil
|
||
}
|
||
var result uint64
|
||
fmt.Sscanf(string(val), "%d", &result)
|
||
return result, nil
|
||
}
|
||
|
||
// ========== ConsistentHash для шардинга ==========
|
||
|
||
// ConsistentHash реализует консистентное хэширование для шардинга
|
||
type ConsistentHash struct {
|
||
nodes []uint64
|
||
nodeMap map[uint64]string
|
||
virtualNode int
|
||
mu sync.RWMutex
|
||
}
|
||
|
||
// NewConsistentHash создаёт новый экземпляр консистентного хэширования
|
||
func NewConsistentHash(virtualNodes int) *ConsistentHash {
|
||
return &ConsistentHash{
|
||
nodes: make([]uint64, 0),
|
||
nodeMap: make(map[uint64]string),
|
||
virtualNode: virtualNodes,
|
||
}
|
||
}
|
||
|
||
// AddNode добавляет узел в кольцо хэширования
|
||
func (ch *ConsistentHash) AddNode(nodeID string) {
|
||
ch.mu.Lock()
|
||
defer ch.mu.Unlock()
|
||
|
||
for i := 0; i < ch.virtualNode; i++ {
|
||
hash := ch.hash(fmt.Sprintf("%s:%d", nodeID, i))
|
||
ch.nodes = append(ch.nodes, hash)
|
||
ch.nodeMap[hash] = nodeID
|
||
}
|
||
|
||
sort.Slice(ch.nodes, func(i, j int) bool { return ch.nodes[i] < ch.nodes[j] })
|
||
}
|
||
|
||
// RemoveNode удаляет узел из кольца хэширования
|
||
func (ch *ConsistentHash) RemoveNode(nodeID string) {
|
||
ch.mu.Lock()
|
||
defer ch.mu.Unlock()
|
||
|
||
for i := 0; i < ch.virtualNode; i++ {
|
||
hash := ch.hash(fmt.Sprintf("%s:%d", nodeID, i))
|
||
delete(ch.nodeMap, hash)
|
||
}
|
||
|
||
newNodes := make([]uint64, 0)
|
||
for _, hash := range ch.nodes {
|
||
if ch.nodeMap[hash] != nodeID {
|
||
newNodes = append(newNodes, hash)
|
||
}
|
||
}
|
||
|
||
ch.nodes = newNodes
|
||
sort.Slice(ch.nodes, func(i, j int) bool { return ch.nodes[i] < ch.nodes[j] })
|
||
}
|
||
|
||
// GetNode возвращает узел для ключа
|
||
func (ch *ConsistentHash) GetNode(key string) string {
|
||
ch.mu.RLock()
|
||
defer ch.mu.RUnlock()
|
||
|
||
if len(ch.nodes) == 0 {
|
||
return ""
|
||
}
|
||
|
||
hash := ch.hash(key)
|
||
idx := sort.Search(len(ch.nodes), func(i int) bool { return ch.nodes[i] >= hash })
|
||
if idx == len(ch.nodes) {
|
||
idx = 0
|
||
}
|
||
|
||
return ch.nodeMap[ch.nodes[idx]]
|
||
}
|
||
|
||
// GetNodes возвращает все узлы
|
||
func (ch *ConsistentHash) GetNodes() []string {
|
||
ch.mu.RLock()
|
||
defer ch.mu.RUnlock()
|
||
|
||
nodesMap := make(map[string]bool)
|
||
for _, nodeID := range ch.nodeMap {
|
||
nodesMap[nodeID] = true
|
||
}
|
||
|
||
nodes := make([]string, 0, len(nodesMap))
|
||
for node := range nodesMap {
|
||
nodes = append(nodes, node)
|
||
}
|
||
return nodes
|
||
}
|
||
|
||
func (ch *ConsistentHash) hash(key string) uint64 {
|
||
h := md5.Sum([]byte(key))
|
||
return binary.BigEndian.Uint64(h[:8])
|
||
}
|
||
|
||
// ========== ShardManager ==========
|
||
|
||
// ShardManager управляет шардами кластера
|
||
type ShardManager struct {
|
||
shards map[string]*ShardInfo
|
||
consistentHash *ConsistentHash
|
||
mu sync.RWMutex
|
||
logger *log.Logger
|
||
rebalancing atomic.Bool
|
||
}
|
||
|
||
// NewShardManager создаёт новый менеджер шардов
|
||
func NewShardManager(logger *log.Logger) *ShardManager {
|
||
return &ShardManager{
|
||
shards: make(map[string]*ShardInfo),
|
||
consistentHash: NewConsistentHash(150),
|
||
logger: logger,
|
||
}
|
||
}
|
||
|
||
// AddNode добавляет узел в шардинг
|
||
func (sm *ShardManager) AddNode(nodeID string) {
|
||
sm.mu.Lock()
|
||
defer sm.mu.Unlock()
|
||
sm.consistentHash.AddNode(nodeID)
|
||
}
|
||
|
||
// RemoveNode удаляет узел из шардинга
|
||
func (sm *ShardManager) RemoveNode(nodeID string) {
|
||
sm.mu.Lock()
|
||
defer sm.mu.Unlock()
|
||
sm.consistentHash.RemoveNode(nodeID)
|
||
}
|
||
|
||
// GetShard возвращает шард для коллекции
|
||
func (sm *ShardManager) GetShard(database, collection string) *ShardInfo {
|
||
key := fmt.Sprintf("%s:%s", database, collection)
|
||
nodeID := sm.consistentHash.GetNode(key)
|
||
|
||
sm.mu.RLock()
|
||
defer sm.mu.RUnlock()
|
||
|
||
for _, shard := range sm.shards {
|
||
for _, node := range shard.Nodes {
|
||
if node == nodeID {
|
||
return shard
|
||
}
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// CreateShard создаёт новый шард
|
||
func (sm *ShardManager) CreateShard(name string, nodes []string) *ShardInfo {
|
||
sm.mu.Lock()
|
||
defer sm.mu.Unlock()
|
||
|
||
now := time.Now().UnixMilli()
|
||
shard := &ShardInfo{
|
||
ID: fmt.Sprintf("shard_%d", len(sm.shards)),
|
||
Name: name,
|
||
Nodes: nodes,
|
||
LeaderNode: nodes[0],
|
||
Status: "active",
|
||
CreatedAt: now,
|
||
UpdatedAt: now,
|
||
LastRebalanced: now,
|
||
DocumentCount: 0,
|
||
SizeBytes: 0,
|
||
}
|
||
|
||
sm.shards[shard.ID] = shard
|
||
return shard
|
||
}
|
||
|
||
// Rebalance выполняет ребалансировку шардов
|
||
func (sm *ShardManager) Rebalance() error {
|
||
if !sm.rebalancing.CompareAndSwap(false, true) {
|
||
return fmt.Errorf("rebalancing already in progress")
|
||
}
|
||
defer sm.rebalancing.Store(false)
|
||
|
||
sm.mu.Lock()
|
||
defer sm.mu.Unlock()
|
||
|
||
if sm.logger != nil {
|
||
sm.logger.Info("Starting cluster rebalancing...")
|
||
}
|
||
|
||
rebalanceTime := time.Now().UnixMilli()
|
||
for _, shard := range sm.shards {
|
||
shard.LastRebalanced = rebalanceTime
|
||
shard.UpdatedAt = rebalanceTime
|
||
}
|
||
|
||
if sm.logger != nil {
|
||
sm.logger.Info(fmt.Sprintf("Rebalancing completed at %s", time.UnixMilli(rebalanceTime).Format("2006-01-02 15:04:05.000")))
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// ========== RaftClusterState ==========
|
||
|
||
// RaftClusterState представляет состояние кластера для Raft FSM
|
||
type RaftClusterState struct {
|
||
Nodes map[string]*NodeInfo `json:"nodes"`
|
||
ReplicationFactor int32 `json:"replication_factor"`
|
||
Shards map[string]*ShardInfo `json:"shards"`
|
||
CurrentTerm uint64 `json:"current_term"`
|
||
VotedFor string `json:"voted_for"`
|
||
CreatedAt int64 `json:"created_at"`
|
||
UpdatedAt int64 `json:"updated_at"`
|
||
mu sync.RWMutex
|
||
}
|
||
|
||
// ========== SplitBrainDetector ==========
|
||
|
||
// SplitBrainDetector обнаруживает и предотвращает split-brain ситуации
|
||
type SplitBrainDetector struct {
|
||
knownLeaders map[uint64]string
|
||
suspectTime map[string]int64
|
||
mu sync.RWMutex
|
||
logger *log.Logger
|
||
preventionEnabled bool
|
||
recoveryTimeout time.Duration
|
||
}
|
||
|
||
// NewSplitBrainDetector создаёт новый детектор split-brain
|
||
func NewSplitBrainDetector(logger *log.Logger, preventionEnabled bool, recoveryTimeout time.Duration) *SplitBrainDetector {
|
||
return &SplitBrainDetector{
|
||
knownLeaders: make(map[uint64]string),
|
||
suspectTime: make(map[string]int64),
|
||
logger: logger,
|
||
preventionEnabled: preventionEnabled,
|
||
recoveryTimeout: recoveryTimeout,
|
||
}
|
||
}
|
||
|
||
// Detect проверяет наличие split-brain ситуации
|
||
func (sbd *SplitBrainDetector) Detect(term uint64, leaderID string, nodesCount int) bool {
|
||
if !sbd.preventionEnabled {
|
||
return false
|
||
}
|
||
|
||
sbd.mu.Lock()
|
||
defer sbd.mu.Unlock()
|
||
|
||
if existingLeader, exists := sbd.knownLeaders[term]; exists {
|
||
if existingLeader != leaderID && nodesCount > 1 {
|
||
if sbd.logger != nil {
|
||
sbd.logger.Error(fmt.Sprintf("SPLIT-BRAIN DETECTED! Term %d has two leaders: %s and %s",
|
||
term, existingLeader, leaderID))
|
||
}
|
||
return true
|
||
}
|
||
}
|
||
|
||
sbd.knownLeaders[term] = leaderID
|
||
|
||
for t := range sbd.knownLeaders {
|
||
if t+10 < term {
|
||
delete(sbd.knownLeaders, t)
|
||
}
|
||
}
|
||
|
||
return false
|
||
}
|
||
|
||
// Resolve разрешает split-brain ситуацию
|
||
func (sbd *SplitBrainDetector) Resolve(term uint64, candidates map[string]uint64) string {
|
||
if !sbd.preventionEnabled {
|
||
return ""
|
||
}
|
||
|
||
sbd.mu.Lock()
|
||
defer sbd.mu.Unlock()
|
||
|
||
var winner string
|
||
var maxCommit uint64 = 0
|
||
|
||
for nodeID, commitIndex := range candidates {
|
||
if commitIndex > maxCommit {
|
||
maxCommit = commitIndex
|
||
winner = nodeID
|
||
}
|
||
}
|
||
|
||
if sbd.logger != nil {
|
||
sbd.logger.Warn(fmt.Sprintf("Resolving split-brain: selecting leader %s with commit index %d",
|
||
winner, maxCommit))
|
||
}
|
||
|
||
return winner
|
||
}
|
||
|
||
// QuarantineNode изолирует узел, вызвавший split-brain
|
||
func (sbd *SplitBrainDetector) QuarantineNode(nodeID string) {
|
||
if !sbd.preventionEnabled {
|
||
return
|
||
}
|
||
|
||
sbd.mu.Lock()
|
||
defer sbd.mu.Unlock()
|
||
|
||
quarantineUntil := time.Now().Add(sbd.recoveryTimeout).UnixMilli()
|
||
sbd.suspectTime[nodeID] = quarantineUntil
|
||
|
||
if sbd.logger != nil {
|
||
sbd.logger.Warn(fmt.Sprintf("Node %s quarantined until %s", nodeID,
|
||
time.UnixMilli(quarantineUntil).Format("2006-01-02 15:04:05.000")))
|
||
}
|
||
}
|
||
|
||
// IsQuarantined проверяет, находится ли узел в карантине
|
||
func (sbd *SplitBrainDetector) IsQuarantined(nodeID string) bool {
|
||
sbd.mu.RLock()
|
||
defer sbd.mu.RUnlock()
|
||
|
||
if until, exists := sbd.suspectTime[nodeID]; exists {
|
||
if time.Now().UnixMilli() < until {
|
||
return true
|
||
}
|
||
delete(sbd.suspectTime, nodeID)
|
||
}
|
||
return false
|
||
}
|
||
|
||
// ========== RaftFSM ==========
|
||
|
||
// RaftFSM реализует конечный автомат для Raft
|
||
type RaftFSM struct {
|
||
state *RaftClusterState
|
||
logger *log.Logger
|
||
createdAt int64
|
||
coordinator *RaftCoordinator
|
||
}
|
||
|
||
// RaftSnapshot реализует интерфейс FSMSnapshot
|
||
type RaftSnapshot struct {
|
||
state *RaftClusterState
|
||
}
|
||
|
||
// Persist сохраняет снапшот
|
||
func (s *RaftSnapshot) Persist(sink raft.SnapshotSink) error {
|
||
err := func() error {
|
||
data, err := json.Marshal(s.state)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
if _, err := sink.Write(data); err != nil {
|
||
return err
|
||
}
|
||
|
||
return sink.Close()
|
||
}()
|
||
|
||
if err != nil {
|
||
sink.Cancel()
|
||
return err
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// Release освобождает ресурсы
|
||
func (s *RaftSnapshot) Release() {}
|
||
|
||
// Apply применяет команду к FSM
|
||
func (f *RaftFSM) Apply(log *raft.Log) interface{} {
|
||
var batchCmd BatchCommand
|
||
if err := json.Unmarshal(log.Data, &batchCmd); err == nil && batchCmd.Type == "batch" {
|
||
for _, cmd := range batchCmd.Commands {
|
||
f.applySingleCommand(cmd)
|
||
}
|
||
if f.logger != nil {
|
||
f.logger.Debug(fmt.Sprintf("Applied batch command %s with %d sub-commands", batchCmd.BatchID, batchCmd.Size))
|
||
}
|
||
return nil
|
||
}
|
||
|
||
var jointCmd JointConsensusCommand
|
||
if err := json.Unmarshal(log.Data, &jointCmd); err == nil && jointCmd.Type == "joint_consensus" {
|
||
return f.applyJointConsensusCommand(jointCmd)
|
||
}
|
||
|
||
var moveCmd struct {
|
||
Type string `json:"type"`
|
||
ShardID string `json:"shard_id"`
|
||
TargetNode string `json:"target_node"`
|
||
}
|
||
if err := json.Unmarshal(log.Data, &moveCmd); err == nil && moveCmd.Type == "move_shard" {
|
||
return f.applyMoveShardCommand(moveCmd)
|
||
}
|
||
|
||
var moveDocCmd struct {
|
||
Type string `json:"type"`
|
||
ShardID string `json:"shard_id"`
|
||
Document map[string]interface{} `json:"document"`
|
||
DocumentID string `json:"document_id"`
|
||
TargetNode string `json:"target_node"`
|
||
}
|
||
if err := json.Unmarshal(log.Data, &moveDocCmd); err == nil && moveDocCmd.Type == "move_document" {
|
||
return f.applyMoveDocumentCommand(moveDocCmd)
|
||
}
|
||
|
||
var cmd NodeRegistrationCommand
|
||
if err := json.Unmarshal(log.Data, &cmd); err != nil {
|
||
var statusCmd NodeStatusCommand
|
||
if err2 := json.Unmarshal(log.Data, &statusCmd); err2 == nil {
|
||
return f.applyStatusCommand(statusCmd)
|
||
}
|
||
f.logger.Error(fmt.Sprintf("Failed to unmarshal raft command: %v", err))
|
||
return err
|
||
}
|
||
|
||
return f.applyRegistrationCommand(cmd)
|
||
}
|
||
|
||
func (f *RaftFSM) applySingleCommand(cmd interface{}) {
|
||
if regCmd, ok := cmd.(NodeRegistrationCommand); ok {
|
||
f.applyRegistrationCommand(regCmd)
|
||
} else if statusCmd, ok := cmd.(NodeStatusCommand); ok {
|
||
f.applyStatusCommand(statusCmd)
|
||
}
|
||
}
|
||
|
||
func (f *RaftFSM) applyRegistrationCommand(cmd NodeRegistrationCommand) interface{} {
|
||
f.state.mu.Lock()
|
||
defer f.state.mu.Unlock()
|
||
|
||
now := time.Now().UnixMilli()
|
||
f.state.UpdatedAt = now
|
||
|
||
switch cmd.Type {
|
||
case "register":
|
||
cmd.Node.JoinedAt = cmd.Timestamp
|
||
cmd.Node.UpdatedAt = now
|
||
cmd.Node.LastSeen = cmd.Timestamp
|
||
f.state.Nodes[cmd.Node.ID] = &cmd.Node
|
||
|
||
if f.coordinator != nil {
|
||
f.coordinator.shardManager.AddNode(cmd.Node.ID)
|
||
}
|
||
|
||
if f.logger != nil {
|
||
f.logger.Debug(fmt.Sprintf("Raft: Node registered: %s", cmd.Node.ID))
|
||
}
|
||
|
||
case "remove":
|
||
nodeID := cmd.NodeID
|
||
delete(f.state.Nodes, nodeID)
|
||
|
||
if f.coordinator != nil {
|
||
f.coordinator.shardManager.RemoveNode(nodeID)
|
||
}
|
||
|
||
if f.logger != nil {
|
||
f.logger.Debug(fmt.Sprintf("Raft: Node removed: %s", nodeID))
|
||
}
|
||
|
||
case "set_replication_factor":
|
||
oldFactor := f.state.ReplicationFactor
|
||
f.state.ReplicationFactor = cmd.Factor
|
||
if f.logger != nil {
|
||
f.logger.Debug(fmt.Sprintf("Raft: Replication factor changed from %d to %d", oldFactor, cmd.Factor))
|
||
}
|
||
|
||
case "create_shard":
|
||
if cmd.Shard != nil {
|
||
f.state.Shards[cmd.Shard.ID] = cmd.Shard
|
||
if f.logger != nil {
|
||
f.logger.Debug(fmt.Sprintf("Raft: Shard created: %s", cmd.Shard.Name))
|
||
}
|
||
}
|
||
|
||
case "move_shard":
|
||
if cmd.ShardID != "" && cmd.TargetNode != "" {
|
||
if shard, ok := f.state.Shards[cmd.ShardID]; ok {
|
||
oldLeader := shard.LeaderNode
|
||
shard.LeaderNode = cmd.TargetNode
|
||
shard.UpdatedAt = now
|
||
if f.logger != nil {
|
||
f.logger.Debug(fmt.Sprintf("Raft: Shard %s moved from %s to %s", cmd.ShardID, oldLeader, cmd.TargetNode))
|
||
}
|
||
}
|
||
}
|
||
|
||
case "move_document":
|
||
if f.logger != nil {
|
||
f.logger.Debug(fmt.Sprintf("Raft: Document %s moved to shard %s", cmd.Data["document_id"], cmd.ShardID))
|
||
}
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
func (f *RaftFSM) applyStatusCommand(cmd NodeStatusCommand) interface{} {
|
||
f.state.mu.Lock()
|
||
defer f.state.mu.Unlock()
|
||
|
||
if node, ok := f.state.Nodes[cmd.NodeID]; ok {
|
||
statusStr := "offline"
|
||
switch cmd.Status {
|
||
case 1:
|
||
statusStr = "active"
|
||
case 2:
|
||
statusStr = "syncing"
|
||
case 3:
|
||
statusStr = "failed"
|
||
default:
|
||
statusStr = "offline"
|
||
}
|
||
node.Status = statusStr
|
||
node.UpdatedAt = cmd.Timestamp
|
||
if f.logger != nil {
|
||
f.logger.Debug(fmt.Sprintf("Raft: Node %s status updated to %s", cmd.NodeID, statusStr))
|
||
}
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
func (f *RaftFSM) applyMoveShardCommand(cmd struct {
|
||
Type string `json:"type"`
|
||
ShardID string `json:"shard_id"`
|
||
TargetNode string `json:"target_node"`
|
||
}) interface{} {
|
||
f.state.mu.Lock()
|
||
defer f.state.mu.Unlock()
|
||
|
||
if shard, ok := f.state.Shards[cmd.ShardID]; ok {
|
||
oldLeader := shard.LeaderNode
|
||
shard.LeaderNode = cmd.TargetNode
|
||
shard.UpdatedAt = time.Now().UnixMilli()
|
||
|
||
if f.logger != nil {
|
||
f.logger.Debug(fmt.Sprintf("Raft: Shard %s moved from %s to %s", cmd.ShardID, oldLeader, cmd.TargetNode))
|
||
}
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
func (f *RaftFSM) applyMoveDocumentCommand(cmd struct {
|
||
Type string `json:"type"`
|
||
ShardID string `json:"shard_id"`
|
||
Document map[string]interface{} `json:"document"`
|
||
DocumentID string `json:"document_id"`
|
||
TargetNode string `json:"target_node"`
|
||
}) interface{} {
|
||
if f.logger != nil {
|
||
f.logger.Debug(fmt.Sprintf("Raft: Document %s moved to node %s, shard %s", cmd.DocumentID, cmd.TargetNode, cmd.ShardID))
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (f *RaftFSM) applyJointConsensusCommand(cmd JointConsensusCommand) interface{} {
|
||
f.state.mu.Lock()
|
||
defer f.state.mu.Unlock()
|
||
|
||
if f.logger != nil {
|
||
f.logger.Info(fmt.Sprintf("Raft: Joint consensus transition: %s at term %d", cmd.Transition, f.state.CurrentTerm))
|
||
}
|
||
|
||
switch cmd.Transition {
|
||
case "start":
|
||
if f.logger != nil {
|
||
f.logger.Info(fmt.Sprintf("Joint consensus started: old config has %d servers, new config has %d servers",
|
||
len(cmd.OldConfig), len(cmd.NewConfig)))
|
||
}
|
||
case "complete":
|
||
if f.logger != nil {
|
||
f.logger.Info("Joint consensus completed successfully")
|
||
}
|
||
case "abort":
|
||
if f.logger != nil {
|
||
f.logger.Warn("Joint consensus aborted")
|
||
}
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// Snapshot реализует создание снапшота
|
||
func (f *RaftFSM) Snapshot() (raft.FSMSnapshot, error) {
|
||
f.state.mu.RLock()
|
||
defer f.state.mu.RUnlock()
|
||
|
||
stateCopy := &RaftClusterState{
|
||
Nodes: make(map[string]*NodeInfo),
|
||
Shards: make(map[string]*ShardInfo),
|
||
ReplicationFactor: f.state.ReplicationFactor,
|
||
CurrentTerm: f.state.CurrentTerm,
|
||
VotedFor: f.state.VotedFor,
|
||
CreatedAt: f.state.CreatedAt,
|
||
UpdatedAt: time.Now().UnixMilli(),
|
||
}
|
||
|
||
for k, v := range f.state.Nodes {
|
||
stateCopy.Nodes[k] = v
|
||
}
|
||
for k, v := range f.state.Shards {
|
||
stateCopy.Shards[k] = v
|
||
}
|
||
|
||
return &RaftSnapshot{state: stateCopy}, nil
|
||
}
|
||
|
||
// Restore восстанавливает состояние из снапшота
|
||
func (f *RaftFSM) Restore(snapshot io.ReadCloser) error {
|
||
defer snapshot.Close()
|
||
|
||
var state RaftClusterState
|
||
decoder := json.NewDecoder(snapshot)
|
||
if err := decoder.Decode(&state); err != nil {
|
||
return err
|
||
}
|
||
|
||
f.state.mu.Lock()
|
||
defer f.state.mu.Unlock()
|
||
|
||
f.state.Nodes = state.Nodes
|
||
f.state.Shards = state.Shards
|
||
f.state.ReplicationFactor = state.ReplicationFactor
|
||
f.state.CurrentTerm = state.CurrentTerm
|
||
f.state.VotedFor = state.VotedFor
|
||
f.state.CreatedAt = state.CreatedAt
|
||
f.state.UpdatedAt = time.Now().UnixMilli()
|
||
|
||
if f.coordinator != nil {
|
||
f.coordinator.currentTerm.Store(state.CurrentTerm)
|
||
f.coordinator.replicationFactor.Store(state.ReplicationFactor)
|
||
f.coordinator.stableStore.Set([]byte("currentTerm"), []byte(fmt.Sprintf("%d", state.CurrentTerm)))
|
||
f.coordinator.stableStore.Set([]byte("votedFor"), []byte(state.VotedFor))
|
||
}
|
||
|
||
if f.logger != nil {
|
||
f.logger.Debug(fmt.Sprintf("Raft state restored from snapshot at %s with %d nodes",
|
||
time.UnixMilli(f.state.UpdatedAt).Format("2006-01-02 15:04:05.000"), len(state.Nodes)))
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// getLocalIP получает локальный IP адрес
|
||
func getLocalIP() string {
|
||
addrs, err := net.InterfaceAddrs()
|
||
if err != nil {
|
||
return "127.0.0.1"
|
||
}
|
||
for _, addr := range addrs {
|
||
if ipnet, ok := addr.(*net.IPNet); ok && !ipnet.IP.IsLoopback() && ipnet.IP.To4() != nil {
|
||
return ipnet.IP.String()
|
||
}
|
||
}
|
||
return "127.0.0.1"
|
||
}
|
||
|
||
// mapStatusToString преобразует статус в строку
|
||
func mapStatusToString(status int32) string {
|
||
switch status {
|
||
case 0:
|
||
return "offline"
|
||
case 1:
|
||
return "active"
|
||
case 2:
|
||
return "syncing"
|
||
case 3:
|
||
return "failed"
|
||
default:
|
||
return "unknown"
|
||
}
|
||
}
|
||
|
||
// ========== ClusterStatus ==========
|
||
|
||
// ClusterStatus представляет статус кластера
|
||
type ClusterStatus struct {
|
||
Name string `json:"name"`
|
||
TotalNodes int `json:"total_nodes"`
|
||
ActiveNodes int `json:"active_nodes"`
|
||
SyncingNodes int `json:"syncing_nodes"`
|
||
FailedNodes int `json:"failed_nodes"`
|
||
ReplicationFactor int `json:"replication_factor"`
|
||
LeaderID string `json:"leader_id"`
|
||
Health string `json:"health"`
|
||
CreatedAt int64 `json:"created_at"`
|
||
UpdatedAt int64 `json:"updated_at"`
|
||
PipelineEnabled bool `json:"pipeline_enabled"`
|
||
BatchCommitEnabled bool `json:"batch_commit_enabled"`
|
||
ReshardingEnabled bool `json:"resharding_enabled"`
|
||
JointConsensusActive bool `json:"joint_consensus_active"`
|
||
FallbackMode bool `json:"fallback_mode"`
|
||
}
|
||
|
||
// ========== RaftCoordinator ==========
|
||
|
||
// RaftCoordinator реализует координацию кластера через Raft
|
||
type RaftCoordinator struct {
|
||
raft *raft.Raft
|
||
fsm *RaftFSM
|
||
address string
|
||
raftAddr string
|
||
clusterName string
|
||
logger *log.Logger
|
||
config *config.Config
|
||
store *storage.Storage
|
||
stopChan chan struct{}
|
||
nodes sync.Map
|
||
replicationFactor atomic.Int32
|
||
replicationEnabled bool
|
||
masterMasterEnabled bool
|
||
syncReplication bool
|
||
isLeader atomic.Bool
|
||
leaderMonitor chan bool
|
||
singleNodeMode bool
|
||
localNodeInfo *NodeInfo
|
||
logStore *InmemStore
|
||
stableStore *InmemStore
|
||
createdAt int64
|
||
leaderSince atomic.Int64
|
||
lastElection atomic.Int64
|
||
electionCount atomic.Uint64
|
||
currentTerm atomic.Uint64
|
||
shardManager *ShardManager
|
||
splitBrainDetector *SplitBrainDetector
|
||
|
||
pipelineReplicator *PipelineReplicator
|
||
batchCommitManager *BatchCommitManager
|
||
reshardingManager *ReshardingManager
|
||
jointConsensusManager *JointConsensusManager
|
||
|
||
recoveryManager *RecoveryManager
|
||
authenticator *NodeAuthenticator
|
||
persistenceMgr *storage.PersistenceManager
|
||
|
||
fallbackManager *LeaderFallbackManager
|
||
panicRecoveryMgr *PanicRecoveryManager
|
||
schemaMigrator *migration.SchemaMigrator
|
||
}
|
||
|
||
// NewRaftCoordinator создаёт новый Raft координатор
|
||
func NewRaftCoordinator(cfg *config.Config, store *storage.Storage, logger *log.Logger) (*RaftCoordinator, error) {
|
||
now := time.Now().UnixMilli()
|
||
|
||
nodeIP := cfg.Cluster.NodeIP
|
||
if nodeIP == "" || nodeIP == "0.0.0.0" {
|
||
nodeIP = getLocalIP()
|
||
}
|
||
|
||
raftAddr := fmt.Sprintf("%s:%d", nodeIP, cfg.Cluster.RaftPort)
|
||
|
||
logger.Debug(fmt.Sprintf("Creating Raft coordinator at %s", raftAddr))
|
||
|
||
singleNodeMode := len(cfg.Cluster.Nodes) <= 1 || cfg.Cluster.Bootstrap
|
||
|
||
rc := &RaftCoordinator{
|
||
address: fmt.Sprintf("%s:%d", nodeIP, cfg.Cluster.NodePort),
|
||
raftAddr: raftAddr,
|
||
clusterName: cfg.Cluster.Name,
|
||
logger: logger,
|
||
config: cfg,
|
||
store: store,
|
||
stopChan: make(chan struct{}),
|
||
leaderMonitor: make(chan bool, 1),
|
||
replicationEnabled: cfg.Replication.Enabled,
|
||
masterMasterEnabled: cfg.Replication.MasterMaster,
|
||
syncReplication: cfg.Replication.SyncReplication,
|
||
singleNodeMode: singleNodeMode,
|
||
createdAt: now,
|
||
localNodeInfo: &NodeInfo{
|
||
ID: fmt.Sprintf("%s-%s", cfg.Cluster.Name, nodeIP),
|
||
IP: nodeIP,
|
||
Port: cfg.Cluster.NodePort,
|
||
Status: "active",
|
||
LastSeen: now,
|
||
JoinedAt: now,
|
||
UpdatedAt: now,
|
||
Version: 1,
|
||
},
|
||
}
|
||
|
||
rc.shardManager = NewShardManager(logger)
|
||
rc.splitBrainDetector = NewSplitBrainDetector(
|
||
logger,
|
||
cfg.Cluster.IsSplitBrainPreventionEnabled(),
|
||
cfg.Cluster.GetRecoveryTimeout(),
|
||
)
|
||
|
||
defaultReplicationFactor := int32(3)
|
||
if len(cfg.Cluster.Nodes) < 3 {
|
||
defaultReplicationFactor = int32(len(cfg.Cluster.Nodes))
|
||
}
|
||
rc.replicationFactor.Store(defaultReplicationFactor)
|
||
|
||
rc.fsm = &RaftFSM{
|
||
state: &RaftClusterState{
|
||
Nodes: make(map[string]*NodeInfo),
|
||
Shards: make(map[string]*ShardInfo),
|
||
ReplicationFactor: rc.replicationFactor.Load(),
|
||
CurrentTerm: 0,
|
||
VotedFor: "",
|
||
CreatedAt: now,
|
||
UpdatedAt: now,
|
||
},
|
||
logger: logger,
|
||
createdAt: now,
|
||
coordinator: rc,
|
||
}
|
||
|
||
if singleNodeMode {
|
||
rc.fsm.state.mu.Lock()
|
||
rc.fsm.state.Nodes[rc.localNodeInfo.ID] = rc.localNodeInfo
|
||
rc.fsm.state.mu.Unlock()
|
||
rc.isLeader.Store(true)
|
||
rc.leaderSince.Store(now)
|
||
rc.currentTerm.Store(1)
|
||
rc.shardManager.AddNode(rc.localNodeInfo.ID)
|
||
logger.Debug(fmt.Sprintf("Single-node mode: local node added to state: %s", rc.localNodeInfo.ID))
|
||
}
|
||
|
||
raftConfig := raft.DefaultConfig()
|
||
raftConfig.LocalID = raft.ServerID(rc.localNodeInfo.ID)
|
||
|
||
raftConfig.HeartbeatTimeout = cfg.Cluster.GetHeartbeatTimeout()
|
||
raftConfig.ElectionTimeout = cfg.Cluster.GetElectionTimeout()
|
||
raftConfig.CommitTimeout = cfg.Cluster.GetCommitTimeout()
|
||
raftConfig.LeaderLeaseTimeout = cfg.Cluster.GetCommitTimeout()
|
||
raftConfig.SnapshotInterval = cfg.Cluster.GetSnapshotInterval()
|
||
raftConfig.SnapshotThreshold = cfg.Cluster.GetSnapshotThreshold()
|
||
|
||
if singleNodeMode {
|
||
raftConfig.HeartbeatTimeout = 500 * time.Millisecond
|
||
raftConfig.ElectionTimeout = 500 * time.Millisecond
|
||
raftConfig.LeaderLeaseTimeout = 500 * time.Millisecond
|
||
raftConfig.LogOutput = io.Discard
|
||
logger.Debug("Running in single-node mode (warnings suppressed)")
|
||
} else {
|
||
raftConfig.LogOutput = os.Stderr
|
||
}
|
||
|
||
dataDir := cfg.Cluster.RaftDataDir
|
||
if err := os.MkdirAll(dataDir, 0755); err != nil {
|
||
return nil, fmt.Errorf("failed to create raft data dir: %v", err)
|
||
}
|
||
|
||
logger.Debug(fmt.Sprintf("Raft data directory: %s", dataDir))
|
||
|
||
rc.logStore = NewInmemStore(filepath.Join(dataDir, "raft-log.json"))
|
||
rc.stableStore = NewInmemStore(filepath.Join(dataDir, "raft-stable.json"))
|
||
|
||
if termData, err := rc.stableStore.Get([]byte("currentTerm")); err == nil && termData != nil {
|
||
var term uint64
|
||
fmt.Sscanf(string(termData), "%d", &term)
|
||
rc.currentTerm.Store(term)
|
||
rc.fsm.state.CurrentTerm = term
|
||
}
|
||
|
||
if votedForData, err := rc.stableStore.Get([]byte("votedFor")); err == nil && votedForData != nil {
|
||
rc.fsm.state.VotedFor = string(votedForData)
|
||
}
|
||
|
||
snapshotStore, err := raft.NewFileSnapshotStore(dataDir, 3, os.Stderr)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("failed to create snapshot store: %v", err)
|
||
}
|
||
|
||
transport, err := raft.NewTCPTransport(raftAddr, nil, 3, 10*time.Second, os.Stderr)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("failed to create transport: %v", err)
|
||
}
|
||
|
||
r, err := raft.NewRaft(raftConfig, rc.fsm, rc.logStore, rc.stableStore, snapshotStore, transport)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("failed to create raft: %v", err)
|
||
}
|
||
|
||
rc.raft = r
|
||
|
||
time.Sleep(500 * time.Millisecond)
|
||
|
||
bootstrapPath := filepath.Join(dataDir, "raft-log.json")
|
||
_, statErr := os.Stat(bootstrapPath)
|
||
needsBootstrap := os.IsNotExist(statErr)
|
||
|
||
if needsBootstrap && singleNodeMode {
|
||
logger.Debug("Bootstrapping single-node cluster...")
|
||
|
||
configuration := raft.Configuration{
|
||
Servers: []raft.Server{
|
||
{
|
||
ID: raftConfig.LocalID,
|
||
Address: transport.LocalAddr(),
|
||
},
|
||
},
|
||
}
|
||
|
||
future := r.BootstrapCluster(configuration)
|
||
if err := future.Error(); err != nil {
|
||
logger.Warn(fmt.Sprintf("Bootstrap error: %v", err))
|
||
} else {
|
||
logger.Debug("Single-node cluster bootstrapped successfully")
|
||
}
|
||
|
||
time.Sleep(1 * time.Second)
|
||
|
||
} else if needsBootstrap && len(cfg.Cluster.Nodes) > 1 {
|
||
logger.Debug("Bootstrapping multi-node cluster...")
|
||
|
||
servers := make([]raft.Server, 0, len(cfg.Cluster.Nodes))
|
||
for i, nodeAddr := range cfg.Cluster.Nodes {
|
||
serverID := raft.ServerID(fmt.Sprintf("%s-node%d", rc.clusterName, i+1))
|
||
servers = append(servers, raft.Server{
|
||
ID: serverID,
|
||
Address: raft.ServerAddress(nodeAddr),
|
||
})
|
||
}
|
||
|
||
configuration := raft.Configuration{
|
||
Servers: servers,
|
||
}
|
||
|
||
future := r.BootstrapCluster(configuration)
|
||
if err := future.Error(); err != nil {
|
||
logger.Warn(fmt.Sprintf("Bootstrap error: %v", err))
|
||
} else {
|
||
logger.Debug("Multi-node cluster bootstrapped successfully")
|
||
}
|
||
|
||
go rc.monitorLeadership()
|
||
|
||
logger.Debug("Waiting for leader election...")
|
||
timeout := time.After(cfg.Cluster.GetElectionTimeout() * 5)
|
||
leaderElected := false
|
||
|
||
for !leaderElected {
|
||
select {
|
||
case isLeader := <-rc.leaderMonitor:
|
||
if isLeader {
|
||
leaderElected = true
|
||
rc.isLeader.Store(true)
|
||
rc.leaderSince.Store(time.Now().UnixMilli())
|
||
rc.lastElection.Store(time.Now().UnixMilli())
|
||
logger.Debug("This node is now the cluster leader")
|
||
}
|
||
case <-timeout:
|
||
logger.Warn("Leader election timeout")
|
||
leaderElected = true
|
||
}
|
||
}
|
||
} else {
|
||
logger.Debug("Existing Raft state found, joining cluster...")
|
||
go rc.monitorLeadership()
|
||
|
||
if !singleNodeMode {
|
||
time.Sleep(1 * time.Second)
|
||
if r.State() == raft.Leader {
|
||
rc.isLeader.Store(true)
|
||
rc.leaderSince.Store(time.Now().UnixMilli())
|
||
logger.Debug("This node is the cluster leader")
|
||
}
|
||
}
|
||
}
|
||
|
||
rc.pipelineReplicator = NewPipelineReplicator(rc, 50, 10*time.Millisecond, logger)
|
||
rc.batchCommitManager = NewBatchCommitManager(100, 50*time.Millisecond, true, logger)
|
||
rc.reshardingManager = NewReshardingManager(rc, logger)
|
||
rc.jointConsensusManager = NewJointConsensusManager(rc, logger)
|
||
|
||
authConfig := &NodeAuthConfig{
|
||
Enabled: true,
|
||
TokenTTL: 24 * time.Hour,
|
||
PrivateKeyPath: filepath.Join(dataDir, "node_private.key"),
|
||
PublicKeyPath: filepath.Join(dataDir, "node_public.key"),
|
||
AllowedNodes: cfg.Cluster.Nodes,
|
||
RequireMTLS: false,
|
||
}
|
||
|
||
authenticator, err := NewNodeAuthenticator(authConfig, logger)
|
||
if err != nil {
|
||
logger.Warn(fmt.Sprintf("Failed to initialize node authenticator: %v", err))
|
||
}
|
||
rc.authenticator = authenticator
|
||
|
||
rc.recoveryManager = NewRecoveryManager(rc, logger)
|
||
rc.recoveryManager.Start()
|
||
|
||
persistenceConfig := storage.DefaultPersistenceConfig()
|
||
persistenceConfig.DataDir = filepath.Join(dataDir, "persistence")
|
||
rc.persistenceMgr = storage.NewPersistenceManager(persistenceConfig, store, logger)
|
||
rc.persistenceMgr.Start()
|
||
|
||
if err := rc.persistenceMgr.LoadAll(); err != nil {
|
||
logger.Warn(fmt.Sprintf("Failed to load persisted data: %v", err))
|
||
}
|
||
|
||
panicRecoveryMgr := NewPanicRecoveryManager(logger)
|
||
rc.panicRecoveryMgr = panicRecoveryMgr
|
||
|
||
panicRecoveryMgr.RegisterRecoveryFunc("Raft", func(p interface{}) error {
|
||
logger.Error(fmt.Sprintf("Raft panic recovered: %v, attempting to restart coordinator", p))
|
||
return nil
|
||
})
|
||
|
||
panicRecoveryMgr.RegisterRecoveryFunc("PipelineReplicator", func(p interface{}) error {
|
||
logger.Error(fmt.Sprintf("Pipeline replicator panic: %v, reinitializing", p))
|
||
if rc.pipelineReplicator != nil {
|
||
rc.pipelineReplicator.Stop()
|
||
rc.pipelineReplicator = NewPipelineReplicator(rc, 50, 10*time.Millisecond, logger)
|
||
}
|
||
return nil
|
||
})
|
||
|
||
panicRecoveryMgr.RegisterRecoveryFunc("BatchCommitManager", func(p interface{}) error {
|
||
logger.Error(fmt.Sprintf("Batch commit manager panic: %v, reinitializing", p))
|
||
if rc.batchCommitManager != nil {
|
||
rc.batchCommitManager.Stop()
|
||
rc.batchCommitManager = NewBatchCommitManager(100, 50*time.Millisecond, true, logger)
|
||
}
|
||
return nil
|
||
})
|
||
|
||
panicRecoveryMgr.RegisterRecoveryFunc("ReshardingManager", func(p interface{}) error {
|
||
logger.Error(fmt.Sprintf("Resharding manager panic: %v, reinitializing", p))
|
||
if rc.reshardingManager != nil {
|
||
rc.reshardingManager.Stop()
|
||
rc.reshardingManager = NewReshardingManager(rc, logger)
|
||
}
|
||
return nil
|
||
})
|
||
|
||
panicRecoveryMgr.RegisterRecoveryFunc("RecoveryManager", func(p interface{}) error {
|
||
logger.Error(fmt.Sprintf("Recovery manager panic: %v, reinitializing", p))
|
||
if rc.recoveryManager != nil {
|
||
rc.recoveryManager.Stop()
|
||
rc.recoveryManager = NewRecoveryManager(rc, logger)
|
||
rc.recoveryManager.Start()
|
||
}
|
||
return nil
|
||
})
|
||
|
||
fallbackConfig := DefaultFallbackConfig()
|
||
fallbackConfig.ElectionTimeout = cfg.Cluster.GetElectionTimeout()
|
||
fallbackConfig.FallbackTimeout = cfg.Cluster.GetRecoveryTimeout()
|
||
rc.fallbackManager = NewLeaderFallbackManager(rc, logger, fallbackConfig)
|
||
|
||
migrationDir := filepath.Join(dataDir, "migrations")
|
||
rc.schemaMigrator = migration.NewSchemaMigrator(store, logger, migrationDir)
|
||
|
||
status := rc.schemaMigrator.GetStatus()
|
||
if status.PendingMigrations > 0 {
|
||
logger.Info(fmt.Sprintf("Found %d pending schema migrations, current version: %s",
|
||
status.PendingMigrations, status.CurrentVersion))
|
||
|
||
if err := rc.schemaMigrator.Migrate("2.1.0"); err != nil {
|
||
logger.Warn(fmt.Sprintf("Schema migration warning: %v", err))
|
||
} else {
|
||
logger.Info("Schema migrations completed successfully")
|
||
}
|
||
}
|
||
|
||
if !singleNodeMode {
|
||
go rc.rebalanceMonitor()
|
||
}
|
||
|
||
logger.InfoWithFields("Raft coordinator started", map[string]interface{}{
|
||
"is_leader": rc.isLeader.Load(),
|
||
"single_node": singleNodeMode,
|
||
"term": rc.currentTerm.Load(),
|
||
"created_at": time.UnixMilli(rc.createdAt).Format("2006-01-02 15:04:05.000"),
|
||
"fallback_enabled": rc.fallbackManager != nil,
|
||
"panic_recovery": rc.panicRecoveryMgr != nil,
|
||
"schema_migration": rc.schemaMigrator != nil,
|
||
})
|
||
|
||
return rc, nil
|
||
}
|
||
|
||
// RealFsync выполняет реальную синхронизацию с диском
|
||
func (rc *RaftCoordinator) RealFsync(file *os.File) error {
|
||
if file == nil {
|
||
return nil
|
||
}
|
||
return file.Sync()
|
||
}
|
||
|
||
// RealFsyncWithRetry выполняет fsync с повторными попытками
|
||
func (rc *RaftCoordinator) RealFsyncWithRetry(file *os.File, maxRetries int) error {
|
||
var lastErr error
|
||
for i := 0; i < maxRetries; i++ {
|
||
if err := rc.RealFsync(file); err != nil {
|
||
lastErr = err
|
||
if rc.logger != nil {
|
||
rc.logger.Warn(fmt.Sprintf("Fsync attempt %d failed: %v, retrying...", i+1, err))
|
||
}
|
||
time.Sleep(time.Duration(100*(i+1)) * time.Millisecond)
|
||
continue
|
||
}
|
||
return nil
|
||
}
|
||
return lastErr
|
||
}
|
||
|
||
// rebalanceMonitor периодически проверяет необходимость ребалансировки
|
||
func (rc *RaftCoordinator) rebalanceMonitor() {
|
||
ticker := time.NewTicker(5 * time.Minute)
|
||
defer ticker.Stop()
|
||
|
||
for {
|
||
select {
|
||
case <-rc.stopChan:
|
||
return
|
||
case <-ticker.C:
|
||
if rc.IsLeader() && rc.reshardingManager != nil {
|
||
rc.reshardingManager.TriggerResharding("periodic_rebalance")
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// monitorLeadership отслеживает изменения лидера
|
||
func (rc *RaftCoordinator) monitorLeadership() {
|
||
ticker := time.NewTicker(rc.config.Cluster.GetHeartbeatTimeout() / 2)
|
||
defer ticker.Stop()
|
||
|
||
wasLeader := false
|
||
|
||
for {
|
||
select {
|
||
case <-rc.stopChan:
|
||
return
|
||
case <-ticker.C:
|
||
if rc.raft == nil {
|
||
continue
|
||
}
|
||
isLeader := rc.raft.State() == raft.Leader
|
||
if isLeader != wasLeader {
|
||
wasLeader = isLeader
|
||
select {
|
||
case rc.leaderMonitor <- isLeader:
|
||
default:
|
||
}
|
||
if isLeader {
|
||
rc.isLeader.Store(true)
|
||
newTerm := rc.currentTerm.Add(1)
|
||
rc.leaderSince.Store(time.Now().UnixMilli())
|
||
rc.electionCount.Add(1)
|
||
rc.fsm.state.CurrentTerm = newTerm
|
||
rc.stableStore.Set([]byte("currentTerm"), []byte(fmt.Sprintf("%d", newTerm)))
|
||
rc.logger.Debug(fmt.Sprintf("Leadership acquired at term %d (election #%d)",
|
||
newTerm, rc.electionCount.Load()))
|
||
|
||
nodes := rc.GetAllNodes()
|
||
for _, node := range nodes {
|
||
rc.shardManager.AddNode(node.ID)
|
||
}
|
||
} else {
|
||
rc.isLeader.Store(false)
|
||
rc.lastElection.Store(time.Now().UnixMilli())
|
||
rc.logger.Debug("Leadership lost")
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// RegisterNode регистрирует узел через Raft
|
||
func (rc *RaftCoordinator) RegisterNode(node *Node) error {
|
||
startTime := time.Now()
|
||
now := time.Now().UnixMilli()
|
||
|
||
if rc.splitBrainDetector.IsQuarantined(node.ID) {
|
||
return fmt.Errorf("node %s is quarantined due to previous split-brain", node.ID)
|
||
}
|
||
|
||
nodeInfo := &NodeInfo{
|
||
ID: node.ID,
|
||
IP: node.IP,
|
||
Port: node.Port,
|
||
Status: "active",
|
||
LastSeen: now,
|
||
JoinedAt: now,
|
||
UpdatedAt: now,
|
||
Version: 1,
|
||
}
|
||
|
||
if rc.singleNodeMode {
|
||
rc.logger.Debug("Single-node mode: registering node without Raft consensus")
|
||
|
||
rc.nodes.Store(node.ID, nodeInfo)
|
||
|
||
rc.fsm.state.mu.Lock()
|
||
rc.fsm.state.Nodes[node.ID] = nodeInfo
|
||
rc.fsm.state.UpdatedAt = now
|
||
rc.fsm.state.mu.Unlock()
|
||
|
||
rc.shardManager.AddNode(node.ID)
|
||
|
||
rc.logger.Debug(fmt.Sprintf("Node registered locally in single-node mode: %s", node.ID))
|
||
return nil
|
||
}
|
||
|
||
if !rc.IsLeader() {
|
||
leader := rc.GetLeader()
|
||
if leader != nil {
|
||
rc.logger.Warn(fmt.Sprintf("Current node is not leader. Leader is %s:%d",
|
||
leader.IP, leader.Port))
|
||
return fmt.Errorf("node is not the leader. Please connect to leader at %s:%d", leader.IP, leader.Port)
|
||
}
|
||
return fmt.Errorf("node is not the leader and no leader found")
|
||
}
|
||
|
||
cmd := NodeRegistrationCommand{
|
||
Type: "register",
|
||
Node: *nodeInfo,
|
||
Timestamp: now,
|
||
}
|
||
|
||
data, err := json.Marshal(cmd)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
future := rc.raft.Apply(data, rc.config.Replication.GetReplicationTimeout())
|
||
if err := future.Error(); err != nil {
|
||
rc.logger.ErrorWithFields("Failed to register node", map[string]interface{}{
|
||
"error": err.Error(),
|
||
"node": node.ID,
|
||
})
|
||
return fmt.Errorf("failed to register node via raft: %v", err)
|
||
}
|
||
|
||
rc.nodes.Store(node.ID, nodeInfo)
|
||
rc.shardManager.AddNode(node.ID)
|
||
|
||
duration := time.Since(startTime).Milliseconds()
|
||
rc.logger.InfoWithFields("Node registered", map[string]interface{}{
|
||
"node_id": node.ID,
|
||
"address": node.GetAddress(),
|
||
"duration_ms": duration,
|
||
})
|
||
|
||
return nil
|
||
}
|
||
|
||
// UpdateNodeStatus обновляет статус узла через Raft
|
||
func (rc *RaftCoordinator) UpdateNodeStatus(nodeID string, status NodeStatus) error {
|
||
now := time.Now().UnixMilli()
|
||
|
||
if rc.splitBrainDetector.IsQuarantined(nodeID) {
|
||
return fmt.Errorf("node %s is quarantined, cannot update status", nodeID)
|
||
}
|
||
|
||
if rc.singleNodeMode {
|
||
rc.fsm.state.mu.Lock()
|
||
if node, ok := rc.fsm.state.Nodes[nodeID]; ok {
|
||
node.Status = mapStatusToString(int32(status))
|
||
node.UpdatedAt = now
|
||
}
|
||
rc.fsm.state.mu.Unlock()
|
||
return nil
|
||
}
|
||
|
||
if !rc.IsLeader() {
|
||
return fmt.Errorf("node is not the leader")
|
||
}
|
||
|
||
cmd := NodeStatusCommand{
|
||
Type: "update_status",
|
||
NodeID: nodeID,
|
||
Status: int32(status),
|
||
Timestamp: now,
|
||
}
|
||
|
||
data, err := json.Marshal(cmd)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
future := rc.raft.Apply(data, rc.config.Replication.GetReplicationTimeout())
|
||
return future.Error()
|
||
}
|
||
|
||
// RemoveNode удаляет узел через Raft
|
||
func (rc *RaftCoordinator) RemoveNode(nodeID string) error {
|
||
now := time.Now().UnixMilli()
|
||
|
||
if rc.singleNodeMode {
|
||
rc.nodes.Delete(nodeID)
|
||
rc.fsm.state.mu.Lock()
|
||
delete(rc.fsm.state.Nodes, nodeID)
|
||
rc.fsm.state.UpdatedAt = now
|
||
rc.fsm.state.mu.Unlock()
|
||
rc.shardManager.RemoveNode(nodeID)
|
||
rc.logger.Debug(fmt.Sprintf("Node removed locally in single-node mode: %s", nodeID))
|
||
return nil
|
||
}
|
||
|
||
if !rc.IsLeader() {
|
||
return fmt.Errorf("node is not the leader")
|
||
}
|
||
|
||
cmd := NodeRegistrationCommand{
|
||
Type: "remove",
|
||
NodeID: nodeID,
|
||
Timestamp: now,
|
||
}
|
||
|
||
data, err := json.Marshal(cmd)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
future := rc.raft.Apply(data, rc.config.Replication.GetReplicationTimeout())
|
||
if err := future.Error(); err != nil {
|
||
return fmt.Errorf("failed to remove node via raft: %v", err)
|
||
}
|
||
|
||
rc.nodes.Delete(nodeID)
|
||
rc.shardManager.RemoveNode(nodeID)
|
||
rc.logger.Debug(fmt.Sprintf("Node removed via Raft: %s", nodeID))
|
||
return nil
|
||
}
|
||
|
||
// GetActiveNodes возвращает активные узлы
|
||
func (rc *RaftCoordinator) GetActiveNodes() []*NodeInfo {
|
||
nodes := make([]*NodeInfo, 0)
|
||
now := time.Now().UnixMilli()
|
||
|
||
state := rc.fsm.state
|
||
state.mu.RLock()
|
||
defer state.mu.RUnlock()
|
||
|
||
for _, nodeInfo := range state.Nodes {
|
||
if now-nodeInfo.LastSeen < 30000 && nodeInfo.Status == "active" {
|
||
if !rc.splitBrainDetector.IsQuarantined(nodeInfo.ID) {
|
||
nodes = append(nodes, nodeInfo)
|
||
}
|
||
}
|
||
}
|
||
|
||
if rc.singleNodeMode && len(nodes) == 0 && rc.localNodeInfo != nil {
|
||
nodes = append(nodes, rc.localNodeInfo)
|
||
}
|
||
|
||
return nodes
|
||
}
|
||
|
||
// GetAllNodes возвращает все узлы
|
||
func (rc *RaftCoordinator) GetAllNodes() []*NodeInfo {
|
||
state := rc.fsm.state
|
||
state.mu.RLock()
|
||
defer state.mu.RUnlock()
|
||
|
||
nodes := make([]*NodeInfo, 0, len(state.Nodes))
|
||
for _, node := range state.Nodes {
|
||
nodes = append(nodes, node)
|
||
}
|
||
|
||
if rc.singleNodeMode && len(nodes) == 0 && rc.localNodeInfo != nil {
|
||
nodes = append(nodes, rc.localNodeInfo)
|
||
}
|
||
|
||
return nodes
|
||
}
|
||
|
||
// GetNodeByID возвращает узел по ID
|
||
func (rc *RaftCoordinator) GetNodeByID(nodeID string) *NodeInfo {
|
||
state := rc.fsm.state
|
||
state.mu.RLock()
|
||
defer state.mu.RUnlock()
|
||
|
||
if node, ok := state.Nodes[nodeID]; ok {
|
||
return node
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// GetShardByID возвращает шард по ID
|
||
func (rc *RaftCoordinator) GetShardByID(shardID string) *ShardInfo {
|
||
state := rc.fsm.state
|
||
state.mu.RLock()
|
||
defer state.mu.RUnlock()
|
||
|
||
if shard, ok := state.Shards[shardID]; ok {
|
||
return shard
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// GetLeader возвращает лидера
|
||
func (rc *RaftCoordinator) GetLeader() *NodeInfo {
|
||
if rc.singleNodeMode {
|
||
return rc.localNodeInfo
|
||
}
|
||
|
||
leaderAddr := rc.raft.Leader()
|
||
if leaderAddr == "" {
|
||
return nil
|
||
}
|
||
|
||
state := rc.fsm.state
|
||
state.mu.RLock()
|
||
defer state.mu.RUnlock()
|
||
|
||
for _, node := range state.Nodes {
|
||
nodeAddr := fmt.Sprintf("%s:%d", node.IP, node.Port)
|
||
if nodeAddr == string(leaderAddr) {
|
||
return node
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// IsLeader проверяет, является ли текущий узел лидером
|
||
func (rc *RaftCoordinator) IsLeader() bool {
|
||
if rc.singleNodeMode {
|
||
return true
|
||
}
|
||
return rc.isLeader.Load()
|
||
}
|
||
|
||
// GetCurrentTerm возвращает текущий терм Raft
|
||
func (rc *RaftCoordinator) GetCurrentTerm() uint64 {
|
||
return rc.currentTerm.Load()
|
||
}
|
||
|
||
// GetLeaderSince возвращает время начала лидерства
|
||
func (rc *RaftCoordinator) GetLeaderSince() int64 {
|
||
return rc.leaderSince.Load()
|
||
}
|
||
|
||
// GetElectionCount возвращает количество выборов
|
||
func (rc *RaftCoordinator) GetElectionCount() uint64 {
|
||
return rc.electionCount.Load()
|
||
}
|
||
|
||
// SendHeartbeat обновляет heartbeat узла
|
||
func (rc *RaftCoordinator) SendHeartbeat(nodeID string) {
|
||
now := time.Now().UnixMilli()
|
||
|
||
if val, ok := rc.nodes.Load(nodeID); ok {
|
||
nodeInfo := val.(*NodeInfo)
|
||
nodeInfo.LastSeen = now
|
||
nodeInfo.UpdatedAt = now
|
||
rc.nodes.Store(nodeID, nodeInfo)
|
||
}
|
||
|
||
rc.fsm.state.mu.Lock()
|
||
if nodeInfo, ok := rc.fsm.state.Nodes[nodeID]; ok {
|
||
nodeInfo.LastSeen = now
|
||
nodeInfo.UpdatedAt = now
|
||
}
|
||
rc.fsm.state.mu.Unlock()
|
||
}
|
||
|
||
// HandleStatusSync обрабатывает синхронизацию статуса
|
||
func (rc *RaftCoordinator) HandleStatusSync(leaderID string, term uint64, clusterSize int) {
|
||
if rc.splitBrainDetector.Detect(term, leaderID, clusterSize) {
|
||
candidates := make(map[string]uint64)
|
||
candidates[leaderID] = rc.getCommitIndex()
|
||
candidates[rc.localNodeInfo.ID] = rc.getCommitIndex()
|
||
|
||
winner := rc.splitBrainDetector.Resolve(term, candidates)
|
||
if winner == rc.localNodeInfo.ID && !rc.IsLeader() {
|
||
rc.raft.LeadershipTransfer()
|
||
rc.logger.Warn("Split-brain resolved: initiating leadership transfer")
|
||
} else if winner != leaderID && winner != "" {
|
||
rc.splitBrainDetector.QuarantineNode(leaderID)
|
||
rc.logger.Warn(fmt.Sprintf("Quarantining node %s due to split-brain", leaderID))
|
||
}
|
||
}
|
||
}
|
||
|
||
func (rc *RaftCoordinator) getCommitIndex() uint64 {
|
||
if rc.raft == nil {
|
||
return 0
|
||
}
|
||
return rc.raft.AppliedIndex()
|
||
}
|
||
|
||
// GetClusterStatus возвращает статус кластера
|
||
func (rc *RaftCoordinator) GetClusterStatus() *ClusterStatus {
|
||
nodes := rc.GetAllNodes()
|
||
activeNodes := rc.GetActiveNodes()
|
||
|
||
syncingNodes := 0
|
||
for _, node := range nodes {
|
||
if node.Status == "syncing" {
|
||
syncingNodes++
|
||
}
|
||
}
|
||
|
||
leader := rc.GetLeader()
|
||
leaderID := ""
|
||
if leader != nil {
|
||
leaderID = leader.ID
|
||
}
|
||
|
||
now := time.Now().UnixMilli()
|
||
|
||
health := rc.calculateHealth()
|
||
|
||
if rc.splitBrainDetector.Detect(rc.currentTerm.Load(), leaderID, len(nodes)) {
|
||
health = "split_brain"
|
||
}
|
||
|
||
return &ClusterStatus{
|
||
Name: rc.clusterName,
|
||
TotalNodes: len(nodes),
|
||
ActiveNodes: len(activeNodes),
|
||
SyncingNodes: syncingNodes,
|
||
FailedNodes: len(nodes) - len(activeNodes),
|
||
ReplicationFactor: int(rc.replicationFactor.Load()),
|
||
LeaderID: leaderID,
|
||
Health: health,
|
||
CreatedAt: rc.createdAt,
|
||
UpdatedAt: now,
|
||
PipelineEnabled: rc.pipelineReplicator != nil,
|
||
BatchCommitEnabled: rc.batchCommitManager != nil,
|
||
ReshardingEnabled: rc.reshardingManager != nil,
|
||
JointConsensusActive: rc.jointConsensusManager != nil && rc.jointConsensusManager.IsJointConsensusActive(),
|
||
FallbackMode: rc.fallbackManager != nil && rc.fallbackManager.IsFallbackMode(),
|
||
}
|
||
}
|
||
|
||
// calculateHealth вычисляет здоровье кластера
|
||
func (rc *RaftCoordinator) calculateHealth() string {
|
||
activeNodes := rc.GetActiveNodes()
|
||
totalNodes := rc.GetAllNodes()
|
||
|
||
if len(totalNodes) == 0 {
|
||
return "critical"
|
||
}
|
||
|
||
ratio := float64(len(activeNodes)) / float64(len(totalNodes))
|
||
if ratio >= 0.8 {
|
||
return "healthy"
|
||
} else if ratio >= 0.5 {
|
||
return "degraded"
|
||
}
|
||
return "critical"
|
||
}
|
||
|
||
// GetReplicationFactor возвращает фактор репликации
|
||
func (rc *RaftCoordinator) GetReplicationFactor() int {
|
||
return int(rc.replicationFactor.Load())
|
||
}
|
||
|
||
// SetReplicationFactor устанавливает фактор репликации через Raft
|
||
func (rc *RaftCoordinator) SetReplicationFactor(factor int) error {
|
||
now := time.Now().UnixMilli()
|
||
|
||
if !rc.IsLeader() {
|
||
return fmt.Errorf("node is not the leader")
|
||
}
|
||
|
||
cmd := NodeRegistrationCommand{
|
||
Type: "set_replication_factor",
|
||
Factor: int32(factor),
|
||
Timestamp: now,
|
||
}
|
||
|
||
data, err := json.Marshal(cmd)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
future := rc.raft.Apply(data, rc.config.Replication.GetReplicationTimeout())
|
||
if err := future.Error(); err != nil {
|
||
return fmt.Errorf("failed to set replication factor via raft: %v", err)
|
||
}
|
||
|
||
rc.replicationFactor.Store(int32(factor))
|
||
rc.logger.Debug(fmt.Sprintf("Replication factor set to %d via Raft", factor))
|
||
return nil
|
||
}
|
||
|
||
// CreateShard создаёт новый шард
|
||
func (rc *RaftCoordinator) CreateShard(name string, nodes []string) (*ShardInfo, error) {
|
||
if !rc.IsLeader() {
|
||
return nil, fmt.Errorf("only leader can create shards")
|
||
}
|
||
|
||
now := time.Now().UnixMilli()
|
||
shard := &ShardInfo{
|
||
ID: fmt.Sprintf("shard_%s_%d", name, now),
|
||
Name: name,
|
||
Nodes: nodes,
|
||
LeaderNode: nodes[0],
|
||
Status: "active",
|
||
CreatedAt: now,
|
||
UpdatedAt: now,
|
||
LastRebalanced: now,
|
||
DocumentCount: 0,
|
||
SizeBytes: 0,
|
||
}
|
||
|
||
cmd := NodeRegistrationCommand{
|
||
Type: "create_shard",
|
||
Shard: shard,
|
||
Timestamp: now,
|
||
}
|
||
|
||
data, err := json.Marshal(cmd)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
future := rc.raft.Apply(data, rc.config.Replication.GetReplicationTimeout())
|
||
if err := future.Error(); err != nil {
|
||
return nil, fmt.Errorf("failed to create shard via raft: %v", err)
|
||
}
|
||
|
||
rc.logger.Debug(fmt.Sprintf("Shard created: %s with nodes %v", name, nodes))
|
||
return shard, nil
|
||
}
|
||
|
||
// GetShardForCollection возвращает шард для коллекции
|
||
func (rc *RaftCoordinator) GetShardForCollection(database, collection string) *ShardInfo {
|
||
return rc.shardManager.GetShard(database, collection)
|
||
}
|
||
|
||
// GetAllShards возвращает все шарды
|
||
func (rc *RaftCoordinator) GetAllShards() []*ShardInfo {
|
||
state := rc.fsm.state
|
||
state.mu.RLock()
|
||
defer state.mu.RUnlock()
|
||
|
||
shards := make([]*ShardInfo, 0, len(state.Shards))
|
||
for _, shard := range state.Shards {
|
||
shards = append(shards, shard)
|
||
}
|
||
return shards
|
||
}
|
||
|
||
// GetPipelineStats возвращает статистику пайплайна
|
||
func (rc *RaftCoordinator) GetPipelineStats() map[string]interface{} {
|
||
if rc.pipelineReplicator == nil {
|
||
return map[string]interface{}{"enabled": false}
|
||
}
|
||
return rc.pipelineReplicator.GetStats()
|
||
}
|
||
|
||
// GetBatchCommitStats возвращает статистику пакетных коммитов
|
||
func (rc *RaftCoordinator) GetBatchCommitStats() map[string]interface{} {
|
||
if rc.batchCommitManager == nil {
|
||
return map[string]interface{}{"enabled": false}
|
||
}
|
||
return rc.batchCommitManager.GetStats()
|
||
}
|
||
|
||
// GetReshardingStats возвращает статистику решардинга
|
||
func (rc *RaftCoordinator) GetReshardingStats() map[string]interface{} {
|
||
if rc.reshardingManager == nil {
|
||
return map[string]interface{}{"enabled": false}
|
||
}
|
||
return rc.reshardingManager.GetMetrics()
|
||
}
|
||
|
||
// GetJointConsensusStatus возвращает статус совместного консенсуса
|
||
func (rc *RaftCoordinator) GetJointConsensusStatus() map[string]interface{} {
|
||
if rc.jointConsensusManager == nil {
|
||
return map[string]interface{}{"active": false}
|
||
}
|
||
return rc.jointConsensusManager.GetJointConsensusStatus()
|
||
}
|
||
|
||
// TriggerResharding запускает перераспределение шардов
|
||
func (rc *RaftCoordinator) TriggerResharding(reason string) error {
|
||
if rc.reshardingManager == nil {
|
||
return fmt.Errorf("resharding manager not initialized")
|
||
}
|
||
return rc.reshardingManager.TriggerResharding(reason)
|
||
}
|
||
|
||
// StartJointConsensus начинает совместный консенсус
|
||
func (rc *RaftCoordinator) StartJointConsensus(newServers []raft.Server) error {
|
||
if rc.jointConsensusManager == nil {
|
||
return fmt.Errorf("joint consensus manager not initialized")
|
||
}
|
||
return rc.jointConsensusManager.StartJointConsensus(newServers)
|
||
}
|
||
|
||
// CompleteJointConsensus завершает совместный консенсус
|
||
func (rc *RaftCoordinator) CompleteJointConsensus() error {
|
||
if rc.jointConsensusManager == nil {
|
||
return fmt.Errorf("joint consensus manager not initialized")
|
||
}
|
||
return rc.jointConsensusManager.CompleteJointConsensus()
|
||
}
|
||
|
||
// AbortJointConsensus прерывает совместный консенсус
|
||
func (rc *RaftCoordinator) AbortJointConsensus() error {
|
||
if rc.jointConsensusManager == nil {
|
||
return fmt.Errorf("joint consensus manager not initialized")
|
||
}
|
||
return rc.jointConsensusManager.AbortJointConsensus()
|
||
}
|
||
|
||
// GetFallbackManager возвращает менеджер fallback
|
||
func (rc *RaftCoordinator) GetFallbackManager() *LeaderFallbackManager {
|
||
return rc.fallbackManager
|
||
}
|
||
|
||
// GetPanicRecoveryManager возвращает менеджер восстановления после паник
|
||
func (rc *RaftCoordinator) GetPanicRecoveryManager() *PanicRecoveryManager {
|
||
return rc.panicRecoveryMgr
|
||
}
|
||
|
||
// GetSchemaMigrator возвращает мигратор схемы данных
|
||
func (rc *RaftCoordinator) GetSchemaMigrator() *migration.SchemaMigrator {
|
||
return rc.schemaMigrator
|
||
}
|
||
|
||
// GetPersistenceManager возвращает менеджер персистентности
|
||
func (rc *RaftCoordinator) GetPersistenceManager() *storage.PersistenceManager {
|
||
return rc.persistenceMgr
|
||
}
|
||
|
||
// SubmitWriteWithFallback отправляет запись с поддержкой fallback
|
||
func (rc *RaftCoordinator) SubmitWriteWithFallback(data []byte) error {
|
||
if rc.fallbackManager != nil {
|
||
return rc.fallbackManager.SubmitWrite(data)
|
||
}
|
||
|
||
if rc.IsLeader() {
|
||
future := rc.raft.Apply(data, 10*time.Second)
|
||
return future.Error()
|
||
}
|
||
return fmt.Errorf("not leader and fallback manager not available")
|
||
}
|
||
|
||
// GetFallbackStats возвращает статистику fallback менеджера
|
||
func (rc *RaftCoordinator) GetFallbackStats() map[string]interface{} {
|
||
if rc.fallbackManager != nil {
|
||
return rc.fallbackManager.GetStats()
|
||
}
|
||
return map[string]interface{}{"enabled": false}
|
||
}
|
||
|
||
// GetPanicRecoveryStats возвращает статистику восстановления после паник
|
||
func (rc *RaftCoordinator) GetPanicRecoveryStats() map[string]interface{} {
|
||
if rc.panicRecoveryMgr != nil {
|
||
return rc.panicRecoveryMgr.GetStats()
|
||
}
|
||
return map[string]interface{}{"enabled": false}
|
||
}
|
||
|
||
// GetMigrationStatus возвращает статус миграций схемы данных
|
||
func (rc *RaftCoordinator) GetMigrationStatus() *migration.MigrationStatus {
|
||
if rc.schemaMigrator != nil {
|
||
return rc.schemaMigrator.GetStatus()
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// Stop останавливает координатор
|
||
func (rc *RaftCoordinator) Stop() {
|
||
now := time.Now().UnixMilli()
|
||
|
||
rc.logger.Info("Stopping Raft coordinator...")
|
||
|
||
if rc.pipelineReplicator != nil {
|
||
rc.pipelineReplicator.Stop()
|
||
rc.logger.Debug("Pipeline replicator stopped")
|
||
}
|
||
if rc.batchCommitManager != nil {
|
||
rc.batchCommitManager.Stop()
|
||
rc.logger.Debug("Batch commit manager stopped")
|
||
}
|
||
if rc.reshardingManager != nil {
|
||
rc.reshardingManager.Stop()
|
||
rc.logger.Debug("Resharding manager stopped")
|
||
}
|
||
if rc.recoveryManager != nil {
|
||
rc.recoveryManager.Stop()
|
||
rc.logger.Debug("Recovery manager stopped")
|
||
}
|
||
if rc.persistenceMgr != nil {
|
||
rc.persistenceMgr.Stop()
|
||
rc.logger.Debug("Persistence manager stopped")
|
||
}
|
||
|
||
if rc.fallbackManager != nil {
|
||
rc.fallbackManager.Stop()
|
||
rc.logger.Debug("Fallback manager stopped")
|
||
}
|
||
if rc.panicRecoveryMgr != nil {
|
||
rc.panicRecoveryMgr.Stop()
|
||
rc.logger.Debug("Panic recovery manager stopped")
|
||
}
|
||
|
||
close(rc.stopChan)
|
||
if rc.raft != nil {
|
||
rc.raft.Shutdown()
|
||
}
|
||
|
||
rc.logger.Info(fmt.Sprintf("Raft coordinator stopped at %s", time.UnixMilli(now).Format("2006-01-02 15:04:05.000")))
|
||
}
|
||
|
||
// IsReplicationEnabled возвращает статус репликации
|
||
func (rc *RaftCoordinator) IsReplicationEnabled() bool {
|
||
return rc.replicationEnabled
|
||
}
|
||
|
||
// IsMasterMasterEnabled возвращает статус мастер-мастер репликации
|
||
func (rc *RaftCoordinator) IsMasterMasterEnabled() bool {
|
||
return rc.masterMasterEnabled
|
||
}
|
||
|
||
// IsSyncReplicationEnabled возвращает статус синхронной репликации
|
||
func (rc *RaftCoordinator) IsSyncReplicationEnabled() bool {
|
||
return rc.syncReplication
|
||
}
|
||
|
||
// GetClusterHealth возвращает детальную информацию о здоровье кластера
|
||
func (rc *RaftCoordinator) GetClusterHealth() *ClusterHealth {
|
||
now := time.Now().UnixMilli()
|
||
|
||
health := &ClusterHealth{
|
||
Nodes: make(map[string]*NodeHealth),
|
||
OverallScore: 100.0,
|
||
CheckedAt: now,
|
||
Recommendations: "",
|
||
}
|
||
|
||
state := rc.fsm.state
|
||
state.mu.RLock()
|
||
defer state.mu.RUnlock()
|
||
|
||
for nodeID, nodeInfo := range state.Nodes {
|
||
nodeHealth := &NodeHealth{
|
||
Status: nodeInfo.Status,
|
||
LatencyMs: 0,
|
||
LastCheck: now,
|
||
LastSuccess: nodeInfo.LastSeen,
|
||
FailureCount: 0,
|
||
}
|
||
|
||
if now-nodeInfo.LastSeen > 30000 {
|
||
nodeHealth.Status = "offline"
|
||
nodeHealth.LastFailure = nodeInfo.LastSeen
|
||
nodeHealth.FailureCount++
|
||
health.OverallScore -= 10
|
||
} else if nodeInfo.Status == "syncing" {
|
||
health.OverallScore -= 5
|
||
} else {
|
||
nodeHealth.LastSuccess = nodeInfo.LastSeen
|
||
}
|
||
|
||
if rc.splitBrainDetector.IsQuarantined(nodeID) {
|
||
nodeHealth.Status = "quarantined"
|
||
health.OverallScore -= 20
|
||
}
|
||
|
||
health.Nodes[nodeID] = nodeHealth
|
||
}
|
||
|
||
if rc.fallbackManager != nil && rc.fallbackManager.IsFallbackMode() {
|
||
health.OverallScore -= 30
|
||
health.Recommendations += " WARNING: Cluster is in fallback mode due to leader loss."
|
||
}
|
||
|
||
if rc.splitBrainDetector.Detect(rc.currentTerm.Load(), "", len(state.Nodes)) {
|
||
health.OverallScore = 0
|
||
health.Recommendations = "CRITICAL: Split-brain detected! Manual intervention required."
|
||
} else if health.OverallScore < 50 {
|
||
health.Recommendations = "Critical: Check network connectivity and node health immediately"
|
||
} else if health.OverallScore < 80 {
|
||
health.Recommendations = "Warning: Some nodes are offline or syncing, consider adding more nodes"
|
||
} else {
|
||
health.Recommendations = "Cluster is healthy, all systems operational"
|
||
}
|
||
|
||
return health
|
||
}
|