811 lines
24 KiB
Go
811 lines
24 KiB
Go
// /futriis/internal/cluster/node.go
|
||
// Пакет cluster реализует управление кластером, координацию узлов и репликацию данных.
|
||
// Обеспечивает обнаружение узлов, heartbeat механизм для мониторинга доступности,
|
||
// а также синхронную мастер-мастер репликацию между узлами кластера.
|
||
// Поддерживает автоматическое переключение ролей и балансировку нагрузки.
|
||
|
||
package cluster
|
||
|
||
import (
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"net"
|
||
"sync"
|
||
"sync/atomic"
|
||
"time"
|
||
"unsafe"
|
||
|
||
"futriis/pkg/config"
|
||
)
|
||
|
||
// Простые функции для логирования (без зависимостей)
|
||
func printInfo(format string, args ...interface{}) {
|
||
fmt.Printf("\033[34m[INFO]\033[0m %s\n", fmt.Sprintf(format, args...))
|
||
}
|
||
|
||
func printSuccess(format string, args ...interface{}) {
|
||
fmt.Printf("\033[32m[OK]\033[0m %s\n", fmt.Sprintf(format, args...))
|
||
}
|
||
|
||
func printWarning(format string, args ...interface{}) {
|
||
fmt.Printf("\033[33m[WARN]\033[0m %s\n", fmt.Sprintf(format, args...))
|
||
}
|
||
|
||
func printError(format string, args ...interface{}) {
|
||
fmt.Printf("\033[31m[ERROR]\033[0m %s\n", fmt.Sprintf(format, args...))
|
||
}
|
||
|
||
// NodeState состояние узла
|
||
type NodeState int
|
||
|
||
const (
|
||
StateOffline NodeState = iota
|
||
StateJoining
|
||
StateOnline
|
||
StateLeaving
|
||
StateFailed
|
||
)
|
||
|
||
func (s NodeState) String() string {
|
||
switch s {
|
||
case StateOffline:
|
||
return "offline"
|
||
case StateJoining:
|
||
return "joining"
|
||
case StateOnline:
|
||
return "online"
|
||
case StateLeaving:
|
||
return "leaving"
|
||
case StateFailed:
|
||
return "failed"
|
||
default:
|
||
return "unknown"
|
||
}
|
||
}
|
||
|
||
// ReplicationMessage представляет сообщение репликации
|
||
type ReplicationMessage struct {
|
||
Type string `json:"type"` // "write", "sync", "ack"
|
||
Command string `json:"command"` // Команда (create, update, delete)
|
||
Args []interface{} `json:"args"` // Аргументы команды
|
||
Timestamp int64 `json:"timestamp"` // Временная метка
|
||
NodeID string `json:"node_id"` // ID исходного узла
|
||
ShardID string `json:"shard_id"` // ID шарда
|
||
}
|
||
|
||
// Node представляет узел кластера (с wait-free указателями)
|
||
type Node struct {
|
||
ID string
|
||
Address string
|
||
state int32 // Атомарное состояние
|
||
LastSeen time.Time
|
||
ShardIDs []string
|
||
// Используем atomic.Value для wait-free доступа к изменяемым полям
|
||
nodeState unsafe.Pointer // Атомарный указатель на карту состояний
|
||
}
|
||
|
||
// GetState атомарно получает состояние узла
|
||
func (n *Node) GetState() NodeState {
|
||
return NodeState(atomic.LoadInt32(&n.state))
|
||
}
|
||
|
||
// SetState атомарно устанавливает состояние узла
|
||
func (n *Node) SetState(state NodeState) {
|
||
atomic.StoreInt32(&n.state, int32(state))
|
||
}
|
||
|
||
// ClusterManager управляет кластером (с wait-free операциями)
|
||
type ClusterManager struct {
|
||
nodes unsafe.Pointer // Атомарный указатель на карту узлов
|
||
coordinatorAddr string
|
||
nodeID string
|
||
localAddr string
|
||
isCoordinator int32 // Атомарный флаг
|
||
heartbeatStop chan struct{}
|
||
replicationStop chan struct{}
|
||
stats struct {
|
||
totalNodes int64
|
||
activeNodes int64
|
||
rebalancingCnt int64
|
||
}
|
||
replicationEnabled bool
|
||
masterMaster bool
|
||
replicationQueue chan ReplicationMessage
|
||
replicationWG sync.WaitGroup
|
||
shardManager *ShardManager // Менеджер шардинга
|
||
}
|
||
|
||
// NewClusterManager создаёт новый менеджер кластера
|
||
func NewClusterManager(cfg *config.Config) *ClusterManager {
|
||
// Определяем стратегию шардинга по умолчанию
|
||
shardingStrategy := ConsistentHashing
|
||
shardingEnabled := false
|
||
initialShards := 10
|
||
|
||
// Проверяем наличие настроек шардинга в конфигурации
|
||
// Используем значения по умолчанию, если поля отсутствуют
|
||
|
||
cm := &ClusterManager{
|
||
coordinatorAddr: cfg.Cluster.CoordinatorAddress,
|
||
nodeID: cfg.Node.ID,
|
||
localAddr: cfg.Node.Address,
|
||
heartbeatStop: make(chan struct{}),
|
||
replicationStop: make(chan struct{}),
|
||
replicationEnabled: cfg.Replication.Enabled,
|
||
masterMaster: cfg.Replication.MasterMaster,
|
||
replicationQueue: make(chan ReplicationMessage, 1000),
|
||
}
|
||
|
||
// Создаём менеджер шардинга с значениями по умолчанию
|
||
cm.shardManager = NewShardManager(
|
||
shardingStrategy,
|
||
cfg.Cluster.ReplicationFactor,
|
||
shardingEnabled,
|
||
)
|
||
|
||
// Устанавливаем флаг координатора атомарно
|
||
isCoord := int32(0)
|
||
if cfg.Node.Address == cfg.Cluster.CoordinatorAddress {
|
||
isCoord = 1
|
||
}
|
||
atomic.StoreInt32(&cm.isCoordinator, isCoord)
|
||
|
||
// Создаём начальную карту узлов
|
||
nodes := make(map[string]*Node)
|
||
|
||
// Добавляем себя в кластер
|
||
selfNode := &Node{
|
||
ID: cfg.Node.ID,
|
||
Address: cfg.Node.Address,
|
||
LastSeen: time.Now(),
|
||
}
|
||
selfNode.SetState(StateOnline)
|
||
nodes[cfg.Node.ID] = selfNode
|
||
|
||
// Атомарно сохраняем карту узлов
|
||
atomic.StorePointer(&cm.nodes, unsafe.Pointer(&nodes))
|
||
|
||
atomic.AddInt64(&cm.stats.totalNodes, 1)
|
||
atomic.AddInt64(&cm.stats.activeNodes, 1)
|
||
|
||
// Создаём начальные шарды если включён шардинг
|
||
if shardingEnabled {
|
||
// Получаем список всех узлов
|
||
allNodes := make([]string, 0)
|
||
for id := range nodes {
|
||
allNodes = append(allNodes, id)
|
||
}
|
||
|
||
// Создаём шарды
|
||
cm.shardManager.CreateShards(initialShards, allNodes)
|
||
}
|
||
|
||
// Запускаем обработчик репликации если включена мастер-мастер
|
||
if cm.replicationEnabled && cm.masterMaster {
|
||
go cm.startReplicationHandler()
|
||
}
|
||
|
||
return cm
|
||
}
|
||
|
||
// IsCoordinator атомарно проверяет, является ли узел координатором
|
||
func (cm *ClusterManager) IsCoordinator() bool {
|
||
return atomic.LoadInt32(&cm.isCoordinator) == 1
|
||
}
|
||
|
||
// Start запускает кластерные сервисы
|
||
func (cm *ClusterManager) Start() error {
|
||
if cm.IsCoordinator() {
|
||
// Запускаем координатор
|
||
go cm.startCoordinator()
|
||
}
|
||
|
||
// Запускаем heartbeat
|
||
go cm.heartbeatLoop()
|
||
|
||
printInfo("Кластерный менеджер запущен")
|
||
if cm.replicationEnabled && cm.masterMaster {
|
||
printInfo("Мастер-мастер репликация активирована")
|
||
}
|
||
if cm.shardManager.enabled {
|
||
printInfo("Шардинг активирован, стратегия: %v", cm.shardManager.getStrategyName())
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// Stop останавливает кластерные сервисы
|
||
func (cm *ClusterManager) Stop() {
|
||
close(cm.heartbeatStop)
|
||
if cm.replicationEnabled && cm.masterMaster {
|
||
close(cm.replicationStop)
|
||
cm.replicationWG.Wait()
|
||
}
|
||
}
|
||
|
||
// ReplicateCommand реплицирует команду на все узлы кластера с учётом шардинга
|
||
func (cm *ClusterManager) ReplicateCommand(cmd string, args []interface{}, key string) error {
|
||
if !cm.replicationEnabled || !cm.masterMaster {
|
||
return nil // Репликация не включена
|
||
}
|
||
|
||
// Определяем шард для ключа
|
||
var shardID string
|
||
if cm.shardManager.enabled && key != "" {
|
||
shard, err := cm.shardManager.GetShardForKey(key)
|
||
if err == nil && shard != nil {
|
||
shardID = shard.ID
|
||
cm.shardManager.RecordWrite(shardID)
|
||
}
|
||
}
|
||
|
||
msg := ReplicationMessage{
|
||
Type: "write",
|
||
Command: cmd,
|
||
Args: args,
|
||
Timestamp: time.Now().UnixNano(),
|
||
NodeID: cm.nodeID,
|
||
ShardID: shardID,
|
||
}
|
||
|
||
// Отправляем в очередь репликации
|
||
select {
|
||
case cm.replicationQueue <- msg:
|
||
return nil
|
||
default:
|
||
return errors.New("очередь репликации переполнена")
|
||
}
|
||
}
|
||
|
||
// startReplicationHandler запускает обработчик репликации
|
||
func (cm *ClusterManager) startReplicationHandler() {
|
||
cm.replicationWG.Add(1)
|
||
defer cm.replicationWG.Done()
|
||
|
||
ticker := time.NewTicker(1 * time.Second)
|
||
defer ticker.Stop()
|
||
|
||
for {
|
||
select {
|
||
case msg := <-cm.replicationQueue:
|
||
cm.sendReplicationMessage(msg)
|
||
case <-ticker.C:
|
||
cm.sendSyncRequest()
|
||
case <-cm.replicationStop:
|
||
return
|
||
}
|
||
}
|
||
}
|
||
|
||
// sendReplicationMessage отправляет сообщение репликации на другие узлы
|
||
func (cm *ClusterManager) sendReplicationMessage(msg ReplicationMessage) {
|
||
// Получаем текущую карту узлов атомарно
|
||
nodesPtr := atomic.LoadPointer(&cm.nodes)
|
||
if nodesPtr == nil {
|
||
return
|
||
}
|
||
nodes := *(*map[string]*Node)(nodesPtr)
|
||
|
||
for id, node := range nodes {
|
||
if id == cm.nodeID || node.GetState() != StateOnline {
|
||
continue // Не отправляем себе и офлайн узлам
|
||
}
|
||
|
||
// Если есть шард, проверяем, должен ли узел получать это сообщение
|
||
if msg.ShardID != "" && cm.shardManager.enabled {
|
||
shard, exists := cm.shardManager.GetShardByID(msg.ShardID)
|
||
if exists {
|
||
shouldSend := false
|
||
for _, nodeID := range shard.Nodes {
|
||
if nodeID == id {
|
||
shouldSend = true
|
||
break
|
||
}
|
||
}
|
||
if !shouldSend {
|
||
continue
|
||
}
|
||
}
|
||
}
|
||
|
||
go func(targetNode *Node) {
|
||
conn, err := net.DialTimeout("tcp", targetNode.Address, 3*time.Second)
|
||
if err != nil {
|
||
printWarning("Не удалось подключиться к узлу %s для репликации: %v", targetNode.ID, err)
|
||
return
|
||
}
|
||
defer conn.Close()
|
||
|
||
encoder := json.NewEncoder(conn)
|
||
if err := encoder.Encode(msg); err != nil {
|
||
printWarning("Ошибка отправки репликации на узел %s: %v", targetNode.ID, err)
|
||
return
|
||
}
|
||
|
||
// Ожидаем подтверждение для синхронной репликации
|
||
if msg.Type == "write" {
|
||
var ack map[string]interface{}
|
||
decoder := json.NewDecoder(conn)
|
||
if err := decoder.Decode(&ack); err == nil {
|
||
if status, ok := ack["status"].(string); ok && status == "ok" {
|
||
printSuccess("Репликация на узел %s подтверждена", targetNode.ID)
|
||
}
|
||
}
|
||
}
|
||
}(node)
|
||
}
|
||
}
|
||
|
||
// sendSyncRequest отправляет запрос синхронизации
|
||
func (cm *ClusterManager) sendSyncRequest() {
|
||
if !cm.IsCoordinator() {
|
||
return // Только координатор инициирует синхронизацию
|
||
}
|
||
|
||
msg := ReplicationMessage{
|
||
Type: "sync",
|
||
Timestamp: time.Now().UnixNano(),
|
||
NodeID: cm.nodeID,
|
||
}
|
||
|
||
cm.sendReplicationMessage(msg)
|
||
}
|
||
|
||
// handleReplicationMessage обрабатывает входящее сообщение репликации
|
||
func (cm *ClusterManager) handleReplicationMessage(conn net.Conn) {
|
||
defer conn.Close()
|
||
|
||
var msg ReplicationMessage
|
||
decoder := json.NewDecoder(conn)
|
||
if err := decoder.Decode(&msg); err != nil {
|
||
return
|
||
}
|
||
|
||
switch msg.Type {
|
||
case "write":
|
||
// Получаем команду от другого узла
|
||
printInfo("Получена команда репликации: %s от узла %s (шард: %s)",
|
||
msg.Command, msg.NodeID, msg.ShardID)
|
||
|
||
// Здесь будет вызов движка для выполнения команды
|
||
// TODO: Интегрировать с engine для выполнения реплицированных команд
|
||
|
||
// Отправляем подтверждение
|
||
encoder := json.NewEncoder(conn)
|
||
encoder.Encode(map[string]interface{}{
|
||
"status": "ok",
|
||
"time": time.Now().UnixNano(),
|
||
})
|
||
|
||
case "sync":
|
||
// Запрос синхронизации
|
||
printInfo("Получен запрос синхронизации от узла %s", msg.NodeID)
|
||
// TODO: Отправить текущее состояние
|
||
|
||
case "ack":
|
||
// Подтверждение получения
|
||
printInfo("Получено подтверждение от узла %s", msg.NodeID)
|
||
}
|
||
}
|
||
|
||
// heartbeatLoop отправляет heartbeat сигналы
|
||
func (cm *ClusterManager) heartbeatLoop() {
|
||
ticker := time.NewTicker(5 * time.Second)
|
||
defer ticker.Stop()
|
||
|
||
for {
|
||
select {
|
||
case <-ticker.C:
|
||
cm.sendHeartbeat()
|
||
case <-cm.heartbeatStop:
|
||
return
|
||
}
|
||
}
|
||
}
|
||
|
||
// sendHeartbeat отправляет heartbeat координатору
|
||
func (cm *ClusterManager) sendHeartbeat() {
|
||
if !cm.IsCoordinator() {
|
||
// Отправляем heartbeat координатору
|
||
conn, err := net.DialTimeout("tcp", cm.coordinatorAddr, 3*time.Second)
|
||
if err != nil {
|
||
return
|
||
}
|
||
defer conn.Close()
|
||
|
||
heartbeat := map[string]interface{}{
|
||
"type": "heartbeat",
|
||
"node_id": cm.nodeID,
|
||
"address": cm.localAddr,
|
||
"time": time.Now().Unix(),
|
||
}
|
||
|
||
json.NewEncoder(conn).Encode(heartbeat)
|
||
}
|
||
}
|
||
|
||
// startCoordinator запускает координатор кластера
|
||
func (cm *ClusterManager) startCoordinator() {
|
||
listener, err := net.Listen("tcp", cm.coordinatorAddr)
|
||
if err != nil {
|
||
printError("Ошибка запуска координатора: %v", err)
|
||
return
|
||
}
|
||
defer listener.Close()
|
||
|
||
printInfo("Координатор кластера запущен на " + cm.coordinatorAddr)
|
||
|
||
for {
|
||
conn, err := listener.Accept()
|
||
if err != nil {
|
||
continue
|
||
}
|
||
|
||
go cm.handleCoordinatorRequest(conn)
|
||
}
|
||
}
|
||
|
||
// handleCoordinatorRequest обрабатывает запросы к координатору
|
||
func (cm *ClusterManager) handleCoordinatorRequest(conn net.Conn) {
|
||
defer conn.Close()
|
||
|
||
var req map[string]interface{}
|
||
if err := json.NewDecoder(conn).Decode(&req); err != nil {
|
||
return
|
||
}
|
||
|
||
msgType, _ := req["type"].(string)
|
||
|
||
switch msgType {
|
||
case "heartbeat":
|
||
nodeID, _ := req["node_id"].(string)
|
||
address, _ := req["address"].(string)
|
||
cm.updateNodeHeartbeat(nodeID, address)
|
||
|
||
case "join":
|
||
nodeID, _ := req["node_id"].(string)
|
||
address, _ := req["address"].(string)
|
||
cm.handleNodeJoin(nodeID, address)
|
||
|
||
case "leave":
|
||
nodeID, _ := req["node_id"].(string)
|
||
cm.handleNodeLeave(nodeID)
|
||
|
||
case "replication":
|
||
// Обработка сообщения репликации на координаторе
|
||
cm.handleReplicationMessage(conn)
|
||
}
|
||
}
|
||
|
||
// handleNodeJoin обрабатывает присоединение узла
|
||
func (cm *ClusterManager) handleNodeJoin(nodeID, address string) {
|
||
// Получаем текущую карту узлов
|
||
nodesPtr := atomic.LoadPointer(&cm.nodes)
|
||
if nodesPtr == nil {
|
||
return
|
||
}
|
||
oldNodes := *(*map[string]*Node)(nodesPtr)
|
||
|
||
// Создаём новую карту
|
||
newNodes := make(map[string]*Node)
|
||
for k, v := range oldNodes {
|
||
newNodes[k] = v
|
||
}
|
||
|
||
if node, exists := newNodes[nodeID]; exists {
|
||
node.SetState(StateOnline)
|
||
node.LastSeen = time.Now()
|
||
} else {
|
||
newNode := &Node{
|
||
ID: nodeID,
|
||
Address: address,
|
||
LastSeen: time.Now(),
|
||
}
|
||
newNode.SetState(StateOnline)
|
||
newNodes[nodeID] = newNode
|
||
atomic.AddInt64(&cm.stats.totalNodes, 1)
|
||
}
|
||
|
||
// Атомарно обновляем карту узлов
|
||
atomic.StorePointer(&cm.nodes, unsafe.Pointer(&newNodes))
|
||
|
||
atomic.AddInt64(&cm.stats.activeNodes, 1)
|
||
printSuccess("Узел %s (%s) присоединился к кластеру", nodeID, address)
|
||
|
||
// Если включена мастер-мастер репликация, отправляем текущее состояние новому узлу
|
||
if cm.replicationEnabled && cm.masterMaster {
|
||
go cm.sendInitialSync(nodeID)
|
||
}
|
||
|
||
// Если включён шардинг, обновляем распределение шардов
|
||
if cm.shardManager.enabled {
|
||
cm.rebalanceShards()
|
||
}
|
||
}
|
||
|
||
// sendInitialSync отправляет начальную синхронизацию новому узлу
|
||
func (cm *ClusterManager) sendInitialSync(targetNodeID string) {
|
||
// Получаем текущую карту узлов
|
||
nodesPtr := atomic.LoadPointer(&cm.nodes)
|
||
if nodesPtr == nil {
|
||
return
|
||
}
|
||
nodes := *(*map[string]*Node)(nodesPtr)
|
||
|
||
targetNode, exists := nodes[targetNodeID]
|
||
if !exists || targetNode.GetState() != StateOnline {
|
||
return
|
||
}
|
||
|
||
// TODO: Отправить текущее состояние базы данных новому узлу
|
||
printInfo("Отправка начальной синхронизации узлу %s", targetNodeID)
|
||
}
|
||
|
||
// handleNodeLeave обрабатывает отключение узла
|
||
func (cm *ClusterManager) handleNodeLeave(nodeID string) {
|
||
// Получаем текущую карту узлов
|
||
nodesPtr := atomic.LoadPointer(&cm.nodes)
|
||
if nodesPtr == nil {
|
||
return
|
||
}
|
||
oldNodes := *(*map[string]*Node)(nodesPtr)
|
||
|
||
// Создаём новую карту
|
||
newNodes := make(map[string]*Node)
|
||
for k, v := range oldNodes {
|
||
if k == nodeID {
|
||
v.SetState(StateOffline)
|
||
newNodes[k] = v
|
||
} else {
|
||
newNodes[k] = v
|
||
}
|
||
}
|
||
|
||
// Атомарно обновляем карту узлов
|
||
atomic.StorePointer(&cm.nodes, unsafe.Pointer(&newNodes))
|
||
|
||
atomic.AddInt64(&cm.stats.activeNodes, -1)
|
||
printWarning("Узел %s покинул кластер", nodeID)
|
||
|
||
// Если включён шардинг, обновляем распределение шардов
|
||
if cm.shardManager.enabled {
|
||
cm.rebalanceShards()
|
||
}
|
||
}
|
||
|
||
// updateNodeHeartbeat обновляет время последнего heartbeat узла
|
||
func (cm *ClusterManager) updateNodeHeartbeat(nodeID, address string) {
|
||
// Получаем текущую карту узлов
|
||
nodesPtr := atomic.LoadPointer(&cm.nodes)
|
||
if nodesPtr == nil {
|
||
return
|
||
}
|
||
oldNodes := *(*map[string]*Node)(nodesPtr)
|
||
|
||
// Проверяем, нужно ли обновление
|
||
node, exists := oldNodes[nodeID]
|
||
if exists && node.GetState() == StateOnline && time.Since(node.LastSeen) < 5*time.Second {
|
||
// Обновление не требуется, просто обновляем LastSeen в существующей карте
|
||
node.LastSeen = time.Now()
|
||
return
|
||
}
|
||
|
||
// Создаём новую карту для обновления
|
||
newNodes := make(map[string]*Node)
|
||
for k, v := range oldNodes {
|
||
newNodes[k] = v
|
||
}
|
||
|
||
if exists {
|
||
node.LastSeen = time.Now()
|
||
if node.GetState() == StateOffline {
|
||
node.SetState(StateOnline)
|
||
atomic.AddInt64(&cm.stats.activeNodes, 1)
|
||
}
|
||
} else {
|
||
newNode := &Node{
|
||
ID: nodeID,
|
||
Address: address,
|
||
LastSeen: time.Now(),
|
||
}
|
||
newNode.SetState(StateOnline)
|
||
newNodes[nodeID] = newNode
|
||
atomic.AddInt64(&cm.stats.totalNodes, 1)
|
||
atomic.AddInt64(&cm.stats.activeNodes, 1)
|
||
}
|
||
|
||
// Атомарно обновляем карту узлов
|
||
atomic.StorePointer(&cm.nodes, unsafe.Pointer(&newNodes))
|
||
}
|
||
|
||
// rebalanceShards выполняет ребалансировку шардов при изменении состава кластера
|
||
func (cm *ClusterManager) rebalanceShards() {
|
||
if !cm.shardManager.enabled || !cm.IsCoordinator() {
|
||
return
|
||
}
|
||
|
||
// Получаем текущие узлы
|
||
nodesPtr := atomic.LoadPointer(&cm.nodes)
|
||
if nodesPtr == nil {
|
||
return
|
||
}
|
||
nodes := *(*map[string]*Node)(nodesPtr)
|
||
|
||
// Собираем онлайн узлы
|
||
onlineNodes := make([]string, 0)
|
||
for id, node := range nodes {
|
||
if node.GetState() == StateOnline {
|
||
onlineNodes = append(onlineNodes, id)
|
||
}
|
||
}
|
||
|
||
// Здесь должна быть логика ребалансировки шардов
|
||
// В демо-версии просто перераспределяем существующие шарды
|
||
|
||
cm.shardManager.Rebalance()
|
||
|
||
atomic.AddInt64(&cm.stats.rebalancingCnt, 1)
|
||
}
|
||
|
||
// AddNode добавляет новый узел в кластер
|
||
func (cm *ClusterManager) AddNode(address string) error {
|
||
if !cm.IsCoordinator() {
|
||
return errors.New("только координатор может добавлять узлы")
|
||
}
|
||
|
||
// Получаем текущую карту узлов
|
||
nodesPtr := atomic.LoadPointer(&cm.nodes)
|
||
if nodesPtr == nil {
|
||
return errors.New("карта узлов не инициализирована")
|
||
}
|
||
oldNodes := *(*map[string]*Node)(nodesPtr)
|
||
|
||
// Генерируем ID для нового узла
|
||
nodeID := fmt.Sprintf("node-%d", len(oldNodes)+1)
|
||
|
||
// Создаём новую карту
|
||
newNodes := make(map[string]*Node)
|
||
for k, v := range oldNodes {
|
||
newNodes[k] = v
|
||
}
|
||
|
||
newNode := &Node{
|
||
ID: nodeID,
|
||
Address: address,
|
||
LastSeen: time.Now(),
|
||
}
|
||
newNode.SetState(StateJoining)
|
||
newNodes[nodeID] = newNode
|
||
|
||
// Атомарно обновляем карту узлов
|
||
atomic.StorePointer(&cm.nodes, unsafe.Pointer(&newNodes))
|
||
|
||
atomic.AddInt64(&cm.stats.totalNodes, 1)
|
||
|
||
printSuccess("Узел %s (%s) добавлен в кластер", nodeID, address)
|
||
|
||
// Если включён шардинг, обновляем распределение
|
||
if cm.shardManager.enabled {
|
||
cm.rebalanceShards()
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// RemoveNode удаляет узел из кластера
|
||
func (cm *ClusterManager) RemoveNode(nodeID string) error {
|
||
if !cm.IsCoordinator() {
|
||
return errors.New("только координатор может удалять узлы")
|
||
}
|
||
|
||
// Получаем текущую карту узлов
|
||
nodesPtr := atomic.LoadPointer(&cm.nodes)
|
||
if nodesPtr == nil {
|
||
return errors.New("карта узлов не инициализирована")
|
||
}
|
||
oldNodes := *(*map[string]*Node)(nodesPtr)
|
||
|
||
if _, exists := oldNodes[nodeID]; !exists {
|
||
return errors.New("узел не найден")
|
||
}
|
||
|
||
// Создаём новую карту без удаляемого узла
|
||
newNodes := make(map[string]*Node)
|
||
for k, v := range oldNodes {
|
||
if k != nodeID {
|
||
newNodes[k] = v
|
||
}
|
||
}
|
||
|
||
// Атомарно обновляем карту узлов
|
||
atomic.StorePointer(&cm.nodes, unsafe.Pointer(&newNodes))
|
||
|
||
atomic.AddInt64(&cm.stats.totalNodes, -1)
|
||
|
||
printSuccess("Узел %s удален из кластера", nodeID)
|
||
|
||
// Если включён шардинг, обновляем распределение
|
||
if cm.shardManager.enabled {
|
||
cm.rebalanceShards()
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// GetClusterStatus возвращает статус кластера
|
||
func (cm *ClusterManager) GetClusterStatus() map[string]interface{} {
|
||
status := make(map[string]interface{})
|
||
status["total_nodes"] = atomic.LoadInt64(&cm.stats.totalNodes)
|
||
status["active_nodes"] = atomic.LoadInt64(&cm.stats.activeNodes)
|
||
status["is_coordinator"] = cm.IsCoordinator()
|
||
status["coordinator"] = cm.coordinatorAddr
|
||
status["replication_enabled"] = cm.replicationEnabled
|
||
status["master_master"] = cm.masterMaster
|
||
|
||
// Добавляем информацию о шардинге
|
||
if cm.shardManager.enabled {
|
||
shardStats := cm.shardManager.GetShardStats()
|
||
status["sharding"] = shardStats
|
||
}
|
||
|
||
// Получаем текущую карту узлов атомарно
|
||
nodesPtr := atomic.LoadPointer(&cm.nodes)
|
||
if nodesPtr != nil {
|
||
nodes := *(*map[string]*Node)(nodesPtr)
|
||
nodesList := make([]map[string]interface{}, 0, len(nodes))
|
||
for id, node := range nodes {
|
||
nodesList = append(nodesList, map[string]interface{}{
|
||
"id": id,
|
||
"address": node.Address,
|
||
"state": node.GetState().String(),
|
||
"last_seen": node.LastSeen.Format(time.RFC3339),
|
||
})
|
||
}
|
||
status["nodes"] = nodesList
|
||
}
|
||
|
||
return status
|
||
}
|
||
|
||
// RebalanceCluster выполняет ребалансировку всего кластера
|
||
func (cm *ClusterManager) RebalanceCluster() error {
|
||
if !cm.IsCoordinator() {
|
||
return errors.New("only coordinator can rebalance the cluster")
|
||
}
|
||
|
||
printInfo("Starting cluster rebalance...")
|
||
|
||
// Получаем текущие узлы
|
||
nodesPtr := atomic.LoadPointer(&cm.nodes)
|
||
if nodesPtr == nil {
|
||
return errors.New("node map not initialized")
|
||
}
|
||
nodes := *(*map[string]*Node)(nodesPtr)
|
||
|
||
// Собираем онлайн узлы
|
||
onlineNodes := make([]string, 0)
|
||
for id, node := range nodes {
|
||
if node.GetState() == StateOnline {
|
||
onlineNodes = append(onlineNodes, id)
|
||
}
|
||
}
|
||
|
||
if len(onlineNodes) == 0 {
|
||
return errors.New("no online nodes available for rebalance")
|
||
}
|
||
|
||
// Если включён шардинг, ребалансируем шарды
|
||
if cm.shardManager.enabled {
|
||
err := cm.shardManager.Rebalance()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
}
|
||
|
||
atomic.AddInt64(&cm.stats.rebalancingCnt, 1)
|
||
|
||
printSuccess("Cluster rebalance completed successfully")
|
||
|
||
return nil
|
||
}
|