# frozen_string_literal: true module Streams # Scale-out / warm spare / scale-in dei nodi overflow (lab o Hetzner). # Kill-switch: STREAM_AUTOSCALE_ENABLED!=1 → no-op. class Autoscaler Result = Struct.new(:actions, :metrics, :skipped, :error, keyword_init: true) LOCK_KEY = "streams:autoscaler:lock" class << self def enabled? ENV["STREAM_AUTOSCALE_ENABLED"] == "1" 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 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?, kind: kind } end def worker_id ENV.fetch("HOSTNAME", "autoscaler") 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 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) @provisioner.decommission!(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) m[:overflow_nodes] < self.class.max_overflow_nodes end def provision_overflow! case self.class.kind when "cloud" then @provisioner.provision_cloud! else @provisioner.provision_lab! end 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