diff --git a/internal/cluster/backpressure.go b/internal/cluster/backpressure.go deleted file mode 100644 index 7d4d974..0000000 --- a/internal/cluster/backpressure.go +++ /dev/null @@ -1,301 +0,0 @@ -/* - * Copyright 2026 Safronov Grigorii - * - * Licensed under the CDDL, Version 1.0 (the "License"); - * you may not use this file except in compliance with the License. - * - * You may obtain a copy of the License at - * https://opensource.org/licenses/CDDL-1.0 - */ - -// Файл: internal/cluster/backpressure.go -// Назначение: Backpressure при перегрузке системы - -package cluster - -import ( - "fmt" - "sync" - "sync/atomic" - "time" -) - -// BackpressureLevel представляет уровень перегрузки -type BackpressureLevel int - -const ( - LevelNone BackpressureLevel = iota // Нет перегрузки - LevelLow // Низкая перегрузка - задержки - LevelMedium // Средняя перегрузка - отклонение части запросов - LevelHigh // Высокая перегрузка - отклонение большинства - LevelCritical // Критическая - только чтение -) - -// BackpressureManager управляет backpressure -type BackpressureManager struct { - mu sync.RWMutex - currentLevel BackpressureLevel - cpuThreshold float64 - memoryThreshold float64 - queueSizeThreshold int - connectionThreshold int - currentCPU atomic.Uint64 - currentMemory atomic.Uint64 - currentQueueSize atomic.Int64 - currentConnections atomic.Int64 - rejectedCount atomic.Uint64 - delayedCount atomic.Uint64 - lastCheck time.Time - checkInterval time.Duration - logger LoggerInterface // Используем LoggerInterface из node.go - enabled bool - writeAllowed bool - readAllowed bool - rejectProbability atomic.Uint32 - delayDuration atomic.Int64 -} - -// BackpressureConfig содержит настройки backpressure -type BackpressureConfig struct { - Enabled bool `json:"enabled"` - CPUThreshold float64 `json:"cpu_threshold"` - MemoryThreshold float64 `json:"memory_threshold"` - QueueSizeThreshold int `json:"queue_size_threshold"` - ConnectionThreshold int `json:"connection_threshold"` - CheckIntervalMs int `json:"check_interval_ms"` - LowDelayMs int64 `json:"low_delay_ms"` - MediumRejectProb uint32 `json:"medium_reject_prob"` - HighRejectProb uint32 `json:"high_reject_prob"` -} - -// DefaultBackpressureConfig возвращает конфигурацию по умолчанию -func DefaultBackpressureConfig() *BackpressureConfig { - return &BackpressureConfig{ - Enabled: true, - CPUThreshold: 0.8, - MemoryThreshold: 0.85, - QueueSizeThreshold: 10000, - ConnectionThreshold: 5000, - CheckIntervalMs: 1000, - LowDelayMs: 100, - MediumRejectProb: 30, - HighRejectProb: 70, - } -} - -// NewBackpressureManager создаёт новый менеджер backpressure -func NewBackpressureManager(cfg *BackpressureConfig, logger LoggerInterface) *BackpressureManager { - if cfg == nil { - cfg = DefaultBackpressureConfig() - } - - bpm := &BackpressureManager{ - currentLevel: LevelNone, - cpuThreshold: cfg.CPUThreshold, - memoryThreshold: cfg.MemoryThreshold, - queueSizeThreshold: cfg.QueueSizeThreshold, - connectionThreshold: cfg.ConnectionThreshold, - checkInterval: time.Duration(cfg.CheckIntervalMs) * time.Millisecond, - logger: logger, - enabled: cfg.Enabled, - writeAllowed: true, - readAllowed: true, - rejectProbability: atomic.Uint32{}, - delayDuration: atomic.Int64{}, - } - - bpm.rejectProbability.Store(0) - bpm.delayDuration.Store(0) - - if cfg.Enabled { - go bpm.monitorLoop() - } - - if logger != nil { - logger.Debug("Backpressure manager initialized") - } - - return bpm -} - -// monitorLoop периодически проверяет метрики -func (bpm *BackpressureManager) monitorLoop() { - ticker := time.NewTicker(bpm.checkInterval) - defer ticker.Stop() - - for range ticker.C { - bpm.updateLevel() - } -} - -// updateLevel обновляет уровень перегрузки -func (bpm *BackpressureManager) updateLevel() { - cpu := float64(bpm.currentCPU.Load()) / 100.0 - memory := float64(bpm.currentMemory.Load()) / 100.0 - queueSize := bpm.currentQueueSize.Load() - connections := bpm.currentConnections.Load() - - newLevel := LevelNone - - if cpu >= bpm.cpuThreshold || memory >= bpm.memoryThreshold { - newLevel = LevelHigh - } else if queueSize > int64(bpm.queueSizeThreshold) { - if queueSize > int64(bpm.queueSizeThreshold*2) { - newLevel = LevelCritical - } else { - newLevel = LevelMedium - } - } else if connections > int64(bpm.connectionThreshold) { - newLevel = LevelLow - } - - bpm.mu.Lock() - oldLevel := bpm.currentLevel - bpm.currentLevel = newLevel - bpm.mu.Unlock() - - // Применяем политики в зависимости от уровня - bpm.applyPolicies(newLevel) - - if oldLevel != newLevel && bpm.logger != nil { - bpm.logger.Info(fmt.Sprintf("Backpressure level changed from %v to %v (cpu=%.2f%%, mem=%.2f%%, queue=%d, conns=%d)", - bpm.levelToString(oldLevel), bpm.levelToString(newLevel), cpu*100, memory*100, queueSize, connections)) - } -} - -// applyPolicies применяет политики в зависимости от уровня -func (bpm *BackpressureManager) applyPolicies(level BackpressureLevel) { - bpm.mu.Lock() - defer bpm.mu.Unlock() - - switch level { - case LevelNone: - bpm.writeAllowed = true - bpm.readAllowed = true - bpm.rejectProbability.Store(0) - bpm.delayDuration.Store(0) - - case LevelLow: - bpm.writeAllowed = true - bpm.readAllowed = true - bpm.rejectProbability.Store(0) - bpm.delayDuration.Store(100) // 100ms задержка - - case LevelMedium: - bpm.writeAllowed = true - bpm.readAllowed = true - bpm.rejectProbability.Store(30) // 30% отклонение - bpm.delayDuration.Store(200) - - case LevelHigh: - bpm.writeAllowed = false // Запись запрещена - bpm.readAllowed = true - bpm.rejectProbability.Store(70) // 70% отклонение - bpm.delayDuration.Store(500) - - case LevelCritical: - bpm.writeAllowed = false - bpm.readAllowed = true // Только чтение - bpm.rejectProbability.Store(90) - bpm.delayDuration.Store(1000) - } -} - -// BeforeRequest вызывается перед обработкой запроса -func (bpm *BackpressureManager) BeforeRequest(isWrite bool) error { - if !bpm.enabled { - return nil - } - - bpm.mu.RLock() - level := bpm.currentLevel - writeAllowed := bpm.writeAllowed - readAllowed := bpm.readAllowed - rejectProb := bpm.rejectProbability.Load() - delayDur := bpm.delayDuration.Load() - bpm.mu.RUnlock() - - // Проверяем разрешение на операцию - if isWrite && !writeAllowed { - bpm.rejectedCount.Add(1) - return fmt.Errorf("write operations rejected due to backpressure (level: %v)", bpm.levelToString(level)) - } - if !isWrite && !readAllowed { - bpm.rejectedCount.Add(1) - return fmt.Errorf("read operations rejected due to backpressure (level: %v)", bpm.levelToString(level)) - } - - // Вероятностное отклонение - if rejectProb > 0 { - // Простая вероятностная проверка - if uint32(time.Now().UnixNano()%100) < rejectProb { - bpm.rejectedCount.Add(1) - return fmt.Errorf("request rejected due to backpressure (probability: %d%%)", rejectProb) - } - } - - // Добавляем задержку если нужно - if delayDur > 0 { - bpm.delayedCount.Add(1) - time.Sleep(time.Duration(delayDur) * time.Millisecond) - } - - return nil -} - -// AfterRequest вызывается после обработки запроса -func (bpm *BackpressureManager) AfterRequest(duration time.Duration, success bool) { - // Можно использовать для дополнительной статистики -} - -// UpdateMetrics обновляет метрики для backpressure -func (bpm *BackpressureManager) UpdateMetrics(cpuPercent, memoryPercent uint64, queueSize, connections int64) { - bpm.currentCPU.Store(cpuPercent) - bpm.currentMemory.Store(memoryPercent) - bpm.currentQueueSize.Store(queueSize) - bpm.currentConnections.Store(connections) -} - -// GetCurrentLevel возвращает текущий уровень перегрузки -func (bpm *BackpressureManager) GetCurrentLevel() BackpressureLevel { - bpm.mu.RLock() - defer bpm.mu.RUnlock() - return bpm.currentLevel -} - -// GetStats возвращает статистику backpressure -func (bpm *BackpressureManager) GetStats() map[string]interface{} { - bpm.mu.RLock() - defer bpm.mu.RUnlock() - - return map[string]interface{}{ - "current_level": bpm.levelToString(bpm.currentLevel), - "write_allowed": bpm.writeAllowed, - "read_allowed": bpm.readAllowed, - "reject_probability": bpm.rejectProbability.Load(), - "delay_ms": bpm.delayDuration.Load(), - "rejected_count": bpm.rejectedCount.Load(), - "delayed_count": bpm.delayedCount.Load(), - "cpu_threshold": bpm.cpuThreshold, - "memory_threshold": bpm.memoryThreshold, - "queue_threshold": bpm.queueSizeThreshold, - "conn_threshold": bpm.connectionThreshold, - } -} - -func (bpm *BackpressureManager) levelToString(level BackpressureLevel) string { - switch level { - case LevelNone: - return "none" - case LevelLow: - return "low" - case LevelMedium: - return "medium" - case LevelHigh: - return "high" - case LevelCritical: - return "critical" - default: - return "unknown" - } -}