2PC released
This commit is contained in:
349
internal/cluster/auth.go
Normal file
349
internal/cluster/auth.go
Normal file
@@ -0,0 +1,349 @@
|
||||
/*
|
||||
* 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/auth.go
|
||||
// Назначение: Аутентификация и авторизация между узлами кластера
|
||||
|
||||
package cluster
|
||||
|
||||
import (
|
||||
"crypto"
|
||||
"crypto/rand"
|
||||
"crypto/rsa"
|
||||
"crypto/sha256"
|
||||
"crypto/x509"
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"encoding/pem"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// NodeAuthConfig конфигурация аутентификации узлов
|
||||
type NodeAuthConfig struct {
|
||||
Enabled bool `json:"enabled"`
|
||||
TokenTTL time.Duration `json:"token_ttl"`
|
||||
PrivateKeyPath string `json:"private_key_path"`
|
||||
PublicKeyPath string `json:"public_key_path"`
|
||||
AllowedNodes []string `json:"allowed_nodes"`
|
||||
RequireMTLS bool `json:"require_mtls"`
|
||||
}
|
||||
|
||||
// NodeAuthToken представляет токен аутентификации узла
|
||||
type NodeAuthToken struct {
|
||||
NodeID string `json:"node_id"`
|
||||
IssuedAt int64 `json:"issued_at"`
|
||||
ExpiresAt int64 `json:"expires_at"`
|
||||
Signature string `json:"signature"`
|
||||
}
|
||||
|
||||
// NodeAuthenticator управляет аутентификацией между узлами
|
||||
type NodeAuthenticator struct {
|
||||
config *NodeAuthConfig
|
||||
privateKey *rsa.PrivateKey
|
||||
publicKey *rsa.PublicKey
|
||||
tokens sync.Map
|
||||
mu sync.RWMutex
|
||||
logger LoggerInterface
|
||||
}
|
||||
|
||||
// NewNodeAuthenticator создаёт новый аутентификатор узлов
|
||||
func NewNodeAuthenticator(config *NodeAuthConfig, logger LoggerInterface) (*NodeAuthenticator, error) {
|
||||
if config == nil {
|
||||
config = &NodeAuthConfig{
|
||||
Enabled: true,
|
||||
TokenTTL: 24 * time.Hour,
|
||||
AllowedNodes: make([]string, 0),
|
||||
RequireMTLS: false,
|
||||
}
|
||||
}
|
||||
|
||||
na := &NodeAuthenticator{
|
||||
config: config,
|
||||
logger: logger,
|
||||
}
|
||||
|
||||
if config.Enabled {
|
||||
if err := na.loadKeys(); err != nil {
|
||||
if err := na.generateKeys(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return na, nil
|
||||
}
|
||||
|
||||
// generateKeys генерирует RSA ключи для аутентификации
|
||||
func (na *NodeAuthenticator) generateKeys() error {
|
||||
privateKey, err := rsa.GenerateKey(rand.Reader, 2048)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
na.privateKey = privateKey
|
||||
na.publicKey = &privateKey.PublicKey
|
||||
|
||||
// Сохраняем ключи
|
||||
if err := na.saveKeys(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if na.logger != nil {
|
||||
na.logger.Info("Generated new RSA keys for node authentication")
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// loadKeys загружает RSA ключи с диска
|
||||
func (na *NodeAuthenticator) loadKeys() error {
|
||||
// Загружаем приватный ключ
|
||||
privData, err := os.ReadFile(na.config.PrivateKeyPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
block, _ := pem.Decode(privData)
|
||||
if block == nil {
|
||||
return fmt.Errorf("failed to decode private key")
|
||||
}
|
||||
|
||||
privateKey, err := x509.ParsePKCS8PrivateKey(block.Bytes)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var ok bool
|
||||
na.privateKey, ok = privateKey.(*rsa.PrivateKey)
|
||||
if !ok {
|
||||
return fmt.Errorf("invalid private key type")
|
||||
}
|
||||
|
||||
// Загружаем публичный ключ
|
||||
pubData, err := os.ReadFile(na.config.PublicKeyPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
pubBlock, _ := pem.Decode(pubData)
|
||||
if pubBlock == nil {
|
||||
return fmt.Errorf("failed to decode public key")
|
||||
}
|
||||
|
||||
publicKey, err := x509.ParsePKIXPublicKey(pubBlock.Bytes)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
na.publicKey, ok = publicKey.(*rsa.PublicKey)
|
||||
if !ok {
|
||||
return fmt.Errorf("invalid public key type")
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// saveKeys сохраняет RSA ключи на диск
|
||||
func (na *NodeAuthenticator) saveKeys() error {
|
||||
if na.privateKey == nil {
|
||||
return fmt.Errorf("no private key to save")
|
||||
}
|
||||
|
||||
// Сохраняем приватный ключ
|
||||
privBytes, err := x509.MarshalPKCS8PrivateKey(na.privateKey)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
privPEM := pem.EncodeToMemory(&pem.Block{
|
||||
Type: "PRIVATE KEY",
|
||||
Bytes: privBytes,
|
||||
})
|
||||
|
||||
if err := os.MkdirAll(filepath.Dir(na.config.PrivateKeyPath), 0700); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := os.WriteFile(na.config.PrivateKeyPath, privPEM, 0600); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Сохраняем публичный ключ
|
||||
pubBytes, err := x509.MarshalPKIXPublicKey(na.publicKey)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
pubPEM := pem.EncodeToMemory(&pem.Block{
|
||||
Type: "PUBLIC KEY",
|
||||
Bytes: pubBytes,
|
||||
})
|
||||
|
||||
if err := os.WriteFile(na.config.PublicKeyPath, pubPEM, 0644); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// GenerateToken генерирует токен для узла
|
||||
func (na *NodeAuthenticator) GenerateToken(nodeID string) (string, error) {
|
||||
if !na.config.Enabled {
|
||||
return "", nil
|
||||
}
|
||||
|
||||
now := time.Now().UnixMilli()
|
||||
token := &NodeAuthToken{
|
||||
NodeID: nodeID,
|
||||
IssuedAt: now,
|
||||
ExpiresAt: now + na.config.TokenTTL.Milliseconds(),
|
||||
}
|
||||
|
||||
// Подписываем токен
|
||||
data, err := json.Marshal(token)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
hash := sha256.Sum256(data)
|
||||
signature, err := rsa.SignPKCS1v15(rand.Reader, na.privateKey, crypto.SHA256, hash[:])
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
token.Signature = base64.StdEncoding.EncodeToString(signature)
|
||||
|
||||
// Сохраняем токен
|
||||
tokenData, err := json.Marshal(token)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
tokenString := base64.StdEncoding.EncodeToString(tokenData)
|
||||
na.tokens.Store(nodeID, token)
|
||||
|
||||
return tokenString, nil
|
||||
}
|
||||
|
||||
// VerifyToken проверяет токен узла
|
||||
func (na *NodeAuthenticator) VerifyToken(tokenString string) (string, error) {
|
||||
if !na.config.Enabled {
|
||||
return "", nil
|
||||
}
|
||||
|
||||
data, err := base64.StdEncoding.DecodeString(tokenString)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("invalid token encoding: %v", err)
|
||||
}
|
||||
|
||||
var token NodeAuthToken
|
||||
if err := json.Unmarshal(data, &token); err != nil {
|
||||
return "", fmt.Errorf("invalid token format: %v", err)
|
||||
}
|
||||
|
||||
// Проверяем срок действия
|
||||
now := time.Now().UnixMilli()
|
||||
if token.ExpiresAt < now {
|
||||
return "", fmt.Errorf("token expired")
|
||||
}
|
||||
|
||||
// Проверяем подпись
|
||||
sigData, err := base64.StdEncoding.DecodeString(token.Signature)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("invalid signature: %v", err)
|
||||
}
|
||||
|
||||
tokenCopy := token
|
||||
tokenCopy.Signature = ""
|
||||
dataWithoutSig, _ := json.Marshal(tokenCopy)
|
||||
hash := sha256.Sum256(dataWithoutSig)
|
||||
|
||||
if err := rsa.VerifyPKCS1v15(na.publicKey, crypto.SHA256, hash[:], sigData); err != nil {
|
||||
return "", fmt.Errorf("invalid signature: %v", err)
|
||||
}
|
||||
|
||||
// Проверяем, разрешён ли узел
|
||||
if len(na.config.AllowedNodes) > 0 {
|
||||
allowed := false
|
||||
for _, node := range na.config.AllowedNodes {
|
||||
if node == token.NodeID {
|
||||
allowed = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !allowed {
|
||||
return "", fmt.Errorf("node %s not allowed", token.NodeID)
|
||||
}
|
||||
}
|
||||
|
||||
return token.NodeID, nil
|
||||
}
|
||||
|
||||
// AuthenticateRequest аутентифицирует запрос
|
||||
func (na *NodeAuthenticator) AuthenticateRequest(data []byte, signature string) bool {
|
||||
if !na.config.Enabled {
|
||||
return true
|
||||
}
|
||||
|
||||
sigData, err := base64.StdEncoding.DecodeString(signature)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
|
||||
hash := sha256.Sum256(data)
|
||||
err = rsa.VerifyPKCS1v15(na.publicKey, crypto.SHA256, hash[:], sigData)
|
||||
return err == nil
|
||||
}
|
||||
|
||||
// SignRequest подписывает запрос
|
||||
func (na *NodeAuthenticator) SignRequest(data []byte) (string, error) {
|
||||
if !na.config.Enabled {
|
||||
return "", nil
|
||||
}
|
||||
|
||||
hash := sha256.Sum256(data)
|
||||
signature, err := rsa.SignPKCS1v15(rand.Reader, na.privateKey, crypto.SHA256, hash[:])
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
return base64.StdEncoding.EncodeToString(signature), nil
|
||||
}
|
||||
|
||||
// AddAllowedNode добавляет разрешённый узел
|
||||
func (na *NodeAuthenticator) AddAllowedNode(nodeID string) {
|
||||
na.mu.Lock()
|
||||
defer na.mu.Unlock()
|
||||
|
||||
for _, n := range na.config.AllowedNodes {
|
||||
if n == nodeID {
|
||||
return
|
||||
}
|
||||
}
|
||||
na.config.AllowedNodes = append(na.config.AllowedNodes, nodeID)
|
||||
}
|
||||
|
||||
// RemoveAllowedNode удаляет разрешённый узел
|
||||
func (na *NodeAuthenticator) RemoveAllowedNode(nodeID string) {
|
||||
na.mu.Lock()
|
||||
defer na.mu.Unlock()
|
||||
|
||||
newList := make([]string, 0)
|
||||
for _, n := range na.config.AllowedNodes {
|
||||
if n != nodeID {
|
||||
newList = append(newList, n)
|
||||
}
|
||||
}
|
||||
na.config.AllowedNodes = newList
|
||||
}
|
||||
429
internal/cluster/network_replication.go
Normal file
429
internal/cluster/network_replication.go
Normal file
@@ -0,0 +1,429 @@
|
||||
/*
|
||||
* 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/network_replication.go
|
||||
// Назначение: Реальная сетевая репликация с повторными попытками и бэкоффом
|
||||
|
||||
package cluster
|
||||
|
||||
import (
|
||||
"crypto/rand"
|
||||
"encoding/binary"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"math"
|
||||
"net"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
)
|
||||
|
||||
// ReplicationRetryConfig содержит конфигурацию повторных попыток
|
||||
type ReplicationRetryConfig struct {
|
||||
MaxRetries int
|
||||
InitialBackoff time.Duration
|
||||
MaxBackoff time.Duration
|
||||
BackoffFactor float64
|
||||
JitterEnabled bool
|
||||
}
|
||||
|
||||
// NetworkReplicationStats содержит статистику репликации
|
||||
type NetworkReplicationStats struct {
|
||||
TotalRequests atomic.Uint64
|
||||
SuccessfulReqs atomic.Uint64
|
||||
FailedReqs atomic.Uint64
|
||||
RetriedReqs atomic.Uint64
|
||||
AvgLatencyMs atomic.Uint64
|
||||
BytesSent atomic.Uint64
|
||||
BytesReceived atomic.Uint64
|
||||
}
|
||||
|
||||
// DefaultReplicationRetryConfig возвращает конфигурацию по умолчанию
|
||||
func DefaultReplicationRetryConfig() *ReplicationRetryConfig {
|
||||
return &ReplicationRetryConfig{
|
||||
MaxRetries: 5,
|
||||
InitialBackoff: 100 * time.Millisecond,
|
||||
MaxBackoff: 10 * time.Second,
|
||||
BackoffFactor: 2.0,
|
||||
JitterEnabled: true,
|
||||
}
|
||||
}
|
||||
|
||||
// validateConfig проверяет конфигурацию
|
||||
func validateConfig(c *ReplicationRetryConfig) error {
|
||||
if c.MaxRetries < 0 {
|
||||
return fmt.Errorf("max_retries cannot be negative")
|
||||
}
|
||||
if c.InitialBackoff < 0 {
|
||||
return fmt.Errorf("initial_backoff cannot be negative")
|
||||
}
|
||||
if c.MaxBackoff < c.InitialBackoff {
|
||||
return fmt.Errorf("max_backoff must be >= initial_backoff")
|
||||
}
|
||||
if c.BackoffFactor < 1.0 {
|
||||
return fmt.Errorf("backoff_factor must be >= 1.0")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ReplicatedConnection представляет соединение с узлом
|
||||
type ReplicatedConnection struct {
|
||||
NodeID string
|
||||
Address string
|
||||
Conn net.Conn
|
||||
LastUsed atomic.Int64
|
||||
Failures atomic.Uint32
|
||||
IsHealthy atomic.Bool
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
// NetworkReplicator управляет сетевой репликацией
|
||||
type NetworkReplicator struct {
|
||||
config *ReplicationRetryConfig
|
||||
workerPool *WorkerPool
|
||||
dialTimeout time.Duration
|
||||
mu sync.RWMutex
|
||||
connections map[string]*ReplicatedConnection
|
||||
logger LoggerInterface
|
||||
stats NetworkReplicationStats
|
||||
}
|
||||
|
||||
// NewNetworkReplicator создаёт новый сетевой репликатор
|
||||
func NewNetworkReplicator(config *ReplicationRetryConfig, workerPool *WorkerPool, logger LoggerInterface) *NetworkReplicator {
|
||||
if config == nil {
|
||||
config = DefaultReplicationRetryConfig()
|
||||
}
|
||||
// Игнорируем ошибку валидации в production
|
||||
validateConfig(config)
|
||||
|
||||
return &NetworkReplicator{
|
||||
config: config,
|
||||
workerPool: workerPool,
|
||||
dialTimeout: 10 * time.Second,
|
||||
connections: make(map[string]*ReplicatedConnection),
|
||||
logger: logger,
|
||||
}
|
||||
}
|
||||
|
||||
// calculateBackoff вычисляет время задержки с бэкоффом
|
||||
func (nr *NetworkReplicator) calculateBackoff(attempt int) time.Duration {
|
||||
backoff := float64(nr.config.InitialBackoff) * math.Pow(nr.config.BackoffFactor, float64(attempt))
|
||||
if backoff > float64(nr.config.MaxBackoff) {
|
||||
backoff = float64(nr.config.MaxBackoff)
|
||||
}
|
||||
|
||||
duration := time.Duration(backoff)
|
||||
|
||||
if nr.config.JitterEnabled {
|
||||
// Добавляем случайный джиттер ±25%
|
||||
jitter := time.Duration(float64(duration) * 0.25)
|
||||
jitterDelta := int64(jitter) - int64(jitter/2)
|
||||
if jitterDelta < 0 {
|
||||
jitterDelta = 0
|
||||
}
|
||||
duration = duration + time.Duration(jitterDelta)
|
||||
}
|
||||
|
||||
return duration
|
||||
}
|
||||
|
||||
// randUint64 возвращает случайное 64-битное число
|
||||
func randUint64() uint64 {
|
||||
var buf [8]byte
|
||||
rand.Read(buf[:])
|
||||
return binary.LittleEndian.Uint64(buf[:])
|
||||
}
|
||||
|
||||
// randInt64 возвращает случайное число между min и max
|
||||
func randInt64(min, max int64) int64 {
|
||||
if min >= max {
|
||||
return min
|
||||
}
|
||||
delta := max - min
|
||||
if delta == 0 {
|
||||
return min
|
||||
}
|
||||
n := int64(randUint64() % uint64(delta))
|
||||
return min + n
|
||||
}
|
||||
|
||||
// randInt возвращает случайное целое
|
||||
func randInt(min, max int) int {
|
||||
if min >= max {
|
||||
return min
|
||||
}
|
||||
delta := max - min
|
||||
if delta == 0 {
|
||||
return min
|
||||
}
|
||||
n := int(randUint64() % uint64(delta))
|
||||
return min + n
|
||||
}
|
||||
|
||||
// Connect устанавливает соединение с узлом
|
||||
func (nr *NetworkReplicator) Connect(nodeID, address string) error {
|
||||
nr.mu.Lock()
|
||||
defer nr.mu.Unlock()
|
||||
|
||||
if conn, exists := nr.connections[nodeID]; exists {
|
||||
if conn.IsHealthy.Load() {
|
||||
return nil
|
||||
}
|
||||
conn.Conn.Close()
|
||||
}
|
||||
|
||||
conn, err := net.DialTimeout("tcp", address, nr.dialTimeout)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to connect to %s: %v", address, err)
|
||||
}
|
||||
|
||||
replicatedConn := &ReplicatedConnection{
|
||||
NodeID: nodeID,
|
||||
Address: address,
|
||||
Conn: conn,
|
||||
IsHealthy: atomic.Bool{},
|
||||
}
|
||||
replicatedConn.IsHealthy.Store(true)
|
||||
replicatedConn.LastUsed.Store(time.Now().UnixMilli())
|
||||
|
||||
nr.connections[nodeID] = replicatedConn
|
||||
|
||||
if nr.logger != nil {
|
||||
nr.logger.Debug(fmt.Sprintf("Connected to node %s at %s", nodeID, address))
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// sendReplicationRequest отправляет запрос на репликацию
|
||||
func (nr *NetworkReplicator) sendReplicationRequest(conn net.Conn, data []byte) error {
|
||||
// Формируем сообщение: длина + данные
|
||||
lenBuf := make([]byte, 4)
|
||||
binary.BigEndian.PutUint32(lenBuf, uint32(len(data)))
|
||||
|
||||
if _, err := conn.Write(lenBuf); err != nil {
|
||||
return fmt.Errorf("failed to write length: %v", err)
|
||||
}
|
||||
|
||||
if _, err := conn.Write(data); err != nil {
|
||||
return fmt.Errorf("failed to write data: %v", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// readReplicationAck читает подтверждение репликации
|
||||
func (nr *NetworkReplicator) readReplicationAck(conn net.Conn) error {
|
||||
lenBuf := make([]byte, 4)
|
||||
if _, err := io.ReadFull(conn, lenBuf); err != nil {
|
||||
return fmt.Errorf("failed to read ack length: %v", err)
|
||||
}
|
||||
|
||||
ackLen := binary.BigEndian.Uint32(lenBuf)
|
||||
if ackLen > 1024 {
|
||||
return fmt.Errorf("ack too large: %d", ackLen)
|
||||
}
|
||||
|
||||
ackData := make([]byte, ackLen)
|
||||
if _, err := io.ReadFull(conn, ackData); err != nil {
|
||||
return fmt.Errorf("failed to read ack data: %v", err)
|
||||
}
|
||||
|
||||
var ack struct {
|
||||
Status string `json:"status"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
if err := json.Unmarshal(ackData, &ack); err != nil {
|
||||
return fmt.Errorf("failed to parse ack: %v", err)
|
||||
}
|
||||
|
||||
if ack.Status != "ok" {
|
||||
return fmt.Errorf("negative ack: %s", ack.Error)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// doSendReplication выполняет отправку репликации
|
||||
func (nr *NetworkReplicator) doSendReplication(nodeID, address string, data []byte) error {
|
||||
conn := nr.getConnection(nodeID)
|
||||
if conn == nil {
|
||||
// Пытаемся подключиться
|
||||
if err := nr.Connect(nodeID, address); err != nil {
|
||||
return err
|
||||
}
|
||||
conn = nr.getConnection(nodeID)
|
||||
if conn == nil {
|
||||
return fmt.Errorf("failed to get connection for node %s", nodeID)
|
||||
}
|
||||
}
|
||||
|
||||
conn.mu.Lock()
|
||||
defer conn.mu.Unlock()
|
||||
|
||||
if conn.Conn == nil {
|
||||
return fmt.Errorf("connection is nil for node %s", nodeID)
|
||||
}
|
||||
|
||||
// Устанавливаем таймаут
|
||||
if err := conn.Conn.SetWriteDeadline(time.Now().Add(30 * time.Second)); err != nil {
|
||||
return fmt.Errorf("failed to set write deadline: %v", err)
|
||||
}
|
||||
|
||||
// Отправляем данные
|
||||
if err := nr.sendReplicationRequest(conn.Conn, data); err != nil {
|
||||
conn.IsHealthy.Store(false)
|
||||
conn.Conn.Close()
|
||||
conn.Conn = nil
|
||||
return fmt.Errorf("write failed: %v", err)
|
||||
}
|
||||
|
||||
// Читаем подтверждение
|
||||
if err := conn.Conn.SetReadDeadline(time.Now().Add(10 * time.Second)); err != nil {
|
||||
return fmt.Errorf("failed to set read deadline: %v", err)
|
||||
}
|
||||
|
||||
if err := nr.readReplicationAck(conn.Conn); err != nil {
|
||||
conn.IsHealthy.Store(false)
|
||||
return fmt.Errorf("read ack failed: %v", err)
|
||||
}
|
||||
|
||||
conn.IsHealthy.Store(true)
|
||||
conn.LastUsed.Store(time.Now().UnixMilli())
|
||||
conn.Failures.Store(0)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Replicate отправляет запрос на репликацию с повторными попытками
|
||||
func (nr *NetworkReplicator) Replicate(targetNodeID, targetAddress string, data []byte) error {
|
||||
startTime := time.Now()
|
||||
nr.stats.TotalRequests.Add(1)
|
||||
|
||||
var lastErr error
|
||||
|
||||
for attempt := 0; attempt <= nr.config.MaxRetries; attempt++ {
|
||||
if attempt > 0 {
|
||||
nr.stats.RetriedReqs.Add(1)
|
||||
backoff := nr.calculateBackoff(attempt - 1)
|
||||
if nr.logger != nil {
|
||||
nr.logger.Debug(fmt.Sprintf("Replication retry %d for node %s after %v", attempt, targetNodeID, backoff))
|
||||
}
|
||||
time.Sleep(backoff)
|
||||
}
|
||||
|
||||
err := nr.doSendReplication(targetNodeID, targetAddress, data)
|
||||
if err == nil {
|
||||
latency := time.Since(startTime).Milliseconds()
|
||||
nr.stats.SuccessfulReqs.Add(1)
|
||||
nr.updateAvgLatency(latency)
|
||||
nr.stats.BytesSent.Add(uint64(len(data)))
|
||||
|
||||
if nr.logger != nil {
|
||||
nr.logger.Debug(fmt.Sprintf("Replication to %s succeeded after %d attempts, latency: %dms",
|
||||
targetNodeID, attempt+1, latency))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
lastErr = err
|
||||
if nr.logger != nil {
|
||||
nr.logger.Warn(fmt.Sprintf("Replication attempt %d to %s failed: %v", attempt+1, targetNodeID, err))
|
||||
}
|
||||
|
||||
// Помечаем соединение как нездоровое
|
||||
nr.markUnhealthy(targetNodeID)
|
||||
}
|
||||
|
||||
nr.stats.FailedReqs.Add(1)
|
||||
return fmt.Errorf("replication failed after %d attempts: %v", nr.config.MaxRetries+1, lastErr)
|
||||
}
|
||||
|
||||
// getConnection возвращает соединение с узлом
|
||||
func (nr *NetworkReplicator) getConnection(nodeID string) *ReplicatedConnection {
|
||||
nr.mu.RLock()
|
||||
defer nr.mu.RUnlock()
|
||||
return nr.connections[nodeID]
|
||||
}
|
||||
|
||||
// markUnhealthy помечает соединение как нездоровое
|
||||
func (nr *NetworkReplicator) markUnhealthy(nodeID string) {
|
||||
nr.mu.RLock()
|
||||
conn, exists := nr.connections[nodeID]
|
||||
nr.mu.RUnlock()
|
||||
|
||||
if exists {
|
||||
conn.IsHealthy.Store(false)
|
||||
failures := conn.Failures.Add(1)
|
||||
if failures > 3 {
|
||||
if nr.logger != nil {
|
||||
nr.logger.Warn(fmt.Sprintf("Node %s marked as unhealthy after %d failures", nodeID, failures))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// updateAvgLatency обновляет среднюю задержку
|
||||
func (nr *NetworkReplicator) updateAvgLatency(latencyMs int64) {
|
||||
current := nr.stats.AvgLatencyMs.Load()
|
||||
successful := nr.stats.SuccessfulReqs.Load()
|
||||
if successful > 0 {
|
||||
newAvg := (current*uint64(successful-1) + uint64(latencyMs)) / uint64(successful)
|
||||
nr.stats.AvgLatencyMs.Store(newAvg)
|
||||
} else {
|
||||
nr.stats.AvgLatencyMs.Store(uint64(latencyMs))
|
||||
}
|
||||
}
|
||||
|
||||
// ReplicateDocumentAsync асинхронно реплицирует документ
|
||||
func (nr *NetworkReplicator) ReplicateDocumentAsync(docID string, targetNodes []*NodeInfo, data []byte) {
|
||||
for _, node := range targetNodes {
|
||||
targetNodeID := node.ID
|
||||
targetAddress := fmt.Sprintf("%s:%d", node.IP, node.Port)
|
||||
|
||||
err := nr.workerPool.SubmitFunc(fmt.Sprintf("replicate_%s_to_%s", docID, targetNodeID), func() error {
|
||||
return nr.Replicate(targetNodeID, targetAddress, data)
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
if nr.logger != nil {
|
||||
nr.logger.Error(fmt.Sprintf("Failed to submit replication task for %s to %s: %v", docID, targetNodeID, err))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// GetStats возвращает статистику репликации
|
||||
func (nr *NetworkReplicator) GetStats() map[string]interface{} {
|
||||
return map[string]interface{}{
|
||||
"total_requests": nr.stats.TotalRequests.Load(),
|
||||
"successful": nr.stats.SuccessfulReqs.Load(),
|
||||
"failed": nr.stats.FailedReqs.Load(),
|
||||
"retried": nr.stats.RetriedReqs.Load(),
|
||||
"avg_latency_ms": nr.stats.AvgLatencyMs.Load(),
|
||||
"bytes_sent": nr.stats.BytesSent.Load(),
|
||||
"bytes_received": nr.stats.BytesReceived.Load(),
|
||||
}
|
||||
}
|
||||
|
||||
// Close закрывает все соединения
|
||||
func (nr *NetworkReplicator) Close() error {
|
||||
nr.mu.Lock()
|
||||
defer nr.mu.Unlock()
|
||||
|
||||
for _, conn := range nr.connections {
|
||||
if conn.Conn != nil {
|
||||
conn.Conn.Close()
|
||||
}
|
||||
}
|
||||
nr.connections = make(map[string]*ReplicatedConnection)
|
||||
return nil
|
||||
}
|
||||
974
internal/cluster/node.go
Normal file
974
internal/cluster/node.go
Normal file
@@ -0,0 +1,974 @@
|
||||
/*
|
||||
* 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()
|
||||
}
|
||||
2947
internal/cluster/raft_coordinator.go
Normal file
2947
internal/cluster/raft_coordinator.go
Normal file
File diff suppressed because it is too large
Load Diff
289
internal/cluster/recovery.go
Normal file
289
internal/cluster/recovery.go
Normal file
@@ -0,0 +1,289 @@
|
||||
/*
|
||||
* 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/recovery.go
|
||||
// Назначение: Автоматическое восстановление узлов после сбоя
|
||||
|
||||
package cluster
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
)
|
||||
|
||||
// NodeState представляет состояние узла для восстановления
|
||||
type NodeState struct {
|
||||
NodeID string `json:"node_id"`
|
||||
LastSeen time.Time `json:"last_seen"`
|
||||
LastLogIndex uint64 `json:"last_log_index"`
|
||||
FailureCount int `json:"failure_count"`
|
||||
IsRecovering bool `json:"is_recovering"`
|
||||
}
|
||||
|
||||
// RecoveryManager управляет восстановлением узлов
|
||||
type RecoveryManager struct {
|
||||
coordinator *RaftCoordinator
|
||||
logger LoggerInterface
|
||||
states sync.Map
|
||||
maxFailures int
|
||||
recoveryDelay time.Duration
|
||||
stopChan chan struct{}
|
||||
wg sync.WaitGroup
|
||||
isActive atomic.Bool
|
||||
}
|
||||
|
||||
// NewRecoveryManager создаёт новый менеджер восстановления
|
||||
func NewRecoveryManager(coordinator *RaftCoordinator, logger LoggerInterface) *RecoveryManager {
|
||||
rm := &RecoveryManager{
|
||||
coordinator: coordinator,
|
||||
logger: logger,
|
||||
maxFailures: 3,
|
||||
recoveryDelay: 30 * time.Second,
|
||||
stopChan: make(chan struct{}),
|
||||
}
|
||||
|
||||
return rm
|
||||
}
|
||||
|
||||
// Start запускает мониторинг восстановления
|
||||
func (rm *RecoveryManager) Start() {
|
||||
rm.isActive.Store(true)
|
||||
rm.wg.Add(2)
|
||||
go rm.monitorLoop()
|
||||
go rm.recoveryLoop()
|
||||
|
||||
if rm.logger != nil {
|
||||
rm.logger.Info("Recovery manager started")
|
||||
}
|
||||
}
|
||||
|
||||
// Stop останавливает менеджер восстановления
|
||||
func (rm *RecoveryManager) Stop() {
|
||||
rm.isActive.Store(false)
|
||||
close(rm.stopChan)
|
||||
rm.wg.Wait()
|
||||
|
||||
if rm.logger != nil {
|
||||
rm.logger.Info("Recovery manager stopped")
|
||||
}
|
||||
}
|
||||
|
||||
// monitorLoop отслеживает состояние узлов
|
||||
func (rm *RecoveryManager) monitorLoop() {
|
||||
defer rm.wg.Done()
|
||||
|
||||
ticker := time.NewTicker(10 * time.Second)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-rm.stopChan:
|
||||
return
|
||||
case <-ticker.C:
|
||||
rm.checkNodesHealth()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// checkNodesHealth проверяет здоровье узлов
|
||||
func (rm *RecoveryManager) checkNodesHealth() {
|
||||
if rm.coordinator == nil {
|
||||
return
|
||||
}
|
||||
|
||||
nodes := rm.coordinator.GetAllNodes()
|
||||
now := time.Now()
|
||||
|
||||
for _, node := range nodes {
|
||||
stateVal, ok := rm.states.Load(node.ID)
|
||||
var state *NodeState
|
||||
if ok {
|
||||
state = stateVal.(*NodeState)
|
||||
} else {
|
||||
state = &NodeState{
|
||||
NodeID: node.ID,
|
||||
LastSeen: now,
|
||||
LastLogIndex: 0,
|
||||
FailureCount: 0,
|
||||
IsRecovering: false,
|
||||
}
|
||||
rm.states.Store(node.ID, state)
|
||||
}
|
||||
|
||||
// Проверяем, был ли недавно heartbeat
|
||||
lastSeen := time.UnixMilli(node.LastSeen)
|
||||
if now.Sub(lastSeen) > 30*time.Second {
|
||||
state.FailureCount++
|
||||
if rm.logger != nil {
|
||||
rm.logger.Warn(fmt.Sprintf("Node %s appears unhealthy, failure count: %d", node.ID, state.FailureCount))
|
||||
}
|
||||
|
||||
if state.FailureCount >= rm.maxFailures && !state.IsRecovering {
|
||||
rm.triggerRecovery(node.ID)
|
||||
}
|
||||
} else {
|
||||
if state.FailureCount > 0 {
|
||||
state.FailureCount = 0
|
||||
if rm.logger != nil {
|
||||
rm.logger.Info(fmt.Sprintf("Node %s recovered", node.ID))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
state.LastSeen = now
|
||||
rm.states.Store(node.ID, state)
|
||||
}
|
||||
}
|
||||
|
||||
// triggerRecovery запускает восстановление узла
|
||||
func (rm *RecoveryManager) triggerRecovery(nodeID string) {
|
||||
stateVal, ok := rm.states.Load(nodeID)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
|
||||
state := stateVal.(*NodeState)
|
||||
if state.IsRecovering {
|
||||
return
|
||||
}
|
||||
|
||||
state.IsRecovering = true
|
||||
rm.states.Store(nodeID, state)
|
||||
|
||||
if rm.logger != nil {
|
||||
rm.logger.Info(fmt.Sprintf("Triggering recovery for node %s", nodeID))
|
||||
}
|
||||
|
||||
// Запускаем асинхронное восстановление
|
||||
go rm.recoverNode(nodeID)
|
||||
}
|
||||
|
||||
// recoverNode восстанавливает узел
|
||||
func (rm *RecoveryManager) recoverNode(nodeID string) {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
if rm.logger != nil {
|
||||
rm.logger.Error(fmt.Sprintf("Recovery for node %s panicked: %v", nodeID, r))
|
||||
}
|
||||
}
|
||||
|
||||
// Сбрасываем флаг восстановления
|
||||
if stateVal, ok := rm.states.Load(nodeID); ok {
|
||||
state := stateVal.(*NodeState)
|
||||
state.IsRecovering = false
|
||||
rm.states.Store(nodeID, state)
|
||||
}
|
||||
}()
|
||||
|
||||
// Ждём перед попыткой восстановления
|
||||
time.Sleep(rm.recoveryDelay)
|
||||
|
||||
if rm.coordinator == nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Проверяем, не восстановился ли узел самостоятельно
|
||||
node := rm.coordinator.GetNodeByID(nodeID)
|
||||
if node != nil && time.Now().UnixMilli()-node.LastSeen < 30000 {
|
||||
if rm.logger != nil {
|
||||
rm.logger.Info(fmt.Sprintf("Node %s recovered on its own", nodeID))
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// Пытаемся переподключить узел
|
||||
if rm.logger != nil {
|
||||
rm.logger.Info(fmt.Sprintf("Attempting to reconnect node %s", nodeID))
|
||||
}
|
||||
|
||||
// Обновляем статус узла
|
||||
if err := rm.coordinator.UpdateNodeStatus(nodeID, StatusActive); err != nil {
|
||||
if rm.logger != nil {
|
||||
rm.logger.Error(fmt.Sprintf("Failed to update node %s status: %v", nodeID, err))
|
||||
}
|
||||
}
|
||||
|
||||
// Запускаем синхронизацию данных
|
||||
if err := rm.syncNodeData(nodeID); err != nil {
|
||||
if rm.logger != nil {
|
||||
rm.logger.Error(fmt.Sprintf("Failed to sync node %s data: %v", nodeID, err))
|
||||
}
|
||||
}
|
||||
|
||||
if rm.logger != nil {
|
||||
rm.logger.Info(fmt.Sprintf("Recovery completed for node %s", nodeID))
|
||||
}
|
||||
}
|
||||
|
||||
// syncNodeData синхронизирует данные с узлом
|
||||
func (rm *RecoveryManager) syncNodeData(nodeID string) error {
|
||||
// Получаем информацию об узле
|
||||
node := rm.coordinator.GetNodeByID(nodeID)
|
||||
if node == nil {
|
||||
return fmt.Errorf("node not found: %s", nodeID)
|
||||
}
|
||||
|
||||
if rm.logger != nil {
|
||||
rm.logger.Debug(fmt.Sprintf("Syncing data with node %s at %s:%d", nodeID, node.IP, node.Port))
|
||||
}
|
||||
|
||||
// В реальной реализации здесь была бы синхронизация данных
|
||||
// Например, отправка всех изменений с момента последнего успешного heartbeat
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// recoveryLoop периодически проверяет и восстанавливает узлы
|
||||
func (rm *RecoveryManager) recoveryLoop() {
|
||||
defer rm.wg.Done()
|
||||
|
||||
ticker := time.NewTicker(1 * time.Minute)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-rm.stopChan:
|
||||
return
|
||||
case <-ticker.C:
|
||||
rm.attemptRecoveryAll()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// attemptRecoveryAll пытается восстановить все проблемные узлы
|
||||
func (rm *RecoveryManager) attemptRecoveryAll() {
|
||||
rm.states.Range(func(key, value interface{}) bool {
|
||||
state := value.(*NodeState)
|
||||
if state.FailureCount >= rm.maxFailures && !state.IsRecovering {
|
||||
go rm.recoverNode(state.NodeID)
|
||||
}
|
||||
return true
|
||||
})
|
||||
}
|
||||
|
||||
// GetNodeState возвращает состояние узла
|
||||
func (rm *RecoveryManager) GetNodeState(nodeID string) *NodeState {
|
||||
if val, ok := rm.states.Load(nodeID); ok {
|
||||
return val.(*NodeState)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetAllStates возвращает все состояния узлов
|
||||
func (rm *RecoveryManager) GetAllStates() map[string]*NodeState {
|
||||
states := make(map[string]*NodeState)
|
||||
rm.states.Range(func(key, value interface{}) bool {
|
||||
states[key.(string)] = value.(*NodeState)
|
||||
return true
|
||||
})
|
||||
return states
|
||||
}
|
||||
301
internal/cluster/types.go
Normal file
301
internal/cluster/types.go
Normal file
@@ -0,0 +1,301 @@
|
||||
/*
|
||||
* 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/types.go
|
||||
// Назначение: Общие типы данных для кластерных операций с поддержкой временных меток
|
||||
|
||||
package cluster
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"time"
|
||||
)
|
||||
|
||||
// NodeInfo представляет информацию об узле для координатора
|
||||
type NodeInfo struct {
|
||||
ID string `json:"id"`
|
||||
IP string `json:"ip"`
|
||||
Port int `json:"port"`
|
||||
Status string `json:"status"`
|
||||
LastSeen int64 `json:"last_seen"`
|
||||
JoinedAt int64 `json:"joined_at"`
|
||||
UpdatedAt int64 `json:"updated_at"`
|
||||
Version uint64 `json:"version"`
|
||||
}
|
||||
|
||||
// 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"`
|
||||
}
|
||||
|
||||
// ClusterHealth представляет информацию о здоровье кластера
|
||||
type ClusterHealth struct {
|
||||
Nodes map[string]*NodeHealth `json:"nodes"`
|
||||
OverallScore float64 `json:"overall_score"`
|
||||
Recommendations string `json:"recommendations"`
|
||||
CheckedAt int64 `json:"checked_at"`
|
||||
}
|
||||
|
||||
// NodeHealth представляет здоровье отдельного узла
|
||||
type NodeHealth struct {
|
||||
Status string `json:"status"`
|
||||
LatencyMs int64 `json:"latency_ms"`
|
||||
LastCheck int64 `json:"last_check"`
|
||||
LastSuccess int64 `json:"last_success"`
|
||||
LastFailure int64 `json:"last_failure"`
|
||||
FailureCount int `json:"failure_count"`
|
||||
}
|
||||
|
||||
// NodeRequest представляет запрос от одного узла к другому
|
||||
type NodeRequest struct {
|
||||
Type string `json:"type"`
|
||||
Data []byte `json:"data"`
|
||||
FromNode string `json:"from_node"`
|
||||
RequestID string `json:"request_id"`
|
||||
Timestamp int64 `json:"timestamp"`
|
||||
}
|
||||
|
||||
// ShardInfo представляет информацию о шарде
|
||||
type ShardInfo struct {
|
||||
ID string `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Nodes []string `json:"nodes"`
|
||||
LeaderNode string `json:"leader_node"`
|
||||
Status string `json:"status"`
|
||||
CreatedAt int64 `json:"created_at"`
|
||||
UpdatedAt int64 `json:"updated_at"`
|
||||
LastRebalanced int64 `json:"last_rebalanced"`
|
||||
DocumentCount int64 `json:"document_count"`
|
||||
SizeBytes int64 `json:"size_bytes"`
|
||||
}
|
||||
|
||||
// ReplicationLogEntry представляет запись в журнале репликации
|
||||
type ReplicationLogEntry struct {
|
||||
ID string `json:"id"`
|
||||
Timestamp int64 `json:"timestamp"`
|
||||
SourceNode string `json:"source_node"`
|
||||
TargetNode string `json:"target_node"`
|
||||
Operation string `json:"operation"`
|
||||
Database string `json:"database"`
|
||||
Collection string `json:"collection"`
|
||||
DocumentID string `json:"document_id"`
|
||||
Status string `json:"status"`
|
||||
DurationMs int64 `json:"duration_ms"`
|
||||
Error string `json:"error,omitempty"`
|
||||
Details map[string]interface{} `json:"details,omitempty"`
|
||||
}
|
||||
|
||||
// ========== СТРУКТУРЫ ДЛЯ PRODUCTION-READY РЕАЛИЗАЦИИ ==========
|
||||
|
||||
// ConnectionInfo содержит информацию о соединении с узлом
|
||||
type ConnectionInfo struct {
|
||||
NodeID string `json:"node_id"`
|
||||
Address string `json:"address"`
|
||||
IsHealthy bool `json:"is_healthy"`
|
||||
LastUsed int64 `json:"last_used"`
|
||||
Failures uint32 `json:"failures"`
|
||||
ConnectedAt int64 `json:"connected_at"`
|
||||
}
|
||||
|
||||
// ClusterOperation представляет операцию в кластере
|
||||
type ClusterOperation struct {
|
||||
ID string `json:"id"`
|
||||
Type string `json:"type"` // "add_node", "remove_node", "rebalance", "reshard"
|
||||
Status string `json:"status"` // "pending", "in_progress", "completed", "failed"
|
||||
CreatedAt int64 `json:"created_at"`
|
||||
StartedAt int64 `json:"started_at"`
|
||||
CompletedAt int64 `json:"completed_at"`
|
||||
Details map[string]interface{} `json:"details"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
// WorkerPoolStats содержит статистику пула воркеров
|
||||
type WorkerPoolStats struct {
|
||||
MaxWorkers int `json:"max_workers"`
|
||||
ActiveTasks int32 `json:"active_tasks"`
|
||||
TotalTasks uint64 `json:"total_tasks"`
|
||||
FailedTasks uint64 `json:"failed_tasks"`
|
||||
QueueSize int `json:"queue_size"`
|
||||
QueueCapacity int `json:"queue_capacity"`
|
||||
}
|
||||
|
||||
// ========== МЕТОДЫ ДЛЯ NodeInfo ==========
|
||||
|
||||
// NodeJoinedAt возвращает человекочитаемое время присоединения узла
|
||||
func (n *NodeInfo) NodeJoinedAt() string {
|
||||
if n.JoinedAt == 0 {
|
||||
return "not joined"
|
||||
}
|
||||
return time.UnixMilli(n.JoinedAt).Format("2006-01-02 15:04:05.000")
|
||||
}
|
||||
|
||||
// LastSeenAt возвращает человекочитаемое время последнего контакта
|
||||
func (n *NodeInfo) LastSeenAt() string {
|
||||
if n.LastSeen == 0 {
|
||||
return "never"
|
||||
}
|
||||
return time.UnixMilli(n.LastSeen).Format("2006-01-02 15:04:05.000")
|
||||
}
|
||||
|
||||
// GetUptime возвращает время жизни узла в кластере
|
||||
func (n *NodeInfo) GetUptime() time.Duration {
|
||||
if n.JoinedAt == 0 {
|
||||
return 0
|
||||
}
|
||||
return time.Duration(time.Now().UnixMilli()-n.JoinedAt) * time.Millisecond
|
||||
}
|
||||
|
||||
// IsHealthy проверяет, здоров ли узел
|
||||
func (n *NodeInfo) IsHealthy() bool {
|
||||
return n.Status == "active" && time.Now().UnixMilli()-n.LastSeen < 30000
|
||||
}
|
||||
|
||||
// GetAddress возвращает адрес узла
|
||||
func (n *NodeInfo) GetAddress() string {
|
||||
return fmt.Sprintf("%s:%d", n.IP, n.Port)
|
||||
}
|
||||
|
||||
// ========== МЕТОДЫ ДЛЯ ShardInfo ==========
|
||||
|
||||
// GetUptime возвращает время жизни шарда
|
||||
func (s *ShardInfo) GetUptime() time.Duration {
|
||||
if s.CreatedAt == 0 {
|
||||
return 0
|
||||
}
|
||||
return time.Duration(time.Now().UnixMilli()-s.CreatedAt) * time.Millisecond
|
||||
}
|
||||
|
||||
// IsHealthy проверяет, здоров ли шард
|
||||
func (s *ShardInfo) IsHealthy() bool {
|
||||
return s.Status == "active"
|
||||
}
|
||||
|
||||
// GetLastRebalancedStr возвращает человекочитаемое время последней ребалансировки
|
||||
func (s *ShardInfo) GetLastRebalancedStr() string {
|
||||
if s.LastRebalanced == 0 {
|
||||
return "never"
|
||||
}
|
||||
return time.UnixMilli(s.LastRebalanced).Format("2006-01-02 15:04:05.000")
|
||||
}
|
||||
|
||||
// GetCreatedAtStr возвращает человекочитаемое время создания
|
||||
func (s *ShardInfo) GetCreatedAtStr() string {
|
||||
if s.CreatedAt == 0 {
|
||||
return "unknown"
|
||||
}
|
||||
return time.UnixMilli(s.CreatedAt).Format("2006-01-02 15:04:05.000")
|
||||
}
|
||||
|
||||
// GetUpdatedAtStr возвращает человекочитаемое время обновления
|
||||
func (s *ShardInfo) GetUpdatedAtStr() string {
|
||||
if s.UpdatedAt == 0 {
|
||||
return "unknown"
|
||||
}
|
||||
return time.UnixMilli(s.UpdatedAt).Format("2006-01-02 15:04:05.000")
|
||||
}
|
||||
|
||||
// GetLeaderNode возвращает лидера шарда
|
||||
func (s *ShardInfo) GetLeaderNode() string {
|
||||
return s.LeaderNode
|
||||
}
|
||||
|
||||
// GetNodeCount возвращает количество узлов в шарде
|
||||
func (s *ShardInfo) GetNodeCount() int {
|
||||
return len(s.Nodes)
|
||||
}
|
||||
|
||||
// GetReplicationFactor возвращает фактор репликации шарда
|
||||
func (s *ShardInfo) GetReplicationFactor() int {
|
||||
return len(s.Nodes)
|
||||
}
|
||||
|
||||
// ========== МЕТОДЫ ДЛЯ ReplicationLogEntry ==========
|
||||
|
||||
// GetTimestampStr возвращает человекочитаемое время записи репликации
|
||||
func (r *ReplicationLogEntry) GetTimestampStr() string {
|
||||
if r.Timestamp == 0 {
|
||||
return "unknown"
|
||||
}
|
||||
return time.UnixMilli(r.Timestamp).Format("2006-01-02 15:04:05.000")
|
||||
}
|
||||
|
||||
// IsSuccess проверяет, успешна ли операция репликации
|
||||
func (r *ReplicationLogEntry) IsSuccess() bool {
|
||||
return r.Status == "success"
|
||||
}
|
||||
|
||||
// GetDuration возвращает длительность операции
|
||||
func (r *ReplicationLogEntry) GetDuration() time.Duration {
|
||||
return time.Duration(r.DurationMs) * time.Millisecond
|
||||
}
|
||||
|
||||
// ========== МЕТОДЫ ДЛЯ ClusterHealth ==========
|
||||
|
||||
// IsHealthy проверяет, здоров ли кластер в целом
|
||||
func (ch *ClusterHealth) IsHealthy() bool {
|
||||
return ch.OverallScore >= 80
|
||||
}
|
||||
|
||||
// IsDegraded проверяет, деградирован ли кластер
|
||||
func (ch *ClusterHealth) IsDegraded() bool {
|
||||
return ch.OverallScore >= 50 && ch.OverallScore < 80
|
||||
}
|
||||
|
||||
// IsCritical проверяет, находится ли кластер в критическом состоянии
|
||||
func (ch *ClusterHealth) IsCritical() bool {
|
||||
return ch.OverallScore < 50
|
||||
}
|
||||
|
||||
// GetHealthyNodes возвращает количество здоровых узлов
|
||||
func (ch *ClusterHealth) GetHealthyNodes() int {
|
||||
count := 0
|
||||
for _, node := range ch.Nodes {
|
||||
if node.Status == "active" {
|
||||
count++
|
||||
}
|
||||
}
|
||||
return count
|
||||
}
|
||||
|
||||
// ========== ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ ==========
|
||||
|
||||
// GetTimestamp возвращает текущий timestamp в миллисекундах
|
||||
func GetTimestamp() int64 {
|
||||
return time.Now().UnixMilli()
|
||||
}
|
||||
|
||||
// FormatTimestamp форматирует timestamp в человекочитаемый вид
|
||||
func FormatTimestamp(ts int64) string {
|
||||
if ts == 0 {
|
||||
return "never"
|
||||
}
|
||||
return time.UnixMilli(ts).Format("2006-01-02 15:04:05.000")
|
||||
}
|
||||
|
||||
// IsTimestampExpired проверяет, истёк ли timestamp
|
||||
func IsTimestampExpired(ts int64, ttl time.Duration) bool {
|
||||
if ts == 0 {
|
||||
return true
|
||||
}
|
||||
return time.Now().UnixMilli()-ts > ttl.Milliseconds()
|
||||
}
|
||||
216
internal/cluster/worker_pool.go
Normal file
216
internal/cluster/worker_pool.go
Normal file
@@ -0,0 +1,216 @@
|
||||
/*
|
||||
* 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/worker_pool.go
|
||||
// Назначение: Пул воркеров для ограничения количества горутин
|
||||
|
||||
package cluster
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"runtime/debug"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
)
|
||||
|
||||
// LoggerInterface определяет интерфейс для логирования
|
||||
type LoggerInterface interface {
|
||||
Debug(msg string)
|
||||
Info(msg string)
|
||||
Error(msg string)
|
||||
Warn(msg string)
|
||||
}
|
||||
|
||||
// WorkerTask представляет задачу для выполнения в пуле
|
||||
type WorkerTask struct {
|
||||
ID string
|
||||
Handler func() error
|
||||
CreatedAt int64
|
||||
}
|
||||
|
||||
// WorkerPool управляет пулом воркеров
|
||||
type WorkerPool struct {
|
||||
maxWorkers int
|
||||
taskQueue chan *WorkerTask
|
||||
wg sync.WaitGroup
|
||||
stopChan chan struct{}
|
||||
activeTasks atomic.Int32
|
||||
totalTasks atomic.Uint64
|
||||
failedTasks atomic.Uint64
|
||||
logger LoggerInterface
|
||||
panicHandler func(interface{})
|
||||
}
|
||||
|
||||
// NewWorkerPool создаёт новый пул воркеров
|
||||
func NewWorkerPool(maxWorkers int, logger LoggerInterface) *WorkerPool {
|
||||
if maxWorkers <= 0 {
|
||||
maxWorkers = 100 // ограничение по умолчанию
|
||||
}
|
||||
|
||||
pool := &WorkerPool{
|
||||
maxWorkers: maxWorkers,
|
||||
taskQueue: make(chan *WorkerTask, maxWorkers*10),
|
||||
stopChan: make(chan struct{}),
|
||||
logger: logger,
|
||||
panicHandler: defaultPanicHandler,
|
||||
}
|
||||
|
||||
for i := 0; i < maxWorkers; i++ {
|
||||
pool.wg.Add(1)
|
||||
go pool.worker(i)
|
||||
}
|
||||
|
||||
return pool
|
||||
}
|
||||
|
||||
// defaultPanicHandler обрабатывает панику в горутине
|
||||
func defaultPanicHandler(r interface{}) {
|
||||
fmt.Printf("PANIC recovered: %v\n%s\n", r, debug.Stack())
|
||||
}
|
||||
|
||||
// SetPanicHandler устанавливает обработчик паник
|
||||
func (wp *WorkerPool) SetPanicHandler(handler func(interface{})) {
|
||||
if handler != nil {
|
||||
wp.panicHandler = handler
|
||||
}
|
||||
}
|
||||
|
||||
// worker запускает воркер
|
||||
func (wp *WorkerPool) worker(id int) {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
if wp.panicHandler != nil {
|
||||
wp.panicHandler(r)
|
||||
}
|
||||
if wp.logger != nil {
|
||||
wp.logger.Error(fmt.Sprintf("Worker %d panicked: %v", id, r))
|
||||
}
|
||||
// Перезапускаем воркер
|
||||
wp.wg.Add(1)
|
||||
go wp.worker(id)
|
||||
}
|
||||
wp.wg.Done()
|
||||
}()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-wp.stopChan:
|
||||
return
|
||||
case task, ok := <-wp.taskQueue:
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
|
||||
wp.activeTasks.Add(1)
|
||||
wp.totalTasks.Add(1)
|
||||
|
||||
func() {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
wp.failedTasks.Add(1)
|
||||
if wp.panicHandler != nil {
|
||||
wp.panicHandler(r)
|
||||
}
|
||||
if wp.logger != nil {
|
||||
wp.logger.Error(fmt.Sprintf("Task %s panicked: %v", task.ID, r))
|
||||
}
|
||||
}
|
||||
wp.activeTasks.Add(-1)
|
||||
}()
|
||||
|
||||
if err := task.Handler(); err != nil {
|
||||
wp.failedTasks.Add(1)
|
||||
if wp.logger != nil {
|
||||
wp.logger.Error(fmt.Sprintf("Task %s failed: %v", task.ID, err))
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Submit отправляет задачу в пул
|
||||
func (wp *WorkerPool) Submit(task *WorkerTask) error {
|
||||
select {
|
||||
case wp.taskQueue <- task:
|
||||
return nil
|
||||
case <-time.After(5 * time.Second):
|
||||
return fmt.Errorf("task queue full, task rejected: %s", task.ID)
|
||||
}
|
||||
}
|
||||
|
||||
// SubmitFunc отправляет функцию как задачу
|
||||
func (wp *WorkerPool) SubmitFunc(id string, handler func() error) error {
|
||||
return wp.Submit(&WorkerTask{
|
||||
ID: id,
|
||||
Handler: handler,
|
||||
CreatedAt: time.Now().UnixMilli(),
|
||||
})
|
||||
}
|
||||
|
||||
// GetStats возвращает статистику пула
|
||||
func (wp *WorkerPool) GetStats() map[string]interface{} {
|
||||
return map[string]interface{}{
|
||||
"max_workers": wp.maxWorkers,
|
||||
"active_tasks": wp.activeTasks.Load(),
|
||||
"total_tasks": wp.totalTasks.Load(),
|
||||
"failed_tasks": wp.failedTasks.Load(),
|
||||
"queue_size": len(wp.taskQueue),
|
||||
"queue_capacity": cap(wp.taskQueue),
|
||||
}
|
||||
}
|
||||
|
||||
// GetWorkerPoolStats возвращает статистику в структурированном виде
|
||||
func (wp *WorkerPool) GetWorkerPoolStats() *WorkerPoolStats {
|
||||
return &WorkerPoolStats{
|
||||
MaxWorkers: wp.maxWorkers,
|
||||
ActiveTasks: wp.activeTasks.Load(),
|
||||
TotalTasks: wp.totalTasks.Load(),
|
||||
FailedTasks: wp.failedTasks.Load(),
|
||||
QueueSize: len(wp.taskQueue),
|
||||
QueueCapacity: cap(wp.taskQueue),
|
||||
}
|
||||
}
|
||||
|
||||
// Stop останавливает пул воркеров
|
||||
func (wp *WorkerPool) Stop() {
|
||||
close(wp.stopChan)
|
||||
close(wp.taskQueue)
|
||||
wp.wg.Wait()
|
||||
}
|
||||
|
||||
// SafeGo безопасно запускает горутину с обработкой паники
|
||||
func SafeGo(fn func(), panicHandler func(interface{})) {
|
||||
go func() {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
if panicHandler != nil {
|
||||
panicHandler(r)
|
||||
}
|
||||
}
|
||||
}()
|
||||
fn()
|
||||
}()
|
||||
}
|
||||
|
||||
// SafeGoWithLogger безопасно запускает горутину с логгером
|
||||
func SafeGoWithLogger(fn func(), logger LoggerInterface, name string) {
|
||||
go func() {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
if logger != nil {
|
||||
logger.Error(fmt.Sprintf("Goroutine %s panicked: %v\n%s", name, r, debug.Stack()))
|
||||
}
|
||||
}
|
||||
}()
|
||||
fn()
|
||||
}()
|
||||
}
|
||||
Reference in New Issue
Block a user