futriix/internal/autoscaling/cluster_autoscaler.go

/*
 * 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"`
}