From 021e3aa835d1412aa96230b2191afabcbc696d6a 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: Sat, 18 Jul 2026 20:43:22 +0000 Subject: [PATCH] Upload files to "internal/autoscaling" --- internal/autoscaling/cluster_autoscaler.go | 443 +++++++++++++++++++++ 1 file changed, 443 insertions(+) create mode 100644 internal/autoscaling/cluster_autoscaler.go diff --git a/internal/autoscaling/cluster_autoscaler.go b/internal/autoscaling/cluster_autoscaler.go new file mode 100644 index 0000000..77f9865 --- /dev/null +++ b/internal/autoscaling/cluster_autoscaler.go @@ -0,0 +1,443 @@ +/* + * 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 + */ + +package autoscaling + +import ( + "fmt" + "sync" + "sync/atomic" + "time" + + "futriis/internal/config" + "futriis/internal/log" +) + +// ClusterAutoscaler управляет автоматическим масштабированием кластера +type ClusterAutoscaler struct { + config *config.AutoscalingConfig + logger *log.Logger + mu sync.RWMutex + stopChan chan struct{} + wg sync.WaitGroup + running atomic.Bool + stats *AutoscalingStats + + // Метрики нагрузки + cpuLoad atomic.Float64 + memLoad atomic.Float64 + qps atomic.Uint64 + connections atomic.Uint64 + + // Состояние кластера + currentNodeCount int + targetNodeCount int + + // Коллбэки для реального масштабирования + scaleUpCallback func(count int) error + scaleDownCallback func(count int) error + getNodeCountCallback func() int +} + +// AutoscalingStats статистика автомасштабирования +type AutoscalingStats struct { + TotalScaleUps atomic.Uint64 + TotalScaleDowns atomic.Uint64 + LastScaleUpAt atomic.Int64 + LastScaleDownAt atomic.Int64 + LastEvaluationAt atomic.Int64 + CurrentNodes int + TargetNodes int + mu sync.RWMutex +} + +// NewClusterAutoscaler создаёт новый экземпляр автомасштабирования +func NewClusterAutoscaler(cfg *config.AutoscalingConfig, logger *log.Logger) *ClusterAutoscaler { + if cfg == nil { + return nil + } + + a := &ClusterAutoscaler{ + config: cfg, + logger: logger, + stopChan: make(chan struct{}), + currentNodeCount: cfg.MinNodes, + targetNodeCount: cfg.MinNodes, + stats: &AutoscalingStats{}, + } + + return a +} + +// SetCallbacks устанавливает коллбэки для управления кластером +func (a *ClusterAutoscaler) SetCallbacks( + scaleUp func(count int) error, + scaleDown func(count int) error, + getNodeCount func() int, +) { + a.mu.Lock() + defer a.mu.Unlock() + a.scaleUpCallback = scaleUp + a.scaleDownCallback = scaleDown + a.getNodeCountCallback = getNodeCount +} + +// Start запускает автомасштабирование +func (a *ClusterAutoscaler) Start() { + if !a.config.Enabled { + if a.logger != nil { + a.logger.Info("Autoscaling is disabled in configuration") + } + return + } + + if !a.running.CompareAndSwap(false, true) { + return + } + + a.wg.Add(1) + go a.scalingLoop() + + if a.logger != nil { + a.logger.Info(fmt.Sprintf("ClusterAutoscaler started: min_nodes=%d, max_nodes=%d, threshold_up=%.2f, threshold_down=%.2f", + a.config.MinNodes, a.config.MaxNodes, a.config.ScaleUpThreshold, a.config.ScaleDownThreshold)) + } +} + +// Stop останавливает автомасштабирование +func (a *ClusterAutoscaler) Stop() { + if !a.running.Load() { + return + } + + close(a.stopChan) + a.wg.Wait() + a.running.Store(false) + + if a.logger != nil { + a.logger.Info("ClusterAutoscaler stopped") + } +} + +// ReloadConfig обновляет конфигурацию автомасштабирования +func (a *ClusterAutoscaler) ReloadConfig(cfg *config.AutoscalingConfig) { + a.mu.Lock() + defer a.mu.Unlock() + + oldEnabled := a.config.Enabled + a.config = cfg + + if a.logger != nil { + a.logger.Info(fmt.Sprintf("ClusterAutoscaler configuration reloaded: enabled=%v, min_nodes=%d, max_nodes=%d", + cfg.Enabled, cfg.MinNodes, cfg.MaxNodes)) + } + + if oldEnabled != cfg.Enabled { + if cfg.Enabled { + a.Start() + } else { + a.Stop() + } + } + + if a.currentNodeCount < cfg.MinNodes { + a.currentNodeCount = cfg.MinNodes + a.targetNodeCount = cfg.MinNodes + } + if a.currentNodeCount > cfg.MaxNodes { + a.currentNodeCount = cfg.MaxNodes + a.targetNodeCount = cfg.MaxNodes + } +} + +// UpdateMetrics обновляет метрики нагрузки +func (a *ClusterAutoscaler) UpdateMetrics(cpuLoad, memLoad float64, qps, conns uint64) { + a.cpuLoad.Store(cpuLoad) + a.memLoad.Store(memLoad) + a.qps.Store(qps) + a.connections.Store(conns) +} + +// SetCurrentNodes устанавливает текущее количество узлов +func (a *ClusterAutoscaler) SetCurrentNodes(count int) { + a.mu.Lock() + defer a.mu.Unlock() + + if count < a.config.MinNodes { + count = a.config.MinNodes + } + if count > a.config.MaxNodes { + count = a.config.MaxNodes + } + a.currentNodeCount = count + a.targetNodeCount = count + + a.stats.mu.Lock() + a.stats.CurrentNodes = count + a.stats.TargetNodes = count + a.stats.mu.Unlock() +} + +// GetCurrentNodes возвращает текущее количество узлов +func (a *ClusterAutoscaler) GetCurrentNodes() int { + a.mu.RLock() + defer a.mu.RUnlock() + return a.currentNodeCount +} + +// GetTargetNodes возвращает целевое количество узлов +func (a *ClusterAutoscaler) GetTargetNodes() int { + a.mu.RLock() + defer a.mu.RUnlock() + return a.targetNodeCount +} + +// GetStats возвращает статистику автомасштабирования +func (a *ClusterAutoscaler) GetStats() map[string]interface{} { + a.mu.RLock() + defer a.mu.RUnlock() + + a.stats.mu.RLock() + defer a.stats.mu.RUnlock() + + return map[string]interface{}{ + "enabled": a.config.Enabled, + "min_nodes": a.config.MinNodes, + "max_nodes": a.config.MaxNodes, + "current_nodes": a.currentNodeCount, + "target_nodes": a.targetNodeCount, + "scale_up_threshold": a.config.ScaleUpThreshold, + "scale_down_threshold": a.config.ScaleDownThreshold, + "cpu_load": a.cpuLoad.Load(), + "memory_load": a.memLoad.Load(), + "qps": a.qps.Load(), + "connections": a.connections.Load(), + "total_scale_ups": a.stats.TotalScaleUps.Load(), + "total_scale_downs": a.stats.TotalScaleDowns.Load(), + "last_scale_up_at": a.stats.LastScaleUpAt.Load(), + "last_scale_down_at": a.stats.LastScaleDownAt.Load(), + "last_evaluation_at": a.stats.LastEvaluationAt.Load(), + "predictive_enabled": a.config.PredictiveEnabled, + "max_scale_up_nodes": a.config.MaxScaleUpNodes, + "max_scale_down_nodes": a.config.MaxScaleDownNodes, + "scale_up_cooldown_sec": a.config.ScaleUpCooldownSec, + "scale_down_cooldown_sec": a.config.ScaleDownCooldownSec, + "evaluation_interval_sec": a.config.EvaluationIntervalSec, + } +} + +// scalingLoop основной цикл оценки и масштабирования +func (a *ClusterAutoscaler) scalingLoop() { + defer a.wg.Done() + + ticker := time.NewTicker(a.config.GetEvaluationInterval()) + defer ticker.Stop() + + for { + select { + case <-a.stopChan: + return + case <-ticker.C: + a.evaluateAndScale() + } + } +} + +// evaluateAndScale оценивает нагрузку и выполняет масштабирование +func (a *ClusterAutoscaler) evaluateAndScale() { + a.mu.Lock() + defer a.mu.Unlock() + + if !a.config.Enabled { + return + } + + a.stats.LastEvaluationAt.Store(time.Now().UnixMilli()) + + // Получаем текущие метрики + cpu := a.cpuLoad.Load() + mem := a.memLoad.Load() + qps := a.qps.Load() + conns := a.connections.Load() + + // Обновляем текущее количество узлов из внешнего источника + if a.getNodeCountCallback != nil { + actualCount := a.getNodeCountCallback() + if actualCount > 0 { + a.currentNodeCount = actualCount + } + } + + needScaleUp := false + needScaleDown := false + reason := "" + + // Оценка масштабирования вверх + if cpu > a.config.ScaleUpThreshold || mem > a.config.ScaleUpThreshold { + needScaleUp = true + reason = fmt.Sprintf("high resource usage: cpu=%.2f, mem=%.2f", cpu, mem) + } else if qps > 0 && conns > 0 { + expectedLoad := float64(qps) / float64(10000) + if expectedLoad > a.config.ScaleUpThreshold { + needScaleUp = true + reason = fmt.Sprintf("high load: qps=%d, connections=%d", qps, conns) + } + } + + // Оценка масштабирования вниз + if cpu < a.config.ScaleDownThreshold && mem < a.config.ScaleDownThreshold && + float64(qps) < 100 && float64(conns) < 100 { + needScaleDown = true + reason = "low resource usage" + } + + // Проверяем cooldown периоды + now := time.Now().UnixMilli() + lastUp := a.stats.LastScaleUpAt.Load() + lastDown := a.stats.LastScaleDownAt.Load() + + if needScaleUp { + if now-lastUp < int64(a.config.GetScaleUpCooldown().Seconds())*1000 { + needScaleUp = false + } + } + + if needScaleDown { + if now-lastDown < int64(a.config.GetScaleDownCooldown().Seconds())*1000 { + needScaleDown = false + } + } + + // Выполняем масштабирование + if needScaleUp && a.currentNodeCount < a.config.MaxNodes { + a.scaleUp(reason) + } else if needScaleDown && a.currentNodeCount > a.config.MinNodes { + a.scaleDown(reason) + } +} + +// scaleUp выполняет масштабирование вверх +func (a *ClusterAutoscaler) scaleUp(reason string) { + count := a.config.MaxScaleUpNodes + if a.currentNodeCount+count > a.config.MaxNodes { + count = a.config.MaxNodes - a.currentNodeCount + } + + if count <= 0 { + return + } + + a.targetNodeCount = a.currentNodeCount + count + + if a.logger != nil { + a.logger.Info(fmt.Sprintf("Scaling up: %d -> %d nodes (adding %d) reason: %s", + a.currentNodeCount, a.targetNodeCount, count, reason)) + } + + if a.scaleUpCallback != nil { + if err := a.scaleUpCallback(count); err != nil { + if a.logger != nil { + a.logger.Error(fmt.Sprintf("Scale up failed: %v", err)) + } + return + } + } + + a.currentNodeCount = a.targetNodeCount + a.stats.TotalScaleUps.Add(1) + a.stats.LastScaleUpAt.Store(time.Now().UnixMilli()) + + a.stats.mu.Lock() + a.stats.CurrentNodes = a.currentNodeCount + a.stats.TargetNodes = a.targetNodeCount + a.stats.mu.Unlock() +} + +// scaleDown выполняет масштабирование вниз +func (a *ClusterAutoscaler) scaleDown(reason string) { + count := a.config.MaxScaleDownNodes + if a.currentNodeCount-count < a.config.MinNodes { + count = a.currentNodeCount - a.config.MinNodes + } + + if count <= 0 { + return + } + + a.targetNodeCount = a.currentNodeCount - count + + if a.logger != nil { + a.logger.Info(fmt.Sprintf("Scaling down: %d -> %d nodes (removing %d) reason: %s", + a.currentNodeCount, a.targetNodeCount, count, reason)) + } + + if a.scaleDownCallback != nil { + if err := a.scaleDownCallback(count); err != nil { + if a.logger != nil { + a.logger.Error(fmt.Sprintf("Scale down failed: %v", err)) + } + return + } + } + + a.currentNodeCount = a.targetNodeCount + a.stats.TotalScaleDowns.Add(1) + a.stats.LastScaleDownAt.Store(time.Now().UnixMilli()) + + a.stats.mu.Lock() + a.stats.CurrentNodes = a.currentNodeCount + a.stats.TargetNodes = a.targetNodeCount + a.stats.mu.Unlock() +} + +// GetStatus возвращает текущий статус автомасштабирования +func (a *ClusterAutoscaler) GetStatus() *AutoscalingStatus { + a.mu.RLock() + defer a.mu.RUnlock() + + return &AutoscalingStatus{ + Enabled: a.config.Enabled, + MinNodes: a.config.MinNodes, + MaxNodes: a.config.MaxNodes, + CurrentNodes: a.currentNodeCount, + TargetNodes: a.targetNodeCount, + CPU: a.cpuLoad.Load(), + Memory: a.memLoad.Load(), + QPS: a.qps.Load(), + Connections: a.connections.Load(), + LastEvaluationAt: a.stats.LastEvaluationAt.Load(), + ScaleUpThreshold: a.config.ScaleUpThreshold, + ScaleDownThreshold: a.config.ScaleDownThreshold, + ScaleUpCooldown: a.config.GetScaleUpCooldown(), + ScaleDownCooldown: a.config.GetScaleDownCooldown(), + TotalScaleUps: a.stats.TotalScaleUps.Load(), + TotalScaleDowns: a.stats.TotalScaleDowns.Load(), + } +} + +// AutoscalingStatus статус автомасштабирования +type AutoscalingStatus struct { + Enabled bool `json:"enabled"` + MinNodes int `json:"min_nodes"` + MaxNodes int `json:"max_nodes"` + CurrentNodes int `json:"current_nodes"` + TargetNodes int `json:"target_nodes"` + CPU float64 `json:"cpu_load"` + Memory float64 `json:"memory_load"` + QPS uint64 `json:"qps"` + Connections uint64 `json:"connections"` + LastEvaluationAt int64 `json:"last_evaluation_at"` + ScaleUpThreshold float64 `json:"scale_up_threshold"` + ScaleDownThreshold float64 `json:"scale_down_threshold"` + ScaleUpCooldown time.Duration `json:"scale_up_cooldown"` + ScaleDownCooldown time.Duration `json:"scale_down_cooldown"` + TotalScaleUps uint64 `json:"total_scale_ups"` + TotalScaleDowns uint64 `json:"total_scale_downs"` +}