MatchLiveTv/backend/app/services/streams/autoscaler.rb

# frozen_string_literal: true

module Streams
  # Scale-out / warm spare / scale-in dei nodi overflow (lab o Hetzner).
  # Kill-switch: STREAM_AUTOSCALE_ENABLED!=1 OPPURE Redis streams:autoscaler:kill_switch=1.
  class Autoscaler
    Result = Struct.new(:actions, :metrics, :skipped, :error, keyword_init: true)

    LOCK_KEY = "streams:autoscaler:lock"
    KILL_SWITCH_KEY = "streams:autoscaler:kill_switch"

    class << self
      def enabled?
        return false if kill_switch_engaged?
        return false unless ENV["STREAM_AUTOSCALE_ENABLED"] == "1"

        true
      end

      def kill_switch_engaged?
        redis_get(KILL_SWITCH_KEY) == "1"
      end

      def engage_kill_switch!
        redis_set(KILL_SWITCH_KEY, "1")
      end

      def clear_kill_switch!
        redis_del(KILL_SWITCH_KEY)
      end

      def soft_free_slots
        ENV.fetch("STREAM_AUTOSCALE_SOFT_FREE_SLOTS", "2").to_i
      end

      def warm_spare_min
        ENV.fetch("STREAM_AUTOSCALE_WARM_SPARE", "1").to_i
      end

      def idle_minutes
        ENV.fetch("STREAM_AUTOSCALE_IDLE_MINUTES", "30").to_i
      end

      def max_overflow_nodes
        ENV.fetch("STREAM_AUTOSCALE_MAX_NODES", "5").to_i
      end

      def kind
        ENV.fetch("STREAM_AUTOSCALE_KIND", "lab") # lab|cloud
      end

      def allow_cloud?
        ENV["STREAM_AUTOSCALE_ALLOW_CLOUD"] == "1" && ENV["HCLOUD_TOKEN"].present?
      end

      def node_eur_per_hour
        ENV.fetch("STREAM_AUTOSCALE_NODE_EUR_PER_HOUR", "0.015").to_f
      end

      def monthly_budget_eur
        ENV.fetch("STREAM_AUTOSCALE_MONTHLY_BUDGET_EUR", "40").to_f
      end

      def estimated_monthly_eur(overflow_count = nil)
        count = overflow_count || metrics[:overflow_nodes]
        # Worst case: nodi sempre accesi 24/7
        (count * node_eur_per_hour * 24 * 30).round(2)
      end

      def within_budget?(overflow_count = nil)
        estimated_monthly_eur(overflow_count) <= monthly_budget_eur
      end

      def reconcile!(provisioner: nil)
        return Result.new(skipped: true, actions: [], metrics: metrics) unless enabled?

        redis = Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0"))
        unless redis.set(LOCK_KEY, worker_id, nx: true, ex: 55)
          return Result.new(skipped: true, actions: [], metrics: metrics, error: "locked")
        end

        begin
          new(provisioner: provisioner).reconcile!
        ensure
          redis.del(LOCK_KEY)
        end
      end

      def metrics
        NodeRegistry.ensure_home_from_env!
        nodes = StreamNode.ready.to_a
        overflow = StreamNode.where.not(slug: NodeRegistry::HOME_SLUG)
          .where(status: %w[ready draining provisioning]).to_a
        {
          free_slots: nodes.sum(&:free_slots),
          ready_nodes: nodes.size,
          spare_ready: nodes.count { |n| n.slug != NodeRegistry::HOME_SLUG && n.active_publishers.zero? },
          overflow_nodes: overflow.size,
          soft_free_slots: soft_free_slots,
          warm_spare_min: warm_spare_min,
          max_overflow_nodes: max_overflow_nodes,
          enabled: enabled?,
          env_enabled: ENV["STREAM_AUTOSCALE_ENABLED"] == "1",
          kill_switch: kill_switch_engaged?,
          kind: kind,
          allow_cloud: allow_cloud?,
          estimated_monthly_eur: estimated_monthly_eur(overflow.size),
          monthly_budget_eur: monthly_budget_eur,
          within_budget: within_budget?(overflow.size)
        }
      end

      def worker_id
        ENV.fetch("HOSTNAME", "autoscaler")
      end

      def redis_get(key)
        Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0")).get(key)
      rescue Redis::BaseError
        nil
      end

      def redis_set(key, value)
        Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0")).set(key, value)
      end

      def redis_del(key)
        Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0")).del(key)
      end
    end

    def initialize(provisioner: nil)
      @provisioner = provisioner || NodeProvisioner.new
    end

    def reconcile!
      actions = []
      m = self.class.metrics

      if need_capacity?(m) && can_provision?(m)
        provision_overflow!
        actions << :scale_out
        m = self.class.metrics
      elsif need_capacity?(m) && !can_provision?(m)
        actions << :blocked_capacity
        Rails.logger.warn("[Streams::Autoscaler] capacity needed but blocked metrics=#{m.inspect}")
      end

      if warm_spare_desired?(m) && m[:spare_ready] < self.class.warm_spare_min && can_provision?(m)
        provision_overflow!
        actions << :warm_spare
        m = self.class.metrics
      end

      scale_in_candidates.each do |node|
        next if keep_as_warm_spare?(node)

        safe_scale_in!(node)
        actions << :"scale_in_#{node.slug}"
        m = self.class.metrics
      rescue NodeProvisioner::BusyError, NodeProvisioner::Error => e
        Rails.logger.warn("[Streams::Autoscaler] scale-in #{node.slug}: #{e.message}")
      end

      Rails.logger.info("[Streams::Autoscaler] actions=#{actions.inspect} metrics=#{m.inspect}")
      Result.new(actions: actions, metrics: m, skipped: false)
    end

    private

    def need_capacity?(m)
      m[:free_slots] <= self.class.soft_free_slots
    end

    def warm_spare_desired?(m)
      return false if self.class.warm_spare_min <= 0

      need_capacity?(m) || overflow_in_use?
    end

    def overflow_in_use?
      StreamNode.ready.where.not(slug: NodeRegistry::HOME_SLUG).any? { |n| n.active_publishers.positive? }
    end

    def can_provision?(m)
      return false if m[:overflow_nodes] >= self.class.max_overflow_nodes
      return false unless self.class.within_budget?(m[:overflow_nodes] + 1)
      return false if self.class.kind == "cloud" && !self.class.allow_cloud?

      true
    end

    def provision_overflow!
      case self.class.kind
      when "cloud"
        raise NodeProvisioner::Error, "Cloud autoscale disabilitato (STREAM_AUTOSCALE_ALLOW_CLOUD / HCLOUD_TOKEN)" unless self.class.allow_cloud?

        @provisioner.provision_cloud!
      else
        @provisioner.provision_lab!
      end
    end

    def safe_scale_in!(node)
      @provisioner.drain!(node) unless node.status == "draining"
      node.reload
      raise NodeProvisioner::BusyError, "sessioni ancora attive" if node.occupying_sessions.exists?
      raise NodeProvisioner::Error, "idle insufficiente" unless idle_long_enough?(node)

      @provisioner.decommission!(node)
    end

    def scale_in_candidates
      StreamNode.where.not(slug: NodeRegistry::HOME_SLUG)
        .where(status: %w[ready draining])
        .order(:created_at)
        .select { |n| n.active_publishers.zero? && idle_long_enough?(n) }
    end

    def idle_long_enough?(node)
      idle_since(node) <= self.class.idle_minutes.minutes.ago
    end

    def idle_since(node)
      last_end = node.stream_sessions.where(status: %w[ended error]).maximum(:ended_at)
      last_end || node.created_at
    end

    def keep_as_warm_spare?(node)
      return false unless warm_spare_desired?(self.class.metrics)

      spares = StreamNode.ready.where.not(slug: NodeRegistry::HOME_SLUG).select { |n| n.active_publishers.zero? }
      spares.size <= self.class.warm_spare_min && spares.map(&:id).include?(node.id)
    end
  end
end