From 891fcd04c1000107caddac4a0717505310f73000 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: Wed, 16 Sep 2026 21:51:20 +0000 Subject: [PATCH] Update internal/autoscaling/cluster_autoscaler.go --- internal/autoscaling/cluster_autoscaler.go | 803 ++++++++++++--------- 1 file changed, 444 insertions(+), 359 deletions(-) diff --git a/internal/autoscaling/cluster_autoscaler.go b/internal/autoscaling/cluster_autoscaler.go index 77f9865..509381e 100644 --- a/internal/autoscaling/cluster_autoscaler.go +++ b/internal/autoscaling/cluster_autoscaler.go @@ -11,433 +11,518 @@ package autoscaling import ( - "fmt" - "sync" - "sync/atomic" - "time" + "fmt" + "math" + "sync" + "sync/atomic" + "time" - "futriis/internal/config" - "futriis/internal/log" + "futriis/internal/config" + "futriis/internal/log" ) +// ============================================================================= +// СОВМЕСТИМЫЕ АТОМАРНЫЕ ОБЁРТКИ (Go 1.13+) +// ============================================================================= +// В Go 1.19 появились atomic.Bool, atomic.Int32, atomic.Int64, +// atomic.Uint32, atomic.Uint64, atomic.Float64. Для совместимости +// с более старыми версиями Go (в частности, на OpenIndiana, где +// пакет Go может быть устаревшим) реализуем собственные обёртки. + +// atomicBool — совместимая замена atomic.Bool. +type atomicBool struct { + v int32 +} + +func (b *atomicBool) Load() bool { return atomic.LoadInt32(&b.v) != 0 } +func (b *atomicBool) Store(val bool) { + atomic.StoreInt32(&b.v, boolToInt32(val)) +} +func (b *atomicBool) CompareAndSwap(old, new bool) bool { + return atomic.CompareAndSwapInt32(&b.v, boolToInt32(old), boolToInt32(new)) +} + +func boolToInt32(b bool) int32 { + if b { + return 1 + } + return 0 +} + +// atomicInt32 — совместимая замена atomic.Int32. +type atomicInt32 struct { + v int32 +} + +func (a *atomicInt32) Load() int32 { return atomic.LoadInt32(&a.v) } +func (a *atomicInt32) Store(val int32) { atomic.StoreInt32(&a.v, val) } +func (a *atomicInt32) Add(delta int32) int32 { return atomic.AddInt32(&a.v, delta) } +func (a *atomicInt32) CompareAndSwap(old, new int32) bool { + return atomic.CompareAndSwapInt32(&a.v, old, new) +} + +// atomicInt64 — совместимая замена atomic.Int64. +type atomicInt64 struct { + v int64 +} + +func (a *atomicInt64) Load() int64 { return atomic.LoadInt64(&a.v) } +func (a *atomicInt64) Store(val int64) { atomic.StoreInt64(&a.v, val) } +func (a *atomicInt64) Add(delta int64) int64 { return atomic.AddInt64(&a.v, delta) } + +// atomicUint64 — совместимая замена atomic.Uint64. +type atomicUint64 struct { + v uint64 +} + +func (a *atomicUint64) Load() uint64 { return atomic.LoadUint64(&a.v) } +func (a *atomicUint64) Store(val uint64) { atomic.StoreUint64(&a.v, val) } +func (a *atomicUint64) Add(delta uint64) uint64 { return atomic.AddUint64(&a.v, delta) } + +// atomicFloat64 — совместимая замена atomic.Float64. +// Реализована через atomic.Uint64 и math.Float64bits. +type atomicFloat64 struct { + v uint64 +} + +func (f *atomicFloat64) Load() float64 { + return math.Float64frombits(atomic.LoadUint64(&f.v)) +} +func (f *atomicFloat64) Store(val float64) { + atomic.StoreUint64(&f.v, math.Float64bits(val)) +} + +// ============================================================================= +// ClusterAutoscaler +// ============================================================================= + // 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 + config *config.AutoscalingConfig + logger *log.Logger + mu sync.RWMutex + stopChan chan struct{} + stopOnce sync.Once + wg sync.WaitGroup + running atomicBool + stats *AutoscalingStats + + // Метрики нагрузки + cpuLoad atomicFloat64 + memLoad atomicFloat64 + qps atomicUint64 + connections atomicUint64 + + // Состояние кластера + 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 + TotalScaleUps atomicUint64 + TotalScaleDowns atomicUint64 + LastScaleUpAt atomicInt64 + LastScaleDownAt atomicInt64 + LastEvaluationAt atomicInt64 + 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 + 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, + 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 + 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)) - } + 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 останавливает автомасштабирование +// 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") - } + if !a.running.Load() { + return + } + + a.stopOnce.Do(func() { + 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 - } + if cfg == nil { + return + } + + a.mu.Lock() + 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 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 + } + a.mu.Unlock() + + if oldEnabled != cfg.Enabled { + if cfg.Enabled { + a.Start() + } else { + a.Stop() + } + } } // 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) + 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() + 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 + 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 + 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, - } + 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() - } - } + 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) - } + 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 выполняет масштабирование вверх +// scaleUp выполняет масштабирование вверх. +// Вызывается под удержанием a.mu. 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() + 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 выполняет масштабирование вниз +// scaleDown выполняет масштабирование вниз. +// Вызывается под удержанием a.mu. 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() + 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(), - } + 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"` -} + 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"` +} \ No newline at end of file