diff --git a/internal/cluster/raft_coordinator.go b/internal/cluster/raft_coordinator.go deleted file mode 100644 index b28d88c..0000000 --- a/internal/cluster/raft_coordinator.go +++ /dev/null @@ -1,3072 +0,0 @@ -/* - * 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 -}