From 6c4b0fa433b8910c1f5c2a41b7704045b6e4a6c6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=93=D1=80=D0=B8=D0=B3=D0=BE=D1=80=D0=B8=D0=B9=20=D0=A1?= =?UTF-8?q?=D0=B0=D1=84=D1=80=D0=BE=D0=BD=D0=BE=D0=B2?= Date: Fri, 12 Jun 2026 16:21:18 +0000 Subject: [PATCH] Upload files to "internal/cluster" --- internal/cluster/backpressure.go | 301 +++++++++++++++++++++++++++++++ 1 file changed, 301 insertions(+) create mode 100644 internal/cluster/backpressure.go diff --git a/internal/cluster/backpressure.go b/internal/cluster/backpressure.go new file mode 100644 index 0000000..c98231b --- /dev/null +++ b/internal/cluster/backpressure.go @@ -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/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 + 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" + } +}