diff --git a/internal/cluster/types.go b/internal/cluster/types.go new file mode 100644 index 0000000..252fb70 --- /dev/null +++ b/internal/cluster/types.go @@ -0,0 +1,239 @@ +/* + * 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 +// Назначение: Общие типы данных для кластерных операций с поддержкой временных меток +// NOTE: NodeInfo, NodeRequest, ShardInfo, ClusterStatus определены в node.go и raft_coordinator.go +// Чтобы избежать дублирования, здесь определяем только вспомогательные типы и функции + +package cluster + +import ( + "fmt" + "time" +) + + + +// ========== ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ ДЛЯ NodeInfo ========== + +// NodeJoinedAt возвращает человекочитаемое время присоединения узла +func NodeJoinedAt(joinedAt int64) string { + if joinedAt == 0 { + return "not joined" + } + return time.UnixMilli(joinedAt).Format("2006-01-02 15:04:05.000") +} + +// LastSeenAt возвращает человекочитаемое время последнего контакта +func LastSeenAt(lastSeen int64) string { + if lastSeen == 0 { + return "never" + } + return time.UnixMilli(lastSeen).Format("2006-01-02 15:04:05.000") +} + +// GetNodeUptime возвращает время жизни узла в кластере +func GetNodeUptime(joinedAt int64) time.Duration { + if joinedAt == 0 { + return 0 + } + return time.Duration(time.Now().UnixMilli()-joinedAt) * time.Millisecond +} + +// IsNodeHealthy проверяет, здоров ли узел +func IsNodeHealthy(status string, lastSeen int64) bool { + return status == "active" && time.Now().UnixMilli()-lastSeen < 30000 +} + +// GetNodeAddress возвращает адрес узла +func GetNodeAddress(ip string, port int) string { + return fmt.Sprintf("%s:%d", ip, port) +} + +// ========== ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ ДЛЯ ShardInfo ========== + +// GetShardUptime возвращает время жизни шарда +func GetShardUptime(createdAt int64) time.Duration { + if createdAt == 0 { + return 0 + } + return time.Duration(time.Now().UnixMilli()-createdAt) * time.Millisecond +} + +// IsShardHealthy проверяет, здоров ли шард +func IsShardHealthy(status string) bool { + return status == "active" +} + +// GetShardLastRebalancedStr возвращает человекочитаемое время последней ребалансировки +func GetShardLastRebalancedStr(lastRebalanced int64) string { + if lastRebalanced == 0 { + return "never" + } + return time.UnixMilli(lastRebalanced).Format("2006-01-02 15:04:05.000") +} + +// GetShardCreatedAtStr возвращает человекочитаемое время создания +func GetShardCreatedAtStr(createdAt int64) string { + if createdAt == 0 { + return "unknown" + } + return time.UnixMilli(createdAt).Format("2006-01-02 15:04:05.000") +} + +// GetShardUpdatedAtStr возвращает человекочитаемое время обновления +func GetShardUpdatedAtStr(updatedAt int64) string { + if updatedAt == 0 { + return "unknown" + } + return time.UnixMilli(updatedAt).Format("2006-01-02 15:04:05.000") +} + +// GetShardLeaderNode возвращает лидера шарда +func GetShardLeaderNode(leaderNode string) string { + return leaderNode +} + +// GetShardNodeCount возвращает количество узлов в шарде +func GetShardNodeCount(nodes []string) int { + return len(nodes) +} + +// GetShardReplicationFactor возвращает фактор репликации шарда +func GetShardReplicationFactor(nodes []string) int { + return len(nodes) +} + +// ========== ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ ДЛЯ ReplicationLogEntry ========== + +// GetReplicationTimestampStr возвращает человекочитаемое время записи репликации +func GetReplicationTimestampStr(timestamp int64) string { + if timestamp == 0 { + return "unknown" + } + return time.UnixMilli(timestamp).Format("2006-01-02 15:04:05.000") +} + +// IsReplicationSuccess проверяет, успешна ли операция репликации +func IsReplicationSuccess(status string) bool { + return status == "success" +} + +// GetReplicationDuration возвращает длительность операции +func GetReplicationDuration(durationMs int64) time.Duration { + return time.Duration(durationMs) * time.Millisecond +} + +// ========== ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ ДЛЯ ClusterHealth ========== + +// IsClusterHealthy проверяет, здоров ли кластер в целом +func IsClusterHealthy(overallScore float64) bool { + return overallScore >= 80 +} + +// IsClusterDegraded проверяет, деградирован ли кластер +func IsClusterDegraded(overallScore float64) bool { + return overallScore >= 50 && overallScore < 80 +} + +// IsClusterCritical проверяет, находится ли кластер в критическом состоянии +func IsClusterCritical(overallScore float64) bool { + return overallScore < 50 +} + +// GetHealthyNodesCount возвращает количество здоровых узлов +func GetHealthyNodesCount(nodes map[string]*NodeHealth) int { + if nodes == nil { + return 0 + } + count := 0 + for _, node := range 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() +} + +// ========== ТИПЫ, КОТОРЫЕ НЕ ДУБЛИРУЮТСЯ В ДРУГИХ ФАЙЛАХ ========== + +// 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"` +} + +// ClusterHealth представляет информацию о здоровье кластера +type ClusterHealth struct { + Nodes map[string]*NodeHealth `json:"nodes"` + OverallScore float64 `json:"overall_score"` + Recommendations string `json:"recommendations"` + CheckedAt int64 `json:"checked_at"` +} + +// 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"` + Status string `json:"status"` + 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"` +} diff --git a/internal/cluster/worker_pool.go b/internal/cluster/worker_pool.go new file mode 100644 index 0000000..c8d83b7 --- /dev/null +++ b/internal/cluster/worker_pool.go @@ -0,0 +1,395 @@ +/* + * 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 +// Назначение: Lock-free пул воркеров для ограничения количества горутин. +// Использует атомарные операции и lock-free структуры данных. + +package cluster + +import ( + "fmt" + "runtime/debug" + "sync" + "sync/atomic" + "time" + + "futriis/internal/log" +) + +// WorkerPool представляет lock-free пул воркеров для выполнения задач +type WorkerPool struct { + maxWorkers int32 // Максимальное количество воркеров + activeWorkers atomic.Int32 // Текущее количество активных воркеров + tasks *LockFreeQueue // Lock-free очередь задач + stopChan chan struct{} + wg sync.WaitGroup + logger *log.Logger + + // Статистика (атомарные счётчики) + submittedTasks atomic.Uint64 + completedTasks atomic.Uint64 + failedTasks atomic.Uint64 + rejectedTasks atomic.Uint64 + lastSubmitTime atomic.Int64 + lastCompleteTime atomic.Int64 +} + +// Task представляет задачу для выполнения в пуле +type Task struct { + ID string + Execute func() error + CreatedAt int64 + RetryCount int32 +} + +// LockFreeQueue представляет lock-free очередь на основе CAS операций +type LockFreeQueue struct { + head atomic.Value // *queueNode + tail atomic.Value // *queueNode + size atomic.Int64 +} + +type queueNode struct { + value *Task + next atomic.Value // *queueNode +} + +// NewLockFreeQueue создаёт новую lock-free очередь +func NewLockFreeQueue() *LockFreeQueue { + q := &LockFreeQueue{} + dummy := &queueNode{} + q.head.Store(dummy) + q.tail.Store(dummy) + return q +} + +// Enqueue добавляет задачу в очередь (lock-free) +func (q *LockFreeQueue) Enqueue(task *Task) bool { + newNode := &queueNode{value: task} + + for { + tailVal := q.tail.Load() + if tailVal == nil { + continue + } + tail := tailVal.(*queueNode) + + nextVal := tail.next.Load() + var next *queueNode + if nextVal != nil { + next = nextVal.(*queueNode) + } + + if tailVal != q.tail.Load() { + continue + } + + if next != nil { + q.tail.CompareAndSwap(tailVal, next) + continue + } + + if tail.next.CompareAndSwap(nil, newNode) { + q.tail.CompareAndSwap(tailVal, newNode) + q.size.Add(1) + return true + } + } +} + +// Dequeue извлекает задачу из очереди (lock-free) +func (q *LockFreeQueue) Dequeue() *Task { + for { + headVal := q.head.Load() + if headVal == nil { + return nil + } + head := headVal.(*queueNode) + + tailVal := q.tail.Load() + if tailVal == nil { + return nil + } + tail := tailVal.(*queueNode) + + nextVal := head.next.Load() + var next *queueNode + if nextVal != nil { + next = nextVal.(*queueNode) + } + + if headVal != q.head.Load() { + continue + } + + if head == tail { + if next == nil { + return nil + } + q.tail.CompareAndSwap(tailVal, next) + continue + } + + if next == nil { + return nil + } + + task := next.value + if q.head.CompareAndSwap(headVal, next) { + q.size.Add(-1) + return task + } + } +} + +// Size возвращает текущий размер очереди (lock-free) +func (q *LockFreeQueue) Size() int64 { + return q.size.Load() +} + +// NewWorkerPool создаёт новый lock-free пул воркеров +func NewWorkerPool(maxWorkers int, logger *log.Logger) *WorkerPool { + if maxWorkers <= 0 { + maxWorkers = 500 + } + + wp := &WorkerPool{ + maxWorkers: int32(maxWorkers), + tasks: NewLockFreeQueue(), + stopChan: make(chan struct{}), + logger: logger, + } + + // Запускаем диспетчер задач (не блокирующий) + go wp.dispatcher() + + if logger != nil { + logger.Debug(fmt.Sprintf("Lock-free worker pool created: maxWorkers=%d", maxWorkers)) + } + + return wp +} + +// dispatcher управляет воркерами (lock-free) +func (wp *WorkerPool) dispatcher() { + defer func() { + if r := recover(); r != nil { + if wp.logger != nil { + wp.logger.Error(fmt.Sprintf("Worker pool dispatcher panicked: %v\n%s", r, debug.Stack())) + } + // Перезапускаем диспетчер + go wp.dispatcher() + } + }() + + ticker := time.NewTicker(100 * time.Millisecond) + defer ticker.Stop() + + for { + select { + case <-wp.stopChan: + return + case <-ticker.C: + // Динамически регулируем количество воркеров + wp.adjustWorkers() + } + } +} + +// adjustWorkers динамически регулирует количество воркеров (lock-free) +func (wp *WorkerPool) adjustWorkers() { + queueSize := wp.tasks.Size() + activeWorkers := wp.activeWorkers.Load() + + // Если есть задачи и есть место для новых воркеров + if queueSize > 0 && activeWorkers < wp.maxWorkers { + // Проверяем, нужно ли добавить воркер + if activeWorkers == 0 || queueSize > int64(activeWorkers)*2 { + wp.addWorker() + } + } + + // Уменьшаем количество воркеров, если нет задач и воркеры простаивают + if queueSize == 0 && activeWorkers > 0 { + // Не уменьшаем ниже 1 воркера + if activeWorkers > 1 { + // Проверяем, что не было задач в последнее время + lastComplete := wp.lastCompleteTime.Load() + if time.Now().UnixMilli()-lastComplete > 5000 { + wp.removeWorker() + } + } + } +} + +// addWorker добавляет нового воркера (lock-free) +func (wp *WorkerPool) addWorker() { + if wp.activeWorkers.Load() >= wp.maxWorkers { + return + } + + wp.activeWorkers.Add(1) + wp.wg.Add(1) + + go wp.worker() + + if wp.logger != nil { + wp.logger.Debug(fmt.Sprintf("Added worker, active: %d/%d", wp.activeWorkers.Load(), wp.maxWorkers)) + } +} + +// removeWorker удаляет воркера (lock-free) +func (wp *WorkerPool) removeWorker() { + if wp.activeWorkers.Load() <= 1 { + return + } + + // Отправляем специальную задачу на остановку воркера + task := &Task{ + ID: fmt.Sprintf("stop_worker_%d", time.Now().UnixNano()), + Execute: func() error { return nil }, + CreatedAt: time.Now().UnixMilli(), + } + + if wp.tasks.Enqueue(task) { + if wp.logger != nil { + wp.logger.Debug(fmt.Sprintf("Stop signal sent to worker, active: %d/%d", wp.activeWorkers.Load()-1, wp.maxWorkers)) + } + } +} + +// worker выполняет задачи из очереди (lock-free) +func (wp *WorkerPool) worker() { + defer func() { + if r := recover(); r != nil { + if wp.logger != nil { + wp.logger.Error(fmt.Sprintf("Worker panicked: %v\n%s", r, debug.Stack())) + } + wp.activeWorkers.Add(-1) + wp.wg.Done() + // Создаём нового воркера вместо упавшего + if wp.activeWorkers.Load() < wp.maxWorkers { + wp.addWorker() + } + } + }() + + for { + select { + case <-wp.stopChan: + wp.activeWorkers.Add(-1) + wp.wg.Done() + return + default: + task := wp.tasks.Dequeue() + if task == nil { + wp.activeWorkers.Add(-1) + wp.wg.Done() + return + } + + // Проверяем специальную задачу остановки + if len(task.ID) >= 12 && task.ID[:12] == "stop_worker_" { + wp.activeWorkers.Add(-1) + wp.wg.Done() + return + } + + // Выполняем задачу + startTime := time.Now().UnixMilli() + err := task.Execute() + duration := time.Now().UnixMilli() - startTime + + if err != nil { + wp.failedTasks.Add(1) + if wp.logger != nil && duration > 100 { + wp.logger.Warn(fmt.Sprintf("Task %s failed after %dms: %v", task.ID, duration, err)) + } + } else { + wp.completedTasks.Add(1) + wp.lastCompleteTime.Store(time.Now().UnixMilli()) + if wp.logger != nil && duration > 1000 { + wp.logger.Debug(fmt.Sprintf("Task %s completed in %dms", task.ID, duration)) + } + } + } + } +} + +// Submit отправляет задачу в пул (lock-free, неблокирующий) +func (wp *WorkerPool) Submit(task *Task) error { + if task.Execute == nil { + return fmt.Errorf("task execute function is nil") + } + + task.CreatedAt = time.Now().UnixMilli() + + if !wp.tasks.Enqueue(task) { + wp.rejectedTasks.Add(1) + return fmt.Errorf("failed to enqueue task (queue full)") + } + + wp.submittedTasks.Add(1) + wp.lastSubmitTime.Store(task.CreatedAt) + + // Асинхронно добавляем воркера при необходимости + if wp.activeWorkers.Load() == 0 { + go wp.addWorker() + } + + return nil +} + +// SubmitFunc отправляет функцию как задачу (lock-free) +func (wp *WorkerPool) SubmitFunc(id string, fn func() error) error { + return wp.Submit(&Task{ + ID: id, + Execute: fn, + }) +} + +// GetStats возвращает статистику пула (lock-free) +func (wp *WorkerPool) GetStats() map[string]interface{} { + return map[string]interface{}{ + "max_workers": wp.maxWorkers, + "active_workers": wp.activeWorkers.Load(), + "queue_size": wp.tasks.Size(), + "submitted_tasks": wp.submittedTasks.Load(), + "completed_tasks": wp.completedTasks.Load(), + "failed_tasks": wp.failedTasks.Load(), + "rejected_tasks": wp.rejectedTasks.Load(), + "last_submit_time": wp.lastSubmitTime.Load(), + "last_complete_time": wp.lastCompleteTime.Load(), + } +} + +// Stop останавливает пул воркеров +func (wp *WorkerPool) Stop() { + close(wp.stopChan) + + // Ждём завершения всех воркеров с таймаутом + done := make(chan struct{}) + go func() { + wp.wg.Wait() + close(done) + }() + + select { + case <-done: + if wp.logger != nil { + wp.logger.Debug("Worker pool stopped gracefully") + } + case <-time.After(10 * time.Second): + if wp.logger != nil { + wp.logger.Warn("Worker pool stop timeout, forcing shutdown") + } + } +}