279 lines
8.8 KiB
Ruby
279 lines
8.8 KiB
Ruby
# 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 quiet_hours?
|
|
QuietHours.active?
|
|
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),
|
|
quiet_hours: quiet_hours?
|
|
}
|
|
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
|
|
@provisioned_this_round = []
|
|
end
|
|
|
|
def reconcile!
|
|
actions = []
|
|
actions.concat(promote_provisioning_nodes!)
|
|
actions.concat(reclaim_stuck_provisioning!)
|
|
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)
|
|
next if @provisioned_this_round.include?(node.id)
|
|
|
|
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 promote_provisioning_nodes!
|
|
actions = []
|
|
StreamNode.where(status: "provisioning").find_each do |node|
|
|
next unless Streams::NodeHealth.promote_if_healthy!(node)
|
|
|
|
actions << :"ready_#{node.slug}"
|
|
Rails.logger.info("[Streams::Autoscaler] promoted #{node.slug} to ready")
|
|
end
|
|
actions
|
|
end
|
|
|
|
def reclaim_stuck_provisioning!
|
|
actions = []
|
|
stuck_after = ENV.fetch("STREAM_NODE_PROVISIONING_STUCK_MINUTES", "15").to_i.minutes.ago
|
|
StreamNode.where(status: "provisioning").where("created_at < ?", stuck_after).find_each do |node|
|
|
@provisioner.decommission!(node)
|
|
actions << :"reclaim_#{node.slug}"
|
|
Rails.logger.warn("[Streams::Autoscaler] decommissioned stuck provisioning #{node.slug}")
|
|
rescue NodeProvisioner::BusyError, NodeProvisioner::Error => e
|
|
Rails.logger.warn("[Streams::Autoscaler] reclaim #{node.slug}: #{e.message}")
|
|
end
|
|
actions
|
|
end
|
|
|
|
def need_capacity?(m)
|
|
m[:free_slots] <= self.class.soft_free_slots
|
|
end
|
|
|
|
def warm_spare_desired?(m)
|
|
return false if self.class.quiet_hours?
|
|
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 self.class.quiet_hours?
|
|
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!
|
|
node =
|
|
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
|
|
@provisioned_this_round << node.id if node
|
|
node
|
|
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)
|
|
# Di notte chiudi subito i nodi idle (niente attesa IDLE_MINUTES).
|
|
return true if self.class.quiet_hours?
|
|
|
|
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 if self.class.quiet_hours?
|
|
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
|