Delete internal/cluster/node.go
This commit is contained in:
@@ -1,974 +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/node.go
|
|
||||||
// Назначение: Реализация узла кластера (node) для распределённой СУБД с поддержкой временных меток.
|
|
||||||
// Полностью lock-free с использованием атомарных операций.
|
|
||||||
|
|
||||||
package cluster
|
|
||||||
|
|
||||||
import (
|
|
||||||
"crypto/rand"
|
|
||||||
"encoding/base64"
|
|
||||||
"encoding/json"
|
|
||||||
"fmt"
|
|
||||||
"io"
|
|
||||||
"net"
|
|
||||||
"runtime/debug"
|
|
||||||
"sync"
|
|
||||||
"sync/atomic"
|
|
||||||
"time"
|
|
||||||
|
|
||||||
"futriis/internal/log"
|
|
||||||
"futriis/internal/storage"
|
|
||||||
"github.com/google/uuid"
|
|
||||||
)
|
|
||||||
|
|
||||||
// NodeStatus представляет состояние узла кластера
|
|
||||||
type NodeStatus int32
|
|
||||||
|
|
||||||
const (
|
|
||||||
StatusOffline NodeStatus = iota
|
|
||||||
StatusActive
|
|
||||||
StatusSyncing
|
|
||||||
StatusFailed
|
|
||||||
)
|
|
||||||
|
|
||||||
// Node представляет отдельный узел в распределённой системе
|
|
||||||
type Node struct {
|
|
||||||
ID string // Уникальный идентификатор узла
|
|
||||||
IP string // IP-адрес узла
|
|
||||||
Port int // Порт для коммуникации
|
|
||||||
Status atomic.Int32 // Атомарный статус узла (NodeStatus)
|
|
||||||
Storage *storage.Storage
|
|
||||||
logger *log.Logger
|
|
||||||
coordinator *RaftCoordinator // Ссылка на координатора (теперь RaftCoordinator)
|
|
||||||
lastSeen atomic.Int64 // Время последнего heartbeat (Unix миллисекунды)
|
|
||||||
joinedAt atomic.Int64 // Время присоединения к кластеру
|
|
||||||
createdAt int64 // Время создания узла
|
|
||||||
startedAt int64 // Время последнего старта узла
|
|
||||||
stoppedAt int64 // Время остановки узла
|
|
||||||
incomingConn chan net.Conn // Канал для входящих соединений (wait-free)
|
|
||||||
stopChan chan struct{}
|
|
||||||
requestCount atomic.Uint64 // Счётчик обработанных запросов
|
|
||||||
bytesRx atomic.Uint64 // Получено байт
|
|
||||||
bytesTx atomic.Uint64 // Отправлено байт
|
|
||||||
mu sync.RWMutex // Для синхронизации изменений статуса
|
|
||||||
|
|
||||||
// Новые компоненты для production-ready реализации
|
|
||||||
workerPool *WorkerPool // Пул воркеров для ограничения горутин
|
|
||||||
replicator *NetworkReplicator // Сетевой репликатор с retry и backoff
|
|
||||||
connPool sync.Map // Пул соединений к другим узлам
|
|
||||||
}
|
|
||||||
|
|
||||||
// NodeConfig содержит конфигурацию для создания узла
|
|
||||||
type NodeConfig struct {
|
|
||||||
IP string
|
|
||||||
Port int
|
|
||||||
Storage *storage.Storage
|
|
||||||
Logger *log.Logger
|
|
||||||
Coordinator *RaftCoordinator
|
|
||||||
}
|
|
||||||
|
|
||||||
// NewNode создаёт новый экземпляр узла кластера
|
|
||||||
func NewNode(ip string, port int, store *storage.Storage, logger *log.Logger) *Node {
|
|
||||||
now := time.Now().UnixMilli()
|
|
||||||
|
|
||||||
// Создаём пул воркеров с ограничением (максимум 500 одновременных горутин)
|
|
||||||
workerPool := NewWorkerPool(500, logger)
|
|
||||||
|
|
||||||
// Создаём сетевой репликатор с настройками retry и backoff
|
|
||||||
replicator := NewNetworkReplicator(DefaultReplicationRetryConfig(), workerPool, logger)
|
|
||||||
|
|
||||||
node := &Node{
|
|
||||||
ID: uuid.New().String(),
|
|
||||||
IP: ip,
|
|
||||||
Port: port,
|
|
||||||
Storage: store,
|
|
||||||
logger: logger,
|
|
||||||
incomingConn: make(chan net.Conn, 10000), // Увеличенный буфер
|
|
||||||
stopChan: make(chan struct{}),
|
|
||||||
createdAt: now,
|
|
||||||
startedAt: now,
|
|
||||||
workerPool: workerPool,
|
|
||||||
replicator: replicator,
|
|
||||||
}
|
|
||||||
node.Status.Store(int32(StatusActive))
|
|
||||||
node.lastSeen.Store(now)
|
|
||||||
node.joinedAt.Store(0)
|
|
||||||
|
|
||||||
// Безопасный запуск сервера с обработкой паник
|
|
||||||
SafeGoWithLogger(node.startTCPServer, logger, "TCPServer")
|
|
||||||
SafeGoWithLogger(node.handleIncomingConnections, logger, "IncomingConnections")
|
|
||||||
SafeGoWithLogger(node.heartbeatLoop, logger, "HeartbeatLoop")
|
|
||||||
SafeGoWithLogger(node.connectionHealthMonitor, logger, "ConnectionHealthMonitor")
|
|
||||||
|
|
||||||
if logger != nil {
|
|
||||||
logger.Info(fmt.Sprintf("Node %s created at %s with worker pool (max: 500)", node.ID, node.GetCreatedAtStr()))
|
|
||||||
}
|
|
||||||
|
|
||||||
return node
|
|
||||||
}
|
|
||||||
|
|
||||||
// startTCPServer запускает TCP-сервер для приёма запросов от других узлов
|
|
||||||
func (n *Node) startTCPServer() {
|
|
||||||
defer func() {
|
|
||||||
if r := recover(); r != nil {
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Error(fmt.Sprintf("TCP server panicked: %v\n%s", r, debug.Stack()))
|
|
||||||
}
|
|
||||||
// Перезапускаем сервер
|
|
||||||
time.Sleep(5 * time.Second)
|
|
||||||
SafeGoWithLogger(n.startTCPServer, n.logger, "TCPServer")
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
addr := fmt.Sprintf("%s:%d", n.IP, n.Port)
|
|
||||||
listener, err := net.Listen("tcp", addr)
|
|
||||||
if err != nil {
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Error(fmt.Sprintf("Node %s failed to start TCP server: %v", n.ID, err))
|
|
||||||
}
|
|
||||||
n.Status.Store(int32(StatusFailed))
|
|
||||||
return
|
|
||||||
}
|
|
||||||
defer listener.Close()
|
|
||||||
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Info(fmt.Sprintf("Node %s listening on %s (started at %s)", n.ID, addr, n.GetStartedAtStr()))
|
|
||||||
}
|
|
||||||
|
|
||||||
for {
|
|
||||||
select {
|
|
||||||
case <-n.stopChan:
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Info(fmt.Sprintf("Node %s TCP server stopped", n.ID))
|
|
||||||
}
|
|
||||||
return
|
|
||||||
default:
|
|
||||||
// Устанавливаем таймаут для Accept
|
|
||||||
if err := listener.(*net.TCPListener).SetDeadline(time.Now().Add(5 * time.Second)); err != nil {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
conn, err := listener.Accept()
|
|
||||||
if err != nil {
|
|
||||||
if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Error(fmt.Sprintf("Node %s accept error: %v", n.ID, err))
|
|
||||||
}
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
// Устанавливаем таймауты на соединение
|
|
||||||
conn.SetReadDeadline(time.Now().Add(30 * time.Second))
|
|
||||||
conn.SetWriteDeadline(time.Now().Add(30 * time.Second))
|
|
||||||
|
|
||||||
select {
|
|
||||||
case n.incomingConn <- conn:
|
|
||||||
n.bytesRx.Add(1)
|
|
||||||
default:
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Warn(fmt.Sprintf("Node %s incoming connection queue full, dropping connection", n.ID))
|
|
||||||
}
|
|
||||||
conn.Close()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// handleIncomingConnections обрабатывает входящие соединения с использованием пула воркеров
|
|
||||||
func (n *Node) handleIncomingConnections() {
|
|
||||||
defer func() {
|
|
||||||
if r := recover(); r != nil {
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Error(fmt.Sprintf("Incoming connections handler panicked: %v\n%s", r, debug.Stack()))
|
|
||||||
}
|
|
||||||
// Перезапускаем обработчик
|
|
||||||
SafeGoWithLogger(n.handleIncomingConnections, n.logger, "IncomingConnections")
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
for {
|
|
||||||
select {
|
|
||||||
case <-n.stopChan:
|
|
||||||
return
|
|
||||||
case conn := <-n.incomingConn:
|
|
||||||
n.requestCount.Add(1)
|
|
||||||
// Используем пул воркеров вместо создания неограниченного количества горутин
|
|
||||||
taskID := fmt.Sprintf("handle_conn_%d_%s", time.Now().UnixNano(), conn.RemoteAddr().String())
|
|
||||||
err := n.workerPool.SubmitFunc(taskID, func() error {
|
|
||||||
n.handleNodeRequest(conn)
|
|
||||||
return nil
|
|
||||||
})
|
|
||||||
if err != nil {
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Warn(fmt.Sprintf("Failed to submit connection task: %v", err))
|
|
||||||
}
|
|
||||||
conn.Close()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// handleNodeRequest обрабатывает конкретный запрос от другого узла
|
|
||||||
func (n *Node) handleNodeRequest(conn net.Conn) {
|
|
||||||
defer func() {
|
|
||||||
if r := recover(); r != nil {
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Error(fmt.Sprintf("Request handler panicked: %v\n%s", r, debug.Stack()))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
conn.Close()
|
|
||||||
}()
|
|
||||||
|
|
||||||
// Устанавливаем таймаут на чтение
|
|
||||||
conn.SetReadDeadline(time.Now().Add(30 * time.Second))
|
|
||||||
|
|
||||||
decoder := json.NewDecoder(conn)
|
|
||||||
var req NodeRequest
|
|
||||||
if err := decoder.Decode(&req); err != nil {
|
|
||||||
if err != io.EOF && n.logger != nil {
|
|
||||||
n.logger.Error(fmt.Sprintf("Node %s failed to decode request: %v", n.ID, err))
|
|
||||||
}
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
n.lastSeen.Store(time.Now().UnixMilli())
|
|
||||||
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Debug(fmt.Sprintf("Node %s received request type %s from %s at %s",
|
|
||||||
n.ID, req.Type, req.FromNode, time.UnixMilli(req.Timestamp).Format("15:04:05.000")))
|
|
||||||
}
|
|
||||||
|
|
||||||
switch req.Type {
|
|
||||||
case "replicate":
|
|
||||||
n.handleReplicateRequest(req.Data)
|
|
||||||
case "query":
|
|
||||||
n.handleQueryRequest(req.Data, conn)
|
|
||||||
case "sync":
|
|
||||||
n.handleSyncRequest(req.Data, conn)
|
|
||||||
case "heartbeat":
|
|
||||||
n.handleHeartbeatRequest(req, conn)
|
|
||||||
case "status_sync":
|
|
||||||
n.handleStatusSyncRequest(req, conn)
|
|
||||||
default:
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Warn(fmt.Sprintf("Node %s unknown request type: %s", n.ID, req.Type))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// handleReplicateRequest обрабатывает запрос на репликацию документа
|
|
||||||
func (n *Node) handleReplicateRequest(data []byte) {
|
|
||||||
defer func() {
|
|
||||||
if r := recover(); r != nil {
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Error(fmt.Sprintf("Replicate request handler panicked: %v\n%s", r, debug.Stack()))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
startTime := time.Now().UnixMilli()
|
|
||||||
|
|
||||||
var repData struct {
|
|
||||||
Database string `json:"database"`
|
|
||||||
Collection string `json:"collection"`
|
|
||||||
Document map[string]interface{} `json:"document"`
|
|
||||||
SourceNode string `json:"source_node"`
|
|
||||||
ReplicaID string `json:"replica_id"`
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := json.Unmarshal(data, &repData); err != nil {
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Error(fmt.Sprintf("Node %s failed to unmarshal replicate data: %v", n.ID, err))
|
|
||||||
}
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
db, err := n.Storage.GetDatabase(repData.Database)
|
|
||||||
if err != nil {
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Error(fmt.Sprintf("Node %s database not found for replication: %s", n.ID, repData.Database))
|
|
||||||
}
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
coll, err := db.GetCollection(repData.Collection)
|
|
||||||
if err != nil {
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Error(fmt.Sprintf("Node %s collection not found for replication: %s", n.ID, repData.Collection))
|
|
||||||
}
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
// Безопасное получение ID документа
|
|
||||||
docID, ok := repData.Document["_id"].(string)
|
|
||||||
if !ok {
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Error(fmt.Sprintf("Node %s document missing _id field", n.ID))
|
|
||||||
}
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
doc := &storage.Document{
|
|
||||||
ID: docID,
|
|
||||||
Fields: repData.Document,
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := coll.Insert(doc); err != nil {
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Error(fmt.Sprintf("Node %s failed to replicate document: %v", n.ID, err))
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
duration := time.Now().UnixMilli() - startTime
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Debug(fmt.Sprintf("Node %s replicated document %s from %s (took %d ms)",
|
|
||||||
n.ID, doc.ID, repData.SourceNode, duration))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// handleQueryRequest обрабатывает запрос на чтение данных с узла
|
|
||||||
func (n *Node) handleQueryRequest(data []byte, conn net.Conn) {
|
|
||||||
defer func() {
|
|
||||||
if r := recover(); r != nil {
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Error(fmt.Sprintf("Query request handler panicked: %v\n%s", r, debug.Stack()))
|
|
||||||
}
|
|
||||||
n.sendErrorResponse(conn, "Internal server error")
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
startTime := time.Now().UnixMilli()
|
|
||||||
|
|
||||||
var queryData struct {
|
|
||||||
Database string `json:"database"`
|
|
||||||
Collection string `json:"collection"`
|
|
||||||
DocumentID string `json:"document_id"`
|
|
||||||
RequestID string `json:"request_id"`
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := json.Unmarshal(data, &queryData); err != nil {
|
|
||||||
n.sendErrorResponse(conn, err.Error())
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
db, err := n.Storage.GetDatabase(queryData.Database)
|
|
||||||
if err != nil {
|
|
||||||
n.sendErrorResponse(conn, err.Error())
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
coll, err := db.GetCollection(queryData.Collection)
|
|
||||||
if err != nil {
|
|
||||||
n.sendErrorResponse(conn, err.Error())
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
doc, err := coll.Find(queryData.DocumentID)
|
|
||||||
if err != nil {
|
|
||||||
n.sendErrorResponse(conn, err.Error())
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
duration := time.Now().UnixMilli() - startTime
|
|
||||||
|
|
||||||
response := map[string]interface{}{
|
|
||||||
"status": "success",
|
|
||||||
"data": doc,
|
|
||||||
"node_id": n.ID,
|
|
||||||
"request_id": queryData.RequestID,
|
|
||||||
"duration_ms": duration,
|
|
||||||
"timestamp": time.Now().UnixMilli(),
|
|
||||||
}
|
|
||||||
|
|
||||||
conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
|
|
||||||
encoder := json.NewEncoder(conn)
|
|
||||||
if err := encoder.Encode(response); err == nil {
|
|
||||||
responseData, _ := json.Marshal(response)
|
|
||||||
n.bytesTx.Add(uint64(len(responseData)))
|
|
||||||
}
|
|
||||||
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Debug(fmt.Sprintf("Node %s handled query for %s.%s:%s (took %d ms)",
|
|
||||||
n.ID, queryData.Database, queryData.Collection, queryData.DocumentID, duration))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// handleSyncRequest обрабатывает запрос на синхронизацию всей коллекции
|
|
||||||
func (n *Node) handleSyncRequest(data []byte, conn net.Conn) {
|
|
||||||
defer func() {
|
|
||||||
if r := recover(); r != nil {
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Error(fmt.Sprintf("Sync request handler panicked: %v\n%s", r, debug.Stack()))
|
|
||||||
}
|
|
||||||
n.sendErrorResponse(conn, "Internal server error")
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
startTime := time.Now().UnixMilli()
|
|
||||||
|
|
||||||
var syncData struct {
|
|
||||||
Database string `json:"database"`
|
|
||||||
Collection string `json:"collection"`
|
|
||||||
RequestID string `json:"request_id"`
|
|
||||||
Since int64 `json:"since"`
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := json.Unmarshal(data, &syncData); err != nil {
|
|
||||||
n.sendErrorResponse(conn, err.Error())
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
db, err := n.Storage.GetDatabase(syncData.Database)
|
|
||||||
if err != nil {
|
|
||||||
n.sendErrorResponse(conn, err.Error())
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
coll, err := db.GetCollection(syncData.Collection)
|
|
||||||
if err != nil {
|
|
||||||
n.sendErrorResponse(conn, err.Error())
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
docs := coll.GetAllDocuments()
|
|
||||||
|
|
||||||
if syncData.Since > 0 {
|
|
||||||
filtered := make([]*storage.Document, 0)
|
|
||||||
for _, doc := range docs {
|
|
||||||
if doc.UpdatedAt > syncData.Since {
|
|
||||||
filtered = append(filtered, doc)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
docs = filtered
|
|
||||||
}
|
|
||||||
|
|
||||||
duration := time.Now().UnixMilli() - startTime
|
|
||||||
|
|
||||||
response := map[string]interface{}{
|
|
||||||
"status": "success",
|
|
||||||
"docs": docs,
|
|
||||||
"count": len(docs),
|
|
||||||
"node_id": n.ID,
|
|
||||||
"request_id": syncData.RequestID,
|
|
||||||
"duration_ms": duration,
|
|
||||||
"timestamp": time.Now().UnixMilli(),
|
|
||||||
"sync_duration_ms": duration,
|
|
||||||
}
|
|
||||||
|
|
||||||
conn.SetWriteDeadline(time.Now().Add(30 * time.Second))
|
|
||||||
encoder := json.NewEncoder(conn)
|
|
||||||
if err := encoder.Encode(response); err != nil && n.logger != nil {
|
|
||||||
n.logger.Error(fmt.Sprintf("Failed to send sync response: %v", err))
|
|
||||||
}
|
|
||||||
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Info(fmt.Sprintf("Node %s synced %d documents from %s.%s (took %d ms)",
|
|
||||||
n.ID, len(docs), syncData.Database, syncData.Collection, duration))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// handleHeartbeatRequest обрабатывает heartbeat запрос
|
|
||||||
func (n *Node) handleHeartbeatRequest(req NodeRequest, conn net.Conn) {
|
|
||||||
n.lastSeen.Store(time.Now().UnixMilli())
|
|
||||||
|
|
||||||
response := map[string]interface{}{
|
|
||||||
"status": "alive",
|
|
||||||
"node_id": n.ID,
|
|
||||||
"timestamp": time.Now().UnixMilli(),
|
|
||||||
"uptime_ms": time.Now().UnixMilli() - n.startedAt,
|
|
||||||
}
|
|
||||||
|
|
||||||
conn.SetWriteDeadline(time.Now().Add(5 * time.Second))
|
|
||||||
encoder := json.NewEncoder(conn)
|
|
||||||
encoder.Encode(response)
|
|
||||||
}
|
|
||||||
|
|
||||||
// handleStatusSyncRequest обрабатывает запрос синхронизации статуса (для split-brain recovery)
|
|
||||||
func (n *Node) handleStatusSyncRequest(req NodeRequest, conn net.Conn) {
|
|
||||||
var syncStatus struct {
|
|
||||||
LeaderID string `json:"leader_id"`
|
|
||||||
Term uint64 `json:"term"`
|
|
||||||
ClusterSize int `json:"cluster_size"`
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := json.Unmarshal(req.Data, &syncStatus); err != nil {
|
|
||||||
n.sendErrorResponse(conn, err.Error())
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
if n.coordinator != nil {
|
|
||||||
n.coordinator.HandleStatusSync(syncStatus.LeaderID, syncStatus.Term, syncStatus.ClusterSize)
|
|
||||||
}
|
|
||||||
|
|
||||||
response := map[string]interface{}{
|
|
||||||
"status": "synced",
|
|
||||||
"node_id": n.ID,
|
|
||||||
"term": n.coordinator.GetCurrentTerm(),
|
|
||||||
"is_leader": n.coordinator.IsLeader(),
|
|
||||||
"timestamp": time.Now().UnixMilli(),
|
|
||||||
}
|
|
||||||
|
|
||||||
conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
|
|
||||||
encoder := json.NewEncoder(conn)
|
|
||||||
encoder.Encode(response)
|
|
||||||
}
|
|
||||||
|
|
||||||
// sendErrorResponse отправляет ошибку в ответ на запрос
|
|
||||||
func (n *Node) sendErrorResponse(conn net.Conn, errMsg string) {
|
|
||||||
response := map[string]interface{}{
|
|
||||||
"status": "error",
|
|
||||||
"error": errMsg,
|
|
||||||
"node_id": n.ID,
|
|
||||||
"timestamp": time.Now().UnixMilli(),
|
|
||||||
}
|
|
||||||
|
|
||||||
conn.SetWriteDeadline(time.Now().Add(5 * time.Second))
|
|
||||||
encoder := json.NewEncoder(conn)
|
|
||||||
encoder.Encode(response)
|
|
||||||
}
|
|
||||||
|
|
||||||
// heartbeatLoop отправляет периодические сигналы жизни координатору
|
|
||||||
func (n *Node) heartbeatLoop() {
|
|
||||||
defer func() {
|
|
||||||
if r := recover(); r != nil {
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Error(fmt.Sprintf("Heartbeat loop panicked: %v\n%s", r, debug.Stack()))
|
|
||||||
}
|
|
||||||
// Перезапускаем heartbeat loop
|
|
||||||
time.Sleep(5 * time.Second)
|
|
||||||
SafeGoWithLogger(n.heartbeatLoop, n.logger, "HeartbeatLoop")
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
ticker := time.NewTicker(5 * time.Second)
|
|
||||||
defer ticker.Stop()
|
|
||||||
|
|
||||||
for {
|
|
||||||
select {
|
|
||||||
case <-n.stopChan:
|
|
||||||
return
|
|
||||||
case <-ticker.C:
|
|
||||||
if n.coordinator != nil {
|
|
||||||
n.coordinator.SendHeartbeat(n.ID)
|
|
||||||
n.lastSeen.Store(time.Now().UnixMilli())
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Debug(fmt.Sprintf("Node %s sent heartbeat at %s", n.ID, n.GetLastSeenStr()))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// connectionHealthMonitor мониторит здоровье соединений
|
|
||||||
func (n *Node) connectionHealthMonitor() {
|
|
||||||
defer func() {
|
|
||||||
if r := recover(); r != nil {
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Error(fmt.Sprintf("Connection health monitor panicked: %v\n%s", r, debug.Stack()))
|
|
||||||
}
|
|
||||||
SafeGoWithLogger(n.connectionHealthMonitor, n.logger, "ConnectionHealthMonitor")
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
ticker := time.NewTicker(30 * time.Second)
|
|
||||||
defer ticker.Stop()
|
|
||||||
|
|
||||||
for {
|
|
||||||
select {
|
|
||||||
case <-n.stopChan:
|
|
||||||
return
|
|
||||||
case <-ticker.C:
|
|
||||||
n.cleanupStaleConnections()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// cleanupStaleConnections очищает старые соединения
|
|
||||||
func (n *Node) cleanupStaleConnections() {
|
|
||||||
n.connPool.Range(func(key, value interface{}) bool {
|
|
||||||
if conn, ok := value.(net.Conn); ok {
|
|
||||||
// Проверяем, активно ли соединение
|
|
||||||
conn.SetReadDeadline(time.Now().Add(1 * time.Second))
|
|
||||||
buf := make([]byte, 1)
|
|
||||||
_, err := conn.Read(buf)
|
|
||||||
if err != nil {
|
|
||||||
conn.Close()
|
|
||||||
n.connPool.Delete(key)
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Debug(fmt.Sprintf("Cleaned up stale connection for %v", key))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return true
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetNodeStatus возвращает текущий статус узла (атомарно)
|
|
||||||
func (n *Node) GetNodeStatus() NodeStatus {
|
|
||||||
return NodeStatus(n.Status.Load())
|
|
||||||
}
|
|
||||||
|
|
||||||
// IsActive проверяет, активен ли узел
|
|
||||||
func (n *Node) IsActive() bool {
|
|
||||||
return NodeStatus(n.Status.Load()) == StatusActive
|
|
||||||
}
|
|
||||||
|
|
||||||
// SetStatus устанавливает статус узла с синхронизацией через Raft
|
|
||||||
func (n *Node) SetStatus(status NodeStatus) error {
|
|
||||||
n.mu.Lock()
|
|
||||||
defer n.mu.Unlock()
|
|
||||||
|
|
||||||
oldStatus := n.Status.Load()
|
|
||||||
if oldStatus == int32(status) {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
if n.coordinator != nil && n.coordinator.IsLeader() {
|
|
||||||
if err := n.coordinator.UpdateNodeStatus(n.ID, status); err != nil {
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Error(fmt.Sprintf("Failed to update node status via Raft: %v", err))
|
|
||||||
}
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
n.Status.Store(int32(status))
|
|
||||||
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Info(fmt.Sprintf("Node %s status changed from %d to %d at %s",
|
|
||||||
n.ID, oldStatus, status, time.Now().Format("2006-01-02 15:04:05.000")))
|
|
||||||
}
|
|
||||||
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// SetCoordinator устанавливает координатора для узла
|
|
||||||
func (n *Node) SetCoordinator(coord *RaftCoordinator) {
|
|
||||||
n.coordinator = coord
|
|
||||||
now := time.Now().UnixMilli()
|
|
||||||
n.joinedAt.Store(now)
|
|
||||||
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Info(fmt.Sprintf("Node %s joined cluster at %s", n.ID, n.GetJoinedAtStr()))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// JoinCluster присоединяет узел к кластеру
|
|
||||||
func (n *Node) JoinCluster(coord *RaftCoordinator) error {
|
|
||||||
if n.coordinator != nil {
|
|
||||||
return fmt.Errorf("node already joined to cluster")
|
|
||||||
}
|
|
||||||
|
|
||||||
n.SetCoordinator(coord)
|
|
||||||
|
|
||||||
if err := coord.RegisterNode(n); err != nil {
|
|
||||||
return fmt.Errorf("failed to register node: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := n.SetStatus(StatusActive); err != nil {
|
|
||||||
return fmt.Errorf("failed to set active status: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Info(fmt.Sprintf("Node %s successfully joined cluster at %s", n.ID, n.GetJoinedAtStr()))
|
|
||||||
}
|
|
||||||
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// LeaveCluster покидает кластер
|
|
||||||
func (n *Node) LeaveCluster() error {
|
|
||||||
if n.coordinator == nil {
|
|
||||||
return fmt.Errorf("node not in cluster")
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := n.SetStatus(StatusOffline); err != nil {
|
|
||||||
n.logger.Warn(fmt.Sprintf("Failed to set offline status: %v", err))
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := n.coordinator.RemoveNode(n.ID); err != nil {
|
|
||||||
n.logger.Warn(fmt.Sprintf("Failed to remove node from coordinator: %v", err))
|
|
||||||
}
|
|
||||||
|
|
||||||
n.coordinator = nil
|
|
||||||
n.joinedAt.Store(0)
|
|
||||||
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Info(fmt.Sprintf("Node %s left cluster at %s", n.ID, time.Now().Format("2006-01-02 15:04:05.000")))
|
|
||||||
}
|
|
||||||
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetLastSeen возвращает время последнего контакта
|
|
||||||
func (n *Node) GetLastSeen() int64 {
|
|
||||||
return n.lastSeen.Load()
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetLastSeenStr возвращает человекочитаемое время последнего контакта
|
|
||||||
func (n *Node) GetLastSeenStr() string {
|
|
||||||
lastSeen := n.lastSeen.Load()
|
|
||||||
if lastSeen == 0 {
|
|
||||||
return "never"
|
|
||||||
}
|
|
||||||
return time.UnixMilli(lastSeen).Format("2006-01-02 15:04:05.000")
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetJoinedAt возвращает время присоединения к кластеру
|
|
||||||
func (n *Node) GetJoinedAt() int64 {
|
|
||||||
return n.joinedAt.Load()
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetJoinedAtStr возвращает человекочитаемое время присоединения
|
|
||||||
func (n *Node) GetJoinedAtStr() string {
|
|
||||||
joinedAt := n.joinedAt.Load()
|
|
||||||
if joinedAt == 0 {
|
|
||||||
return "not joined"
|
|
||||||
}
|
|
||||||
return time.UnixMilli(joinedAt).Format("2006-01-02 15:04:05.000")
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetStartedAt возвращает время старта узла
|
|
||||||
func (n *Node) GetStartedAt() int64 {
|
|
||||||
return n.startedAt
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetStartedAtStr возвращает человекочитаемое время старта
|
|
||||||
func (n *Node) GetStartedAtStr() string {
|
|
||||||
return time.UnixMilli(n.startedAt).Format("2006-01-02 15:04:05.000")
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetCreatedAt возвращает время создания узла
|
|
||||||
func (n *Node) GetCreatedAt() int64 {
|
|
||||||
return n.createdAt
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetCreatedAtStr возвращает человекочитаемое время создания
|
|
||||||
func (n *Node) GetCreatedAtStr() string {
|
|
||||||
return time.UnixMilli(n.createdAt).Format("2006-01-02 15:04:05.000")
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetUptime возвращает время работы узла
|
|
||||||
func (n *Node) GetUptime() time.Duration {
|
|
||||||
if n.startedAt == 0 {
|
|
||||||
return 0
|
|
||||||
}
|
|
||||||
return time.Duration(time.Now().UnixMilli()-n.startedAt) * time.Millisecond
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetStats возвращает статистику узла
|
|
||||||
func (n *Node) GetStats() map[string]interface{} {
|
|
||||||
stats := map[string]interface{}{
|
|
||||||
"id": n.ID,
|
|
||||||
"ip": n.IP,
|
|
||||||
"port": n.Port,
|
|
||||||
"status": n.GetNodeStatus(),
|
|
||||||
"created_at": n.GetCreatedAtStr(),
|
|
||||||
"started_at": n.GetStartedAtStr(),
|
|
||||||
"joined_at": n.GetJoinedAtStr(),
|
|
||||||
"last_seen": n.GetLastSeenStr(),
|
|
||||||
"uptime": n.GetUptime().String(),
|
|
||||||
"request_count": n.requestCount.Load(),
|
|
||||||
"bytes_rx": n.bytesRx.Load(),
|
|
||||||
"bytes_tx": n.bytesTx.Load(),
|
|
||||||
}
|
|
||||||
|
|
||||||
if n.workerPool != nil {
|
|
||||||
stats["worker_pool"] = n.workerPool.GetStats()
|
|
||||||
}
|
|
||||||
|
|
||||||
if n.replicator != nil {
|
|
||||||
stats["replication"] = n.replicator.GetStats()
|
|
||||||
}
|
|
||||||
|
|
||||||
return stats
|
|
||||||
}
|
|
||||||
|
|
||||||
// Stop останавливает работу узла
|
|
||||||
func (n *Node) Stop() {
|
|
||||||
if n.coordinator != nil && n.coordinator.IsLeader() {
|
|
||||||
n.SetStatus(StatusOffline)
|
|
||||||
}
|
|
||||||
n.Status.Store(int32(StatusOffline))
|
|
||||||
n.stoppedAt = time.Now().UnixMilli()
|
|
||||||
|
|
||||||
close(n.stopChan)
|
|
||||||
|
|
||||||
if n.workerPool != nil {
|
|
||||||
n.workerPool.Stop()
|
|
||||||
}
|
|
||||||
|
|
||||||
if n.replicator != nil {
|
|
||||||
n.replicator.Close()
|
|
||||||
}
|
|
||||||
|
|
||||||
// Закрываем все соединения в пуле
|
|
||||||
n.connPool.Range(func(key, value interface{}) bool {
|
|
||||||
if conn, ok := value.(net.Conn); ok {
|
|
||||||
conn.Close()
|
|
||||||
}
|
|
||||||
return true
|
|
||||||
})
|
|
||||||
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Info(fmt.Sprintf("Node %s stopped at %s", n.ID, time.UnixMilli(n.stoppedAt).Format("2006-01-02 15:04:05.000")))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetAddress возвращает адрес узла в формате "ip:port"
|
|
||||||
func (n *Node) GetAddress() string {
|
|
||||||
return fmt.Sprintf("%s:%d", n.IP, n.Port)
|
|
||||||
}
|
|
||||||
|
|
||||||
// generateReplicationID генерирует уникальный ID для репликации
|
|
||||||
func generateReplicationID() string {
|
|
||||||
bytes := make([]byte, 16)
|
|
||||||
rand.Read(bytes)
|
|
||||||
return base64.URLEncoding.EncodeToString(bytes)
|
|
||||||
}
|
|
||||||
|
|
||||||
// ReplicateDocument отправляет документ на репликацию всем активным узлам с retry и backoff
|
|
||||||
func (n *Node) ReplicateDocument(database, collection string, doc *storage.Document) error {
|
|
||||||
if n.coordinator == nil {
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Warn("No coordinator set, skipping replication")
|
|
||||||
}
|
|
||||||
return fmt.Errorf("no coordinator set")
|
|
||||||
}
|
|
||||||
|
|
||||||
nodes := n.coordinator.GetActiveNodes()
|
|
||||||
if len(nodes) <= 1 {
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Debug("No other nodes for replication")
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// Подготавливаем данные для репликации
|
|
||||||
repData := struct {
|
|
||||||
Database string `json:"database"`
|
|
||||||
Collection string `json:"collection"`
|
|
||||||
Document map[string]interface{} `json:"document"`
|
|
||||||
SourceNode string `json:"source_node"`
|
|
||||||
ReplicaID string `json:"replica_id"`
|
|
||||||
}{
|
|
||||||
Database: database,
|
|
||||||
Collection: collection,
|
|
||||||
Document: doc.GetFields(),
|
|
||||||
SourceNode: n.ID,
|
|
||||||
ReplicaID: generateReplicationID(),
|
|
||||||
}
|
|
||||||
|
|
||||||
data, err := json.Marshal(repData)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("failed to marshal replication data: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
startTime := time.Now().UnixMilli()
|
|
||||||
|
|
||||||
// Отправляем на все узлы, кроме себя
|
|
||||||
var wg sync.WaitGroup
|
|
||||||
var failedCount atomic.Int32
|
|
||||||
var successCount atomic.Int32
|
|
||||||
|
|
||||||
for _, nodeInfo := range nodes {
|
|
||||||
if nodeInfo.ID == n.ID {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
wg.Add(1)
|
|
||||||
targetNodeID := nodeInfo.ID
|
|
||||||
targetAddress := fmt.Sprintf("%s:%d", nodeInfo.IP, nodeInfo.Port)
|
|
||||||
|
|
||||||
// Асинхронная репликация с повторными попытками через пул воркеров
|
|
||||||
taskID := fmt.Sprintf("replicate_%s_to_%s_%s", doc.ID, targetNodeID, repData.ReplicaID)
|
|
||||||
err := n.workerPool.SubmitFunc(taskID, func() error {
|
|
||||||
defer wg.Done()
|
|
||||||
|
|
||||||
if n.replicator == nil {
|
|
||||||
failedCount.Add(1)
|
|
||||||
return fmt.Errorf("replicator not initialized")
|
|
||||||
}
|
|
||||||
|
|
||||||
err := n.replicator.Replicate(targetNodeID, targetAddress, data)
|
|
||||||
if err != nil {
|
|
||||||
failedCount.Add(1)
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Error(fmt.Sprintf("Failed to replicate document %s to node %s after retries: %v",
|
|
||||||
doc.ID, targetNodeID, err))
|
|
||||||
}
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
successCount.Add(1)
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Debug(fmt.Sprintf("Successfully replicated document %s to node %s", doc.ID, targetNodeID))
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
})
|
|
||||||
|
|
||||||
if err != nil {
|
|
||||||
wg.Done()
|
|
||||||
failedCount.Add(1)
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Error(fmt.Sprintf("Failed to submit replication task for %s to %s: %v",
|
|
||||||
doc.ID, targetNodeID, err))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Ждём завершения всех репликаций с таймаутом
|
|
||||||
done := make(chan struct{})
|
|
||||||
go func() {
|
|
||||||
wg.Wait()
|
|
||||||
close(done)
|
|
||||||
}()
|
|
||||||
|
|
||||||
select {
|
|
||||||
case <-done:
|
|
||||||
duration := time.Now().UnixMilli() - startTime
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Info(fmt.Sprintf("Replicated document %s to %d/%d nodes (took %d ms)",
|
|
||||||
doc.ID, successCount.Load(), len(nodes)-1, duration))
|
|
||||||
}
|
|
||||||
case <-time.After(30 * time.Second):
|
|
||||||
if n.logger != nil {
|
|
||||||
n.logger.Warn(fmt.Sprintf("Replication timeout for document %s after %d ms", doc.ID, time.Now().UnixMilli()-startTime))
|
|
||||||
}
|
|
||||||
return fmt.Errorf("replication timeout")
|
|
||||||
}
|
|
||||||
|
|
||||||
if failedCount.Load() > 0 {
|
|
||||||
return fmt.Errorf("replication partially failed: %d of %d nodes failed", failedCount.Load(), len(nodes)-1)
|
|
||||||
}
|
|
||||||
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetWorkerPoolStats возвращает статистику пула воркеров
|
|
||||||
func (n *Node) GetWorkerPoolStats() map[string]interface{} {
|
|
||||||
if n.workerPool == nil {
|
|
||||||
return map[string]interface{}{"enabled": false}
|
|
||||||
}
|
|
||||||
return n.workerPool.GetStats()
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetReplicationStats возвращает статистику репликации
|
|
||||||
func (n *Node) GetReplicationStats() map[string]interface{} {
|
|
||||||
if n.replicator == nil {
|
|
||||||
return map[string]interface{}{"enabled": false}
|
|
||||||
}
|
|
||||||
return n.replicator.GetStats()
|
|
||||||
}
|
|
||||||
Reference in New Issue
Block a user